Functionality for the deleting application attachments

hypha/apply/activity/adapters/emails.py

@@ -175,8 +175,6 @@ def handle_transition(self, new_phase, source, old_phase=None, **kwargs):
         old_index = list(dict(PHASES).keys()).index(old_phase.name)
         target_index = list(dict(PHASES).keys()).index(submission.status)
         is_forward = old_index < target_index
-        print("NEW PHASE")
-        print(new_phase.public_name)
 
         kwargs["old_phase"] = old_phase.public_name
         kwargs["new_phase"] = new_phase.public_name

hypha/apply/funds/apps.py

@@ -1,5 +1,12 @@
 from django.apps import AppConfig
+from django.core.signals import request_finished
 
 
 class ApplyConfig(AppConfig):
     name = "hypha.apply.funds"
+
+    def ready(self):
+        # Connect the attachment deletion handler
+        from . import signals
+
+        request_finished.connect(signals.delete_attachments)

hypha/apply/funds/migrations/0131_delete_orphaned_attachments.py

@@ -0,0 +1,61 @@
+# Generated by Django 4.2.24 on 2025-10-01 15:39
+
+from django.db import migrations
+from django.core.files.storage import default_storage
+import sys
+from ..utils import delete_directory
+
+
+def delete_orphaned_attachments(apps, schema_editor):
+    """Remove all attachments not associated with an application"""
+
+    # TODO: This solution is not ideal but due to our unit tests writing to the filesystem
+    #       this can cause files belonging to the dev's local server to be deleted. Until
+    #       these can be better isolated, this signal will do nothing when pytest is running
+    if "pytest" in sys.modules:
+        return
+
+    ApplicationSubmission = apps.get_model("funds", "ApplicationSubmission")
+
+    submission_attachment_path = "submission"
+
+    folders_to_delete = []
+    folders_to_check = []
+
+    if not default_storage.exists(submission_attachment_path):
+        # If specified path doesn't exist, ignore
+        # edge case that typically comes up in tests
+        return
+
+    for folder in default_storage.listdir(submission_attachment_path)[0]:
+        # `listdir` returns ([folders], [files]) ^
+        try:
+            folders_to_check.append(int(folder))
+        except ValueError:
+            # Folder name is not an int, therefore not a submission ID and can be deleted (an edge case)
+            folders_to_delete.append(folder)
+
+    # Get a list of all undeleted submissions that have a folder
+    valid_ids = set(
+        ApplicationSubmission.objects.filter(id__in=folders_to_check).values_list(
+            "id", flat=True
+        )
+    )
+
+    # Find the set difference and delete those folders
+    folders_to_delete += list(set(folders_to_check) - valid_ids)
+
+    for folder in folders_to_delete:
+        try:
+            delete_directory(f"{submission_attachment_path}/{folder}")
+        except FileNotFoundError:
+            # Will get thrown when unit tests attempt to run migrations
+            pass
+
+
+class Migration(migrations.Migration):
+    dependencies = [
+        ("funds", "0130_alter_applicationsubmission_status"),
+    ]
+
+    operations = [migrations.RunPython(delete_orphaned_attachments)]

hypha/apply/funds/models/mixins.py

@@ -161,7 +161,6 @@ def extract_files(self):
         for field in self.form_fields:
             if isinstance(field.block, UploadableMediaBlock):
                 files[field.id] = self.data(field.id) or []
-                self.form_data.pop(field.id, None)
         return files
 
     @classmethod

hypha/apply/funds/signals.py

@@ -0,0 +1,40 @@
+import sys
+
+from django.core.files.storage import default_storage
+from django.db.models.signals import pre_delete
+from django.dispatch import receiver
+
+from hypha.apply.funds.models.application_revisions import ApplicationRevision
+from hypha.apply.funds.models.submissions import ApplicationSubmission
+from hypha.apply.funds.utils import delete_directory
+
+
+@receiver(signal=pre_delete, sender=ApplicationSubmission)
+@receiver(signal=pre_delete, sender=ApplicationRevision)
+def delete_attachments(sender, instance=None, **kwargs):
+    """
+    Before the deletion of any sub class of AccessFormData, ensure the files associated with it are deleted too.
+
+    This can include things like ApplicationSubmission & ApplicationRevision objects
+    """
+
+    # TODO: This solution is not ideal but due to our unit tests writing to the filesystem
+    #       this can cause files belonging to the dev's local server to be deleted. Until
+    #       these can be better isolated, this signal will do nothing when pytest is running
+    if "pytest" in sys.modules:
+        return
+
+    # This will be called anytime a deletion is ran, so ensure the object being deleted
+    # can have attachments
+    if issubclass(sender, ApplicationRevision):
+        files = instance.extract_files()
+        for value in files.values():
+            if not isinstance(value, list):  # Single file field
+                value.delete()
+            else:  # Multiple file fields
+                [sub_file.delete() for sub_file in value]
+    elif issubclass(sender, ApplicationSubmission):
+        submission_attachment_path = f"submission/{instance.id}"
+
+        if default_storage.exists(submission_attachment_path):
+            delete_directory(submission_attachment_path)

hypha/apply/funds/utils.py

@@ -1,4 +1,5 @@
 import csv
+import os
 import re
 from datetime import datetime
 from functools import reduce
@@ -8,6 +9,7 @@
 from typing import Iterable
 
 import django_filters as filters
+from django.core.files.storage import default_storage
 from django.urls import reverse
 from django.utils.encoding import force_bytes
 from django.utils.html import strip_tags
@@ -267,3 +269,28 @@ def generate_invite_path(invite):
         kwargs={"uidb64": uid, "token": token},
     )
     return login_path
+
+
+def delete_directory(directory_path):
+    """Delete a full directory (empty or not)
+
+    Used in attachment cleanup when deleting submissions/revisions
+    """
+
+    directories, files = default_storage.listdir(directory_path)
+
+    for item in directories:
+        item_path = os.path.join(directory_path, item)
+        if default_storage.exists(item_path):
+            # Recursively delete subdirectories
+            delete_directory(item_path)
+
+    for item in files:
+        item_path = os.path.join(directory_path, item)
+        if default_storage.exists(item_path):
+            # Delete files
+            default_storage.delete(item_path)
+
+    if default_storage.exists(directory_path):
+        # Delete the empty directory
+        default_storage.delete(directory_path)

hypha/apply/stream_forms/files.py

@@ -7,7 +7,7 @@
 
 class StreamFieldDataEncoder(DjangoJSONEncoder):
     def default(self, o):
-        if isinstance(o, StreamFieldFile):
+        if isinstance(o, File):
             return {
                 "name": o.name,
                 "filename": o.filename,

hypha/apply/stream_forms/testing/factories.py

@@ -4,6 +4,7 @@
 
 import factory
 import wagtail_factories
+from django.core.files.base import File
 from django.core.files.uploadedfile import SimpleUploadedFile
 from django.core.serializers.json import DjangoJSONEncoder
 from wagtail.blocks import RichTextBlock, StructValue
@@ -242,6 +243,17 @@ def make_answer(cls, params=None):
         return cls.choices[0]
 
 
+class UploadedFile(SimpleUploadedFile):
+    """Utilized to make functionality closer to that of `StreamFieldFile`
+
+    Requires a `filename` attribute which is pulled from the existing `_name`
+    """
+
+    def __init__(self, name, content, content_type=...):
+        super().__init__(name, content, content_type)
+        self.filename = self._name
+
+
 class UploadableMediaFactory(FormFieldBlockFactory):
     default_value = factory.django.FileField()
 
@@ -252,7 +264,7 @@ def make_answer(cls, params=None):
         if params.get("filename") is None:
             params["filename"] = "test_example.pdf"
         file_name, file = cls.default_value._make_content(params)
-        return SimpleUploadedFile(file_name, file.read())
+        return UploadedFile(file_name, file.read())
 
 
 class ImageFieldBlockFactory(UploadableMediaFactory):
@@ -276,6 +288,16 @@ def make_answer(cls, params=None):
         return [UploadableMediaFactory.make_answer() for _ in range(2)]
 
 
+class StreamFieldDataEncoder(DjangoJSONEncoder):
+    def default(self, o):
+        if isinstance(o, File):
+            return {
+                "name": o.name,
+                "filename": o.filename,
+            }
+        return super().default(o)
+
+
 class StreamFieldUUIDFactory(wagtail_factories.StreamFieldFactory):
     def evaluate(self, instance, step, extra):
         params = self.build_form(extra)