Add and use delete_all_media_for_source task

tubesync/sync/signals.py

@@ -146,9 +146,22 @@ def source_pre_delete(sender, instance, **kwargs):
     instance.deactivate()
     log.info(f'Deleting tasks for source: {instance.name}')
     delete_task_by_source('sync.tasks.index_source_task', instance.pk)
-    for media in Media.objects.filter(source=instance):
+    # Schedule deletion of media
-        log.info(f'Deleting media for source: {instance.name} item: {media.name}')
+    verbose_name = _('Deleting all media for source "{}"')
-        media.delete()
+    delete_all_media_for_source(
+        str(instance.pk),
+        str(instance.name),
+        priority=1,
+        verbose_name=verbose_name.format(instance.name),
+    )
+    # Try to do it all immediately
+    # If this is killed, the scheduled task should do the work instead.
+    delete_all_media_for_source.now(
+        str(instance.pk),
+        str(instance.name),
+        priority=0,
+        verbose_name=verbose_name.format(instance.name),
+    )
 
 
 @receiver(post_delete, sender=Source)

tubesync/sync/tasks.py

@@ -690,3 +690,24 @@ def wait_for_media_premiere(media_id):
         media.title = _(f'Premieres in {hours(media.published - now)} hours')
         media.save()
 
+@background(schedule=300, remove_existing_tasks=False)
+@atomic(durable=True)
+def delete_all_media_for_source(source_id, source_name):
+    source = None
+    try:
+        source = Source.objects.get(pk=source_id)
+    except Source.DoesNotExist:
+        # Task triggered but the source no longer exists, do nothing
+        log.error(f'Task delete_all_media_for_source(pk={source_id}) called but no '
+                  f'source exists with ID: {source_id}')
+        pass
+    mqs = Media.objects.all().defer(
+        'metadata',
+    ).filter(
+        source=source or source_id,
+    )
+    for media in mqs:
+        log.info(f'Deleting media for source: {source_name} item: {media.name}')
+        with atomic():
+            media.delete()
+