feat(storages): move copy to storage field

fix document clone ssec storage usage

Author: Yelinz

alexandria/core/models.py

@@ -1,14 +1,12 @@
 import re
 import uuid
 from pathlib import Path
-from tempfile import NamedTemporaryFile
 
 from django.conf import settings
 from django.contrib.postgres.fields import ArrayField
 from django.contrib.postgres.indexes import GinIndex
 from django.contrib.postgres.search import SearchVectorField
 from django.core.exceptions import ImproperlyConfigured, ObjectDoesNotExist
-from django.core.files import File as DjangoFile
 from django.core.validators import RegexValidator
 from django.db import models, transaction
 from django.dispatch import receiver
@@ -18,7 +16,6 @@
 from manabi.token import Key, Token
 from rest_framework_json_api.relations import reverse
 
-from alexandria.storages.backends.s3 import S3Storage
 from alexandria.storages.fields import DynamicStorageFileField
 
 
@@ -171,38 +168,11 @@ def clone(self):
 
         self.pk = None
         self.save()
+        new_name = f"{self.pk}_{latest_original.name}"
 
-        storage = File.content.field.storage
         latest_original.pk = None
         latest_original.document = self
-        if isinstance(storage, S3Storage):
-            new_name = f"{self.pk}_{latest_original.name}"
-            copy_args = {
-                "CopySource": {
-                    "Bucket": settings.ALEXANDRIA_S3_BUCKET_NAME,
-                    "Key": latest_original.content.name,
-                },
-                # Destination settings
-                "Bucket": settings.ALEXANDRIA_S3_BUCKET_NAME,
-                "Key": new_name,
-            }
-
-            if settings.ALEXANDRIA_ENABLE_AT_REST_ENCRYPTION:
-                copy_args["CopySourceSSECustomerKey"] = storage.ssec_secret
-                copy_args["CopySourceSSECustomerAlgorithm"] = "AES256"
-                copy_args["SSECustomerKey"] = storage.ssec_secret
-                copy_args["SSECustomerAlgorithm"] = "AES256"
-
-            storage.bucket.meta.client.copy_object(**copy_args)
-            latest_original.content = new_name
-            latest_original.save()
-        else:
-            with NamedTemporaryFile() as tmp:
-                temp_file = Path(tmp.name)
-                with temp_file.open("w+b") as file:
-                    file.write(latest_original.content.file.file.read())
-                    latest_original.content = DjangoFile(file)
-                    latest_original.save()
+        latest_original.content.copy(new_name)
 
         return self
 

alexandria/core/tests/test_models.py

@@ -54,9 +54,7 @@ def test_document_no_files(
 
 
 def test_clone_document_s3(db, mocker, settings, file_factory):
-    settings.ALEXANDRIA_FILE_STORAGE = (
-        "alexandria.storages.backends.s3.SsecGlobalS3Storage"
-    )
+    settings.ALEXANDRIA_FILE_STORAGE = "alexandria.storages.backends.s3.S3Storage"
     settings.ALEXANDRIA_ENABLE_AT_REST_ENCRYPTION = True
     name = "name-of-the-file"
     mocker.patch("storages.backends.s3.S3Storage.save", return_value=name)
@@ -73,12 +71,14 @@ def test_clone_document_s3(db, mocker, settings, file_factory):
             content=FileData.png,
             content_type="image/png",
         ),
+        encryption_status=File.EncryptionStatus.SSEC_GLOBAL_KEY,
     )
 
     file_factory(
         original=original_file,
         variant=File.Variant.THUMBNAIL,
         document=original_file.document,
+        encryption_status=File.EncryptionStatus.SSEC_GLOBAL_KEY,
     )
 
     original_document_pk = original_file.document.pk

alexandria/storages/fields.py

@@ -1,4 +1,8 @@
+from pathlib import Path
+from tempfile import NamedTemporaryFile
+
 from django.conf import ImproperlyConfigured, settings
+from django.core.files import File as DjangoFile
 from django.core.files.storage import storages
 from django.db import models
 from django.db.models.fields.files import FieldFile
@@ -10,14 +14,52 @@
 class DynamicStorageFieldFile(FieldFile):
     def __init__(self, instance, field, name):
         super().__init__(instance, field, name)
-        self.storage = storages.create_storage(
-            {"BACKEND": settings.ALEXANDRIA_FILE_STORAGE}
-        )
+        storage_backend = settings.ALEXANDRIA_FILE_STORAGE
         if settings.ALEXANDRIA_ENABLE_AT_REST_ENCRYPTION:
             from alexandria.core.models import File
 
             if instance.encryption_status == File.EncryptionStatus.SSEC_GLOBAL_KEY:
-                self.storage = SsecGlobalS3Storage()
+                storage_backend = "alexandria.storages.backends.s3.SsecGlobalS3Storage"
+        self.storage = storages.create_storage({"BACKEND": storage_backend})
+
+    def copy(self, target_name: str):
+        """
+        Copy file content to target.
+
+        NOTE: this will copy the content and this copy will be the new reference of the FileField.
+        This means that there will be no pointer to the original content,
+        unless you made a copy of the parent model instance beforhands by e.g.
+        setting `file_instance.pk` to `None` before calling `file_instance.content.copy`.
+        """
+        # S3 compatible storage: copy in same bucket with s3 copy
+        if isinstance(self.storage, S3Storage):
+            copy_args = {
+                "CopySource": {
+                    "Bucket": self.storage.bucket,
+                    "Key": self.name,
+                },
+                # Destination settings
+                "Bucket": self.storage.bucket,
+                "Key": target_name,
+            }
+            if isinstance(self.storage, SsecGlobalS3Storage):
+                copy_args["CopySourceSSECustomerKey"] = self.storage.ssec_secret
+                copy_args["CopySourceSSECustomerAlgorithm"] = "AES256"
+                copy_args["SSECustomerKey"] = self.storage.ssec_secret
+                copy_args["SSECustomerAlgorithm"] = "AES256"
+
+            self.storage.bucket.meta.client.copy_object(**copy_args)
+            self.instance.content = target_name
+            self.instance.save()
+            return
+
+        # otherwise use filesystem storage
+        with NamedTemporaryFile() as tmp:
+            temp_file = Path(tmp.name)
+            with temp_file.open("w+b") as file:
+                file.write(self.read())
+                self.instance.content = DjangoFile(file, target_name)
+                self.instance.save()
 
 
 class DynamicStorageFileField(models.FileField):
@@ -56,8 +98,8 @@ def pre_save(self, instance, add):
                     "Set `ALEXANDRIA_FILE_STORAGE` to `alexandria.storages.s3.S3Storage`."
                 )
                 raise ImproperlyConfigured(msg)
-            storage = SsecGlobalS3Storage()
             if instance.encryption_status == File.EncryptionStatus.SSEC_GLOBAL_KEY:
-                self.storage = storage
-        _file = super().pre_save(instance, add)
-        return _file
+                self.storage = storages.create_storage(
+                    {"BACKEND": "alexandria.storages.backends.s3.SsecGlobalS3Storage"}
+                )
+        return super().pre_save(instance, add)

alexandria/storages/tests/test_dynamic_field.py

@@ -27,7 +27,7 @@ def test_dynamic_storage_select_global_ssec(
     mocker.patch("alexandria.core.tasks.create_thumbnail.delay", side_effect=None)
     if raises is not None:
         with pytest.raises(raises):
-            file_factory()
+            file_factory(encryption_status=settings.ALEXANDRIA_ENCRYPTION_METHOD)
         return
     file_factory(encryption_status=settings.ALEXANDRIA_ENCRYPTION_METHOD)
     assert SsecGlobalS3Storage.save.called_once()