Merge pull request #562 from PetrDlouhy/fix_thumbnail_cleanup

fix #516: thumbnail_cleanup command for S3 and different source storages

.github/workflows/python.yml

@@ -13,7 +13,7 @@ jobs:
     strategy:
       max-parallel: 4
       matrix:
-        python-version: ['3.7', '3.8', '3.9', '3.10', '3.11', '3.12']
+        python-version: ['3.8', '3.9', '3.10', '3.11', '3.12']
 
     steps:
     - uses: actions/checkout@v2

easy_thumbnails/files.py

@@ -422,6 +422,10 @@ def get_thumbnail_name(self, thumbnail_options, transparent=False):
         """
         thumbnail_options = self.get_options(thumbnail_options)
         path, source_filename = os.path.split(self.name)
+        # remove storage location
+        path = path.replace(self.source_storage.location, '')
+        # remove leading slash if present
+        path = path.lstrip('/')
         source_extension = os.path.splitext(source_filename)[1][1:].lower()
         preserve_extensions = self.thumbnail_preserve_extensions
         if preserve_extensions is True or isinstance(preserve_extensions, (list, tuple)) and \

easy_thumbnails/management/commands/thumbnail_cleanup.py

@@ -9,6 +9,8 @@
 from easy_thumbnails.conf import settings
 from easy_thumbnails.models import Source
 from easy_thumbnails.storage import get_storage
+from easy_thumbnails.utils import get_storage_hash
+from django.core.files.storage import storages
 
 
 class ThumbnailCollectionCleaner:
@@ -25,8 +27,11 @@ def __init__(self, stdout, stderr):
         self.stdout = stdout
         self.stderr = stderr
 
-    def _get_absolute_path(self, path):
-        return os.path.join(settings.MEDIA_ROOT, path)
+    def _get_absolute_path(self, path, storage):
+        if hasattr(storage, 'location'):
+            return os.path.join(storage.location, path)
+        else:
+            return os.path.join(settings.MEDIA_ROOT, path)
 
     def _get_relative_path(self, path):
         return os.path.relpath(path, settings.MEDIA_ROOT)
@@ -54,6 +59,7 @@ def clean_up(self, dry_run=False, verbosity=1, last_n_days=0,
         if not storage:
             storage = get_storage()
 
+        storage_hash_map = {get_storage_hash(storages[alias]): alias for alias in settings.STORAGES.keys()}
         sources_to_delete = []
         time_start = time.time()
 
@@ -66,18 +72,25 @@ def clean_up(self, dry_run=False, verbosity=1, last_n_days=0,
             query = query.filter(name__startswith=cleanup_path)
 
         for source in queryset_iterator(query):
+            source_storage_alias = storage_hash_map.get(source.storage_hash)
+            source_storage = storages[source_storage_alias] if source_storage_alias else None
+            if not source_storage:
+                self.stdout.write(f"Source storage hash ({source.storage_hash}) not found in STORAGES")
+                self.stdout.write("Can't determine source storage, skipping source")
+                continue
+
             self.sources += 1
-            abs_source_path = self._get_absolute_path(source.name)
+            abs_source_path = self._get_absolute_path(source.name, source_storage)
 
-            if not self._check_if_exists(storage, abs_source_path):
+            if not self._check_if_exists(source_storage, abs_source_path):
                 if verbosity > 0:
                     self.stdout.write("Source not present: {}".format(abs_source_path))
                 self.source_refs_deleted += 1
                 sources_to_delete.append(source.id)
 
                 for thumb in source.thumbnails.all():
                     self.thumbnails_deleted += 1
-                    abs_thumbnail_path = self._get_absolute_path(thumb.name)
+                    abs_thumbnail_path = self._get_absolute_path(thumb.name, storage)
 
                     if self._check_if_exists(storage, abs_thumbnail_path):
                         if not dry_run:

easy_thumbnails/tests/settings.py

@@ -31,7 +31,7 @@
 
 STORAGES = {
     "easy_thumbnails": {
-        "BACKEND": "django.core.files.storage.FileSystemStorage",
+        "BACKEND": "easy_thumbnails.tests.utils.TemporaryStorage",
     },
     "default": {
         "BACKEND": "django.core.files.storage.FileSystemStorage",

easy_thumbnails/tests/test_thumbnail_cleanup.py

@@ -0,0 +1,127 @@
+import os
+from datetime import timedelta
+from django.test import override_settings
+from django.core.management import call_command
+from django.utils.timezone import now
+from easy_thumbnails.models import Source, Thumbnail
+from easy_thumbnails.files import get_thumbnailer
+from easy_thumbnails.tests import utils as test
+from django.conf import settings
+
+
+@override_settings(MEDIA_ROOT=os.path.join(settings.MEDIA_ROOT, "test_media"))
+class ThumbnailCleanupTest(test.BaseTest):
+
+    def setUp(self):
+        super().setUp()
+        self.storage = test.TemporaryStorage()
+
+        # Create a source image
+        filename = self.create_image(self.storage, "test.jpg")
+        self.source_image_path = self.storage.open(filename).name
+
+        # Save a test image in both storages.
+        self.thumbnailer = get_thumbnailer(self.storage, filename)
+        self.thumbnailer.generate_thumbnail({"size": (100, 100)})
+
+        self.thumbnail_name = self.thumbnailer.get_thumbnail_name({"size": (100, 100)})
+        self.thumbnail_path = self.thumbnailer.get_thumbnail({"size": (100, 100)}).path
+
+        self.source = Source.objects.get(name=filename)
+
+    def tearDown(self):
+        # Clean up files
+        if os.path.exists(self.source_image_path):
+            os.remove(self.source_image_path)
+        if os.path.exists(self.thumbnail_path):
+            os.remove(self.thumbnail_path)
+
+        # Clean up the database
+        Source.objects.all().delete()
+        Thumbnail.objects.all().delete()
+
+        # Remove test media directory if empty
+        if os.path.exists(settings.MEDIA_ROOT) and not os.listdir(settings.MEDIA_ROOT):
+            os.rmdir(settings.MEDIA_ROOT)
+
+    def test_cleanup_command(self):
+        print(self.source_image_path)
+        self.assertTrue(os.path.exists(self.source_image_path))
+        self.assertTrue(os.path.exists(self.thumbnail_path))
+
+        # Delete the source image to simulate a missing source image
+        os.remove(self.source_image_path)
+        self.assertFalse(os.path.exists(self.source_image_path))
+
+        # Run the thumbnail cleanup command
+        call_command("thumbnail_cleanup", verbosity=2)
+
+        # Verify the thumbnail has been deleted
+        self.assertFalse(os.path.exists(self.thumbnail_path))
+
+        # Verify the source reference has been deleted
+        with self.assertRaises(Source.DoesNotExist):
+            Source.objects.get(id=self.source.id)
+
+    def test_cleanup_dry_run(self):
+        self.assertTrue(os.path.exists(self.source_image_path))
+        self.assertTrue(os.path.exists(self.thumbnail_path))
+
+        # Delete the source image to simulate a missing source image
+        os.remove(self.source_image_path)
+        self.assertFalse(os.path.exists(self.source_image_path))
+
+        # Run the thumbnail cleanup command in dry run mode
+        call_command("thumbnail_cleanup", dry_run=True, verbosity=2)
+
+        # Verify the thumbnail has not been deleted
+        self.assertTrue(os.path.exists(self.thumbnail_path))
+
+        # Verify the source reference has not been deleted
+        self.assertIsNotNone(Source.objects.get(id=self.source.id))
+
+    def test_cleanup_last_n_days(self):
+        old_time = now() - timedelta(days=10)
+        self.source.modified = old_time
+        self.source.save()
+
+        self.assertTrue(os.path.exists(self.source_image_path))
+        self.assertTrue(os.path.exists(self.thumbnail_path))
+
+        # Delete the source image to simulate a missing source image
+        os.remove(self.source_image_path)
+        self.assertFalse(os.path.exists(self.source_image_path))
+
+        # Run the thumbnail cleanup command with last_n_days parameter
+        call_command("thumbnail_cleanup", last_n_days=5, verbosity=2)
+
+        # Verify the thumbnail has not been deleted
+        self.assertTrue(os.path.exists(self.thumbnail_path))
+
+        # Verify the source reference has not been deleted
+        self.assertIsNotNone(Source.objects.get(id=self.source.id))
+
+        # Run the thumbnail cleanup command with last_n_days parameter that includes the source
+        call_command("thumbnail_cleanup", last_n_days=15, verbosity=2)
+
+        # Verify the thumbnail has been deleted
+        self.assertFalse(os.path.exists(self.thumbnail_path))
+
+        # Verify the source reference has been deleted
+        with self.assertRaises(Source.DoesNotExist):
+            Source.objects.get(id=self.source.id)
+
+    def test_source_storage_hash_not_found(self):
+        self.assertTrue(os.path.exists(self.source_image_path))
+        self.assertTrue(os.path.exists(self.thumbnail_path))
+
+        # Change the source's storage_hash to simulate an unknown storage hash
+        self.source.storage_hash = "unknown_storage_hash"
+        self.source.save()
+
+        # Run the thumbnail cleanup command
+        call_command("thumbnail_cleanup", verbosity=2)
+
+        # Verify the thumbnail and source still exist
+        self.assertTrue(os.path.exists(self.thumbnail_path))
+        self.assertIsNotNone(Source.objects.get(id=self.source.id))

tox.ini

@@ -1,18 +1,13 @@
 [tox]
 distribute = False
 envlist =
-    py{36,37,38,39,310}-django32{-svg,}
-    py{38,39,310,311}-django40{-svg,}
-    py{38,39,310,311}-django41{-svg,}
     py{38,39,310,311,312}-django42{-svg,}
     py{310,311,312}-django50{-svg,}
     py{310,311,312}-django51{-svg,}
 skip_missing_interpreters = True
 
 [gh-actions]
 python =
-    3.6: py36
-    3.7: py37
     3.8: py38
     3.9: py39
     3.10: py310
@@ -26,9 +21,6 @@ usedevelop = True
 extras =
     svg: svg
 deps =
-    django32: Django<3.3
-    django40: Django<4.1
-    django41: Django<4.2
     django42: Django<4.3
     django50: Django<5.1
     django51: Django>=5.1a1,<5.2