data migration for expiration_date

Author: aaxelb

project/settings.py

@@ -446,6 +446,8 @@ def route_urgent_task(name, args, kwargs, options, task=None, **kw):
 SITE_ID = 1
 PUBLIC_SENTRY_DSN = os.environ.get('PUBLIC_SENTRY_DSN')
 
+DATA_MIGRATION_CHUNK_SIZE = int(os.environ.get('DATA_MIGRATION_CHUNK_SIZE', 666))
+
 SHARE_WEB_URL = os.environ.get('SHARE_WEB_URL', 'http://localhost:8003').rstrip('/') + '/'
 SHARE_USER_AGENT = os.environ.get('SHARE_USER_AGENT', 'SHAREbot/{} (+{})'.format(VERSION, SHARE_WEB_URL))
 SHARE_ADMIN_USERNAME = os.environ.get('SHARE_ADMIN_USERNAME', 'admin')

share/migrations/0078_delete_rawdatum.py

@@ -7,7 +7,7 @@ class Migration(migrations.Migration):
 
     dependencies = [
         ('share', '0077_big_cleanup_2025'),
-        ('trove', '0008_no_raw_datum'),
+        ('trove', '0011_no_raw_datum'),
     ]
 
     operations = [

trove/migrations/0008_no_raw_datum.py

@@ -5,7 +5,7 @@
 class Migration(migrations.Migration):
 
     dependencies = [
-        ('trove', '0008_no_raw_datum'),
+        ('trove', '0007_rawdata_fks_do_nothing'),
     ]
 
     operations = [
@@ -41,4 +41,19 @@ class Migration(migrations.Migration):
             name='supplementary_suid',
             field=models.ForeignKey(on_delete=django.db.models.deletion.CASCADE, related_name='supplementary_description_set', to='share.sourceuniqueidentifier'),
         ),
+        migrations.AddField(
+            model_name='archivedresourcedescription',
+            name='expiration_date',
+            field=models.DateField(blank=True, help_text='An (optional) date when this description will no longer be valid.', null=True),
+        ),
+        migrations.AddField(
+            model_name='latestresourcedescription',
+            name='expiration_date',
+            field=models.DateField(blank=True, help_text='An (optional) date when this description will no longer be valid.', null=True),
+        ),
+        migrations.AddField(
+            model_name='supplementaryresourcedescription',
+            name='expiration_date',
+            field=models.DateField(blank=True, help_text='An (optional) date when this description will no longer be valid.', null=True),
+        ),
     ]

…ions/0009_resource_description_rename.py …/migrations/0008_resource_description.pytrove/migrations/0009_resource_description_rename.py renamed to trove/migrations/0008_resource_description.py trove/migrations/0009_resource_description_rename.py renamed to trove/migrations/0008_resource_description.py

@@ -6,7 +6,7 @@ class Migration(migrations.Migration):
     atomic = False  # allow adding indexes concurrently (without locking tables)
 
     dependencies = [
-        ('trove', '0009_resource_description_rename'),
+        ('trove', '0008_resource_description'),
     ]
 
     operations = [

…ons/0010_description_expiration_index.py …ons/0009_description_expiration_index.pytrove/migrations/0010_description_expiration_index.py renamed to trove/migrations/0009_description_expiration_index.py trove/migrations/0010_description_expiration_index.py renamed to trove/migrations/0009_description_expiration_index.py

@@ -0,0 +1,46 @@
+import logging
+import time
+
+from django.conf import settings
+from django.db import migrations
+from django.db.models import F
+
+from trove.util.django import pk_chunked_queryset
+
+_logger = logging.getLogger(__name__)
+
+
+def migrate_expiration_date(apps, schema_editor):
+    SupplementaryResourceDescription = apps.get_model('trove', 'SupplementaryResourceDescription')
+    # on share.osf.io, only supplements have expirations
+    _supp_qs = SupplementaryResourceDescription.objects.filter(
+        expiration_date__isnull=True,
+        from_raw_datum__expiration_date__isnull=False,
+    )
+    _total_updated = 0
+    _before_all = time.perf_counter()
+    for _chunk_qs in pk_chunked_queryset(_supp_qs, settings.DATA_MIGRATION_CHUNK_SIZE):
+        _before_chunk = time.perf_counter()
+        _updated_count = _chunk_qs.update(
+            expiration_date=F('from_raw_datum__expiration_date'),
+        )
+        _chunk_seconds = time.perf_counter() - _before_chunk
+        _logger.info('migrated %s in %ss...', _updated_count, _chunk_seconds)
+        _total_updated += _updated_count
+    _total_seconds = time.perf_counter() - _before_all
+    _logger.info('done! migrated %s in %ss', _total_updated, _total_seconds)
+
+
+class Migration(migrations.Migration):
+    atomic = False  # potentially lorg data migration; allow partial progress
+
+    dependencies = [
+        ('trove', '0009_description_expiration_index'),
+    ]
+
+    operations = [
+        migrations.RunPython(
+            code=migrate_expiration_date,
+            reverse_code=migrations.RunPython.noop,
+        ),
+    ]

trove/migrations/0010_data__copy_expiration_date.py

@@ -0,0 +1,27 @@
+from django.db import migrations
+
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        ('trove', '0010_data__copy_expiration_date'),
+    ]
+
+    operations = [
+        migrations.RemoveConstraint(
+            model_name='archivedresourcedescription',
+            name='trove_archivedindexcardrdf_uniq_archived_version',
+        ),
+        migrations.RemoveField(
+            model_name='archivedresourcedescription',
+            name='from_raw_datum',
+        ),
+        migrations.RemoveField(
+            model_name='latestresourcedescription',
+            name='from_raw_datum',
+        ),
+        migrations.RemoveField(
+            model_name='supplementaryresourcedescription',
+            name='from_raw_datum',
+        ),
+    ]

trove/migrations/0011_no_raw_datum.py

@@ -0,0 +1,33 @@
+from __future__ import annotations
+from collections.abc import Iterator
+
+from django.db.models.query import QuerySet
+
+
+__all__ = ('pk_chunked_queryset',)
+
+
+def pk_chunked_queryset(queryset, chunksize: int) -> Iterator[QuerySet]:
+    '''pk_chunked_queryset: for chunked update/delete on django querysets
+
+    since `.update()` or `.delete()` cannot be used on sliced querysets,
+    this orders the given queryset by pk and yields a filtered (not sliced)
+    queryset for each chunk
+    '''
+    _ordered_qs = queryset.order_by('pk')
+    _prior_end_pk = None
+    while True:  # for each chunk:
+        _qs = (
+            _ordered_qs
+            if _prior_end_pk is None
+            else _ordered_qs.filter(pk__gt=_prior_end_pk)
+        )
+        # query for the last pk in the chunk
+        _pk_qs = _qs.values_list('pk', flat=True)
+        _end_pk = _pk_qs[chunksize - 1: chunksize].first() or _pk_qs.last()
+        if _end_pk is None:
+            break  # end
+        if (_prior_end_pk is not None) and (_end_pk <= _prior_end_pk):
+            raise RuntimeError(f'sentinel pks not ascending?? got {_end_pk} after {_prior_end_pk}')
+        _prior_end_pk = _end_pk
+        yield _qs.filter(pk__lte=_end_pk)