Merge pull request #1481 from wger-project/feature/deletion-log-repla…

…ce-by

Allow to set a replacement in the deletion log

wger/exercises/api/serializers.py

@@ -78,6 +78,7 @@ class Meta:
         fields = [
             'model_type',
             'uuid',
+            'replaced_by',
             'timestamp',
             'comment',
         ]

wger/exercises/api/views.py

@@ -17,6 +17,7 @@
 
 # Standard Library
 import logging
+from uuid import UUID
 
 # Django
 from django.conf import settings
@@ -133,6 +134,17 @@ def perform_update(self, serializer):
             action_object=serializer.instance,
         )
 
+    def perform_destroy(self, instance: ExerciseBase):
+        """Manually delete the exercise and set the replacement, if any"""
+
+        uuid = self.request.query_params.get('replaced_by', '')
+        try:
+            UUID(uuid, version=4)
+        except ValueError:
+            uuid = None
+
+        instance.delete(replace_by=uuid)
+
 
 class ExerciseTranslationViewSet(ModelViewSet):
     """

wger/exercises/fixtures/test-exercises.json

@@ -140,7 +140,7 @@
     "pk": 1,
     "model": "exercises.exercise",
     "fields": {
-      "uuid": "9838235ce38f4ca6921e9d237d8e0813",
+      "uuid": "9838235c-e38f-4ca6-921e-9d237d8e0813",
       "language": 2,
       "exercise_base": 1,
       "description": "Lorem ipsum dolor sit amet",
@@ -166,7 +166,7 @@
     "pk": 3,
     "model": "exercises.exercise",
     "fields": {
-      "uuid": "946afe7b54a644a69c36c3e31e6b4c3b",
+      "uuid": "946afe7b-54a6-44a6-9c36-c3e31e6b4c3b",
       "language": 2,
       "exercise_base": 3,
       "description": "",

wger/exercises/management/commands/sync-exercises.py

@@ -23,7 +23,7 @@
 
 # wger
 from wger.exercises.sync import (
-    delete_entries,
+    handle_deleted_entries,
     sync_categories,
     sync_equipment,
     sync_exercises,
@@ -88,4 +88,4 @@ def handle(self, **options):
         sync_licenses(self.stdout.write, self.remote_url, self.style.SUCCESS)
         sync_exercises(self.stdout.write, self.remote_url, self.style.SUCCESS)
         if not options['skip_delete']:
-            delete_entries(self.stdout.write, self.remote_url, self.style.SUCCESS)
+            handle_deleted_entries(self.stdout.write, self.remote_url, self.style.SUCCESS)

wger/exercises/migrations/0026_deletionlog_replaced_by.py

@@ -0,0 +1,23 @@
+# Generated by Django 4.1.9 on 2023-10-19 11:09
+
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+    dependencies = [
+        ('exercises', '0025_rename_update_date_exercise_last_update_and_more'),
+    ]
+
+    operations = [
+        migrations.AddField(
+            model_name='deletionlog',
+            name='replaced_by',
+            field=models.UUIDField(
+                default=None,
+                editable=False,
+                help_text='UUID of the object ',
+                null=True,
+                verbose_name='Replaced by'
+            ),
+        ),
+    ]

wger/exercises/models/base.py

@@ -234,3 +234,26 @@ def get_exercise(self, language: Optional[str] = None):
             exercise = self.exercises.filter(language__short_name=language).first()
 
         return exercise
+
+    def delete(self, using=None, keep_parents=False, replace_by: str = None):
+        """
+        Save entry to log
+        """
+        # wger
+        from wger.exercises.models import DeletionLog
+
+        if replace_by:
+            try:
+                ExerciseBase.objects.get(uuid=replace_by)
+            except ExerciseBase.DoesNotExist:
+                replace_by = None
+
+        log = DeletionLog(
+            model_type=DeletionLog.MODEL_BASE,
+            uuid=self.uuid,
+            comment=f"Exercise base of {self.get_exercise(ENGLISH_SHORT_NAME).name}",
+            replaced_by=replace_by,
+        )
+        log.save()
+
+        return super().delete(using, keep_parents)

wger/exercises/models/deletion_log.py

@@ -54,6 +54,16 @@ class DeletionLog(models.Model):
         verbose_name='UUID',
     )
 
+    replaced_by = models.UUIDField(
+        default=None,
+        unique=False,
+        editable=False,
+        null=True,
+        verbose_name='Replaced by',
+        help_text='UUID of the object replaced by the deleted one. At the moment only available '
+                  'for exercise bases',
+    )
+
     timestamp = models.DateTimeField(auto_now=True)
 
     comment = models.CharField(max_length=200, default='')

wger/exercises/signals.py

@@ -34,7 +34,6 @@
 from wger.exercises.models import (
     DeletionLog,
     Exercise,
-    ExerciseBase,
     ExerciseImage,
     ExerciseVideo,
 )
@@ -108,19 +107,18 @@ def delete_exercise_video_on_update(sender, instance: ExerciseVideo, **kwargs):
             path.unlink()
 
 
-@receiver(pre_delete, sender=ExerciseBase)
-def add_deletion_log_base(sender, instance: ExerciseBase, **kwargs):
-    log = DeletionLog(
-        model_type=DeletionLog.MODEL_BASE,
-        uuid=instance.uuid,
-    )
-    log.save()
+# Deletion log for exercise bases is handled in the model
+# @receiver(pre_delete, sender=ExerciseBase)
+# def add_deletion_log_base(sender, instance: ExerciseBase, **kwargs):
+#     pass
 
 
 @receiver(pre_delete, sender=Exercise)
 def add_deletion_log_translation(sender, instance: Exercise, **kwargs):
     log = DeletionLog(
-        model_type=DeletionLog.MODEL_TRANSLATION, uuid=instance.uuid, comment=instance.name
+        model_type=DeletionLog.MODEL_TRANSLATION,
+        uuid=instance.uuid,
+        comment=instance.name,
     )
     log.save()
 

wger/exercises/sync.py

@@ -53,6 +53,10 @@
     ExerciseVideo,
     Muscle,
 )
+from wger.manager.models import (
+    Setting,
+    WorkoutLog,
+)
 from wger.utils.requests import (
     get_paginated,
     wger_headers,
@@ -75,6 +79,7 @@ def sync_exercises(
     for data in result:
 
         uuid = data['uuid']
+        created = data['created']
         license_id = data['license']['id']
         category_id = data['category']['id']
         license_author = data['license_author']
@@ -84,7 +89,10 @@ def sync_exercises(
 
         base, base_created = ExerciseBase.objects.update_or_create(
             uuid=uuid,
-            defaults={'category_id': category_id},
+            defaults={
+                'category_id': category_id,
+                'created': created
+            },
         )
         print_fn(f"{'created' if base_created else 'updated'} exercise {uuid}")
 
@@ -272,27 +280,56 @@ def sync_equipment(
     print_fn(style_fn('done!\n'))
 
 
-def delete_entries(
-    print_fn,
+def handle_deleted_entries(
+    print_fn=None,
     remote_url=settings.WGER_SETTINGS['WGER_INSTANCE'],
     style_fn=lambda x: x,
 ):
+    if not print_fn:
+        def print_fn(_):
+            return None
+
     """Delete exercises that were removed on the server"""
-    print_fn('*** Deleting exercises data that was removed on the server...')
+    print_fn('*** Deleting exercise data that was removed on the server...')
 
     headers = wger_headers()
     url = make_uri(DELETION_LOG_ENDPOINT, server_url=remote_url, query={'limit': 100})
     result = get_paginated(url, headers=headers)
 
     for data in result:
         uuid = data['uuid']
+        replaced_by_uuid = data['replaced_by']
         model_type = data['model_type']
 
         if model_type == DeletionLog.MODEL_BASE:
+            obj_replaced = None
+            nr_settings = None
+            nr_logs = None
+            try:
+                obj_replaced = ExerciseBase.objects.get(uuid=replaced_by_uuid)
+            except ExerciseBase.DoesNotExist:
+                pass
+
             try:
                 obj = ExerciseBase.objects.get(uuid=uuid)
+
+                # Replace exercise in workouts and logs
+                if obj_replaced:
+                    nr_settings = (
+                        Setting.objects.filter(exercise_base=obj
+                                               ).update(exercise_base=obj_replaced)
+                    )
+                    nr_logs = (
+                        WorkoutLog.objects.filter(exercise_base=obj
+                                                  ).update(exercise_base=obj_replaced)
+                    )
+
                 obj.delete()
                 print_fn(f'Deleted exercise base {uuid}')
+                if nr_settings:
+                    print_fn(f'- replaced {nr_settings} time(s) in workouts by {replaced_by_uuid}')
+                if nr_logs:
+                    print_fn(f'- replaced {nr_logs} time(s) in workout logs by {replaced_by_uuid}')
             except ExerciseBase.DoesNotExist:
                 pass
 

wger/exercises/tasks.py

@@ -25,9 +25,9 @@
 # wger
 from wger.celery_configuration import app
 from wger.exercises.sync import (
-    delete_entries,
     download_exercise_images,
     download_exercise_videos,
+    handle_deleted_entries,
     sync_categories,
     sync_equipment,
     sync_exercises,
@@ -51,7 +51,7 @@ def sync_exercises_task():
     sync_muscles(logger.info)
     sync_equipment(logger.info)
     sync_exercises(logger.info)
-    delete_entries(logger.info)
+    handle_deleted_entries(logger.info)
 
 
 @app.task

wger/exercises/tests/test_deletion_log.py

@@ -12,6 +12,9 @@
 #
 # You should have received a copy of the GNU Affero General Public License
 
+# Standard Library
+from uuid import UUID
+
 # wger
 from wger.core.tests.base_testcase import WgerTestCase
 from wger.exercises.models import (
@@ -36,14 +39,68 @@ def test_base(self):
         base.delete()
 
         # Base is deleted
-        count_base = DeletionLog.objects.filter(model_type=DeletionLog.MODEL_BASE,
-                                                uuid=base.uuid).count()
-        self.assertEqual(count_base, 1)
+        count_base_logs = DeletionLog.objects.filter(
+            model_type=DeletionLog.MODEL_BASE, uuid=base.uuid
+        ).count()
+        log = DeletionLog.objects.get(pk=1)
+
+        self.assertEqual(count_base_logs, 1)
+        self.assertEqual(log.model_type, 'base')
+        self.assertEqual(log.uuid, base.uuid)
+        self.assertEqual(log.comment, 'Exercise base of An exercise')
+        self.assertEqual(log.replaced_by, None)
 
         # All translations are also deleted
         count = DeletionLog.objects.filter(model_type=DeletionLog.MODEL_TRANSLATION).count()
         self.assertEqual(count, 2)
 
+        # First translation
+        log2 = DeletionLog.objects.get(pk=4)
+        self.assertEqual(log2.model_type, 'translation')
+        self.assertEqual(log2.uuid, UUID('9838235c-e38f-4ca6-921e-9d237d8e0813'))
+        self.assertEqual(log2.comment, 'An exercise')
+        self.assertEqual(log2.replaced_by, None)
+
+        # Second translation
+        log3 = DeletionLog.objects.get(pk=5)
+        self.assertEqual(log3.model_type, 'translation')
+        self.assertEqual(log3.uuid, UUID('13b532f9-d208-462e-a000-7b9982b2b53e'))
+        self.assertEqual(log3.comment, 'Test exercise 123')
+        self.assertEqual(log3.replaced_by, None)
+
+    def test_base_with_replaced_by(self):
+        """
+        Test that an entry is generated when a base is deleted and the replaced by is
+        set correctly
+        """
+        self.assertEqual(DeletionLog.objects.all().count(), 0)
+
+        base = ExerciseBase.objects.get(pk=1)
+        base.delete(replace_by="ae3328ba-9a35-4731-bc23-5da50720c5aa")
+
+        # Base is deleted
+        log = DeletionLog.objects.get(pk=1)
+
+        self.assertEqual(log.model_type, 'base')
+        self.assertEqual(log.uuid, base.uuid)
+        self.assertEqual(log.replaced_by, UUID('ae3328ba-9a35-4731-bc23-5da50720c5aa'))
+
+    def test_base_with_nonexistent_replaced_by(self):
+        """
+        Test that an entry is generated when a base is deleted and the replaced by is
+        set correctly. If the UUID is not found in the DB, it's set to None
+        """
+        self.assertEqual(DeletionLog.objects.all().count(), 0)
+
+        base = ExerciseBase.objects.get(pk=1)
+        base.delete(replace_by="12345678-1234-1234-1234-1234567890ab")
+
+        # Base is deleted
+        log = DeletionLog.objects.get(pk=1)
+
+        self.assertEqual(log.model_type, 'base')
+        self.assertEqual(log.replaced_by, None)
+
     def test_translation(self):
         """
         Test that an entry is generated when a translation is deleted