Allow callables and bools for nocompress, add rotate_storage command

Author: dcwatson

CHANGELOG.md

@@ -3,6 +3,8 @@
 * Drop support for Python 3.9
 * Test on Python 3.14 and Django 5.2
 * Support opening files for writing
+* Allow `nocompress` option to accept a callable, or `True` to indicate never compress
+* Experimental `rotate_storage` Django command for rotating file fields stored in `PZipStorage`
 
 
 ## 1.2.1 (2024-11-08)

README.md

@@ -16,7 +16,7 @@ The simplest way to use `PZipStorage` is by setting your
 without first rotating the keys of all stored files, they will be lost forever.
 
 `PZipStorage` may be used with existing unencrypted files, as a drop-in replacement for `FileSystemStorage`. If it
-determines the requested file is not a PZip file, it will delegate to `FileSystemStorage` after emitting a
+determined the requested file is not a PZip file, it will delegate to `FileSystemStorage` after emitting a
 `needs_encryption` signal (see below).
 
 You may also use `PZipStorage` as a custom storage backend anywhere Django allows it; see

pyproject.toml

@@ -1,6 +1,6 @@
 [project]
 name = "django-pzip-storage"
-version = "1.3.0"
+version = "1.3.0.dev0"
 description = "Storage backend for Django that encrypts/compresses with PZip."
 readme = "README.md"
 authors = [

src/pzip_storage/__init__.py

@@ -73,6 +73,8 @@ def __init__(self, *args, **kwargs):
             "nocompress",
             getattr(settings, "PZIP_STORAGE_NOCOMPRESS", self.DEFAULT_NOCOMPRESS),
         )
+        if callable(self.nocompress):
+            self.nocompress = self.nocompress()
         if not self.keys:
             raise ImproperlyConfigured("PZipStorage requires at least one key.")
         super().__init__(*args, **kwargs)
@@ -129,6 +131,8 @@ def _open(self, name, mode="rb"):
         return super()._open(name, mode)
 
     def should_compress(self, name):
+        if isinstance(self.nocompress, bool):
+            return not self.nocompress
         return os.path.splitext(name)[1].lower() not in self.nocompress
 
     def get_write_key(self):

src/pzip_storage/management/__init__.py

@@ -0,0 +1,82 @@
+import os
+
+from django.apps import apps
+from django.core.management.base import BaseCommand
+from django.db import models, transaction
+
+from pzip_storage import PZipStorage
+
+
+def rotate_model(model, fields, storage_changes, save=True, verbosity=0):
+    for instance in model.objects.all():
+        changed = []
+        for field in fields:
+            old = getattr(instance, field.name)
+            if not old:
+                continue
+            path_changes = storage_changes.setdefault(field.storage, {})
+            if old.name in path_changes:
+                raise ValueError(f"Duplicate path in {instance!r}: {old.name}")
+            try:
+                new_name = field.generate_filename(instance, os.path.basename(old.name))
+                new_path = field.storage.save(new_name, old)
+                setattr(instance, field.attname, new_path)
+                changed.append(field.attname)
+                path_changes[old.name] = new_path
+                if not save and verbosity > 1:
+                    print(old.name, "->", new_path)
+            except OSError as e:
+                if verbosity > 0:
+                    print("error rotating", old, "-", str(e))
+        if changed and save:
+            instance.save(update_fields=changed)
+
+
+class Command(BaseCommand):
+    help = "Rotates FileFields stored in PZipStorage"
+
+    def add_arguments(self, parser):
+        parser.add_argument("-n", "--dry-run", action="store_true", default=False)
+        parser.add_argument("--delete", action="store_true", default=False)
+        parser.add_argument("models", nargs="*")
+
+    def handle(self, *args, **options):
+        verbosity = options["verbosity"]
+        save = not options["dry_run"]
+        rotate_models = [apps.get_model(m) for m in options["models"]]
+        if not rotate_models:
+            rotate_models = apps.get_models()
+        # Track {storage: {old_path: new_path}}
+        storage_changes = {}
+        with transaction.atomic():
+            for model in rotate_models:
+                file_fields = [
+                    f
+                    for f in model._meta.fields
+                    if isinstance(f, models.FileField)
+                    and isinstance(f.storage, PZipStorage)
+                ]
+                if file_fields:
+                    rotate_model(
+                        model,
+                        file_fields,
+                        storage_changes,
+                        save=save,
+                        verbosity=verbosity,
+                    )
+        if options["delete"]:
+            for storage, path_changes in storage_changes.items():
+                if save:
+                    # Real run, delete the old files.
+                    delete_paths = set(path_changes.keys())
+                else:
+                    # For a dry run, delete the newly-created files.
+                    delete_paths = set(path_changes.values())
+                for path in delete_paths:
+                    try:
+                        storage.delete(path)
+                        if verbosity > 1:
+                            print(f"removed {path}")
+                    except Exception as e:
+                        if verbosity > 0:
+                            print("error removing", path, "-", str(e))

src/pzip_storage/management/commands/__init__.py

@@ -60,6 +60,16 @@ def test_no_compression(self):
         with self.storage.open(name) as f:
             self.assertIsInstance(f, pzip.PZip)
             self.assertEqual(f.compression, pzip.Compression.NONE)
+        # Compress everything.
+        compressed = PZipStorage(nocompress=False)
+        with compressed.open("test.jpg", "wb") as f:
+            self.assertIsInstance(f, pzip.PZip)
+            self.assertEqual(f.compression, pzip.Compression.GZIP)
+        # Compress nothing.
+        raw = PZipStorage(nocompress=True)
+        with raw.open("test.txt", "wb") as f:
+            self.assertIsInstance(f, pzip.PZip)
+            self.assertEqual(f.compression, pzip.Compression.NONE)
 
     def test_unencrypted(self):
         handler = MagicMock()