✨(backend) add file specific admin

Adds a file (background image) specific django admin.
Files can be previewed, and deletion is properly managed.

CHANGELOG.md

@@ -8,6 +8,10 @@ and this project adheres to
 
 ## [Unreleased]
 
+### Added
+
+- ✨(backend) add file specific admin #1387
+
 ### Changed
 
 - 🐛(agents) fix bug when closing metadata-collector

src/backend/core/admin.py

@@ -3,17 +3,55 @@
 from django import forms
 from django.contrib import admin, messages
 from django.contrib.auth import admin as auth_admin
+from django.db import transaction
+from django.utils.html import format_html
 from django.utils.translation import gettext_lazy as _
 
 from core.recording.event import notification
 
 from . import models
+from .tasks.file import process_file_deletion
+from .utils import generate_download_file_url
+
+
+def hard_delete_file(file):
+    """Hard delete a file, soft deleting it first when needed."""
+    if file.deleted_at is None:
+        file.soft_delete()
+    file.hard_delete()
+    transaction.on_commit(lambda: process_file_deletion.delay(file.id))
+
+
+class FileInlineFormSet(forms.BaseInlineFormSet):
+    """Inline formset overriding delete behavior for files."""
+
+    def delete_existing(self, obj, commit=True):
+        """Hard delete files instead of calling model.delete()."""
+        hard_delete_file(obj)
+
+
+class FileInline(admin.TabularInline):
+    """Inline class for the File model."""
+
+    model = models.File
+    formset = FileInlineFormSet
+    fk_name = "creator"
+    extra = 0
+    fields = ("id", "title", "type", "upload_state", "created_at")
+    readonly_fields = ("id", "created_at", "upload_state", "type")
+    show_change_link = True
+
+    def get_queryset(self, request):
+        """Hide hard deleted files in the inline."""
+        return super().get_queryset(request).filter(hard_deleted_at__isnull=True)
 
 
 @admin.register(models.User)
 class UserAdmin(auth_admin.UserAdmin):
     """Admin class for the User model"""
 
+    inlines = (FileInline,)
+
     fieldsets = (
         (
             None,
@@ -97,6 +135,136 @@ class UserAdmin(auth_admin.UserAdmin):
     search_fields = ("id", "sub", "admin_email", "email", "full_name")
 
 
+@admin.register(models.File)
+class FileAdmin(admin.ModelAdmin):
+    """Admin class for the File model."""
+
+    list_display = (
+        "id",
+        "title",
+        "type",
+        "creator",
+        "upload_state",
+        "deleted_at",
+        "hard_deleted_at",
+        "created_at",
+        "updated_at",
+    )
+    list_filter = (
+        "type",
+        "upload_state",
+        "created_at",
+        "updated_at",
+        "deleted_at",
+        "hard_deleted_at",
+    )
+    search_fields = (
+        "id",
+        "title",
+        "filename",
+        "mimetype",
+        "description",
+        "creator__email",
+        "creator__admin_email",
+        "creator__full_name",
+    )
+    ordering = ("-created_at",)
+    readonly_fields = (
+        "id",
+        "created_at",
+        "updated_at",
+        "deleted_at",
+        "hard_deleted_at",
+        "description",
+        "malware_detection_info",
+        "is_pending_upload",
+        "preview_url",
+        "extension",
+        "key_base",
+        "file_key",
+        "upload_state",
+        "type",
+        "mimetype",
+        "size",
+    )
+    autocomplete_fields = ("creator",)
+    fieldsets = (
+        (
+            None,
+            {
+                "fields": (
+                    "id",
+                    "title",
+                    "type",
+                    "creator",
+                    "filename",
+                    "upload_state",
+                )
+            },
+        ),
+        (
+            _("Content"),
+            {
+                "fields": (
+                    "mimetype",
+                    "size",
+                    "description",
+                    "malware_detection_info",
+                )
+            },
+        ),
+        (
+            _("Deletion"),
+            {
+                "fields": (
+                    "deleted_at",
+                    "hard_deleted_at",
+                )
+            },
+        ),
+        (
+            _("Derived info"),
+            {
+                "fields": (
+                    "is_pending_upload",
+                    "extension",
+                    "key_base",
+                    "file_key",
+                    "preview_url",
+                )
+            },
+        ),
+        (_("Timestamps"), {"fields": ("created_at", "updated_at")}),
+    )
+
+    @admin.display(description=_("File preview"))
+    def preview_url(self, obj):
+        """Return a clickable preview URL for the file."""
+        if obj.is_pending_upload:
+            return "-"
+        url = generate_download_file_url(obj, expires_in=60 * 60)
+
+        return format_html(
+            '<a href="{}" target="_blank" rel="noopener noreferrer">Open File</a>', url
+        )
+
+    def get_queryset(self, request):
+        """Hide hard deleted files in admin listing and lookups."""
+        return super().get_queryset(request).filter(hard_deleted_at__isnull=True)
+
+    def delete_model(self, request, obj):
+        """Hard delete instead of calling model.delete()."""
+        hard_delete_file(obj)
+
+    def delete_queryset(self, request, queryset):
+        """Hard delete all selected files."""
+        for file in queryset:
+            hard_delete_file(file)
+
+    def has_add_permission(self, request):
+        return False
+
+
 class ResourceAccessInline(admin.TabularInline):
     """Admin class for the room user access model"""
 

src/backend/core/utils.py

@@ -459,6 +459,41 @@ def generate_upload_policy(file):
     return policy
 
 
+def generate_download_file_url(file, *, expires_in: int, override_domain: bool = True):
+    """
+    Generate a S3 signed download url for a given file.
+    """
+
+    key = file.file_key
+
+    # This setting should be used if the backend application and the frontend application
+    # can't connect to the object storage with the same domain. This is the case in the
+    # docker compose stack used in development. The frontend application will use localhost
+    # to connect to the object storage while the backend application will use the object storage
+    # service name declared in the docker compose stack.
+    # This is needed because the domain name is used to compute the signature. So it can't be
+    # changed dynamically by the frontend application.
+    if settings.AWS_S3_DOMAIN_REPLACE and override_domain:
+        s3_client = boto3.client(
+            "s3",
+            aws_access_key_id=settings.AWS_S3_ACCESS_KEY_ID,
+            aws_secret_access_key=settings.AWS_S3_SECRET_ACCESS_KEY,
+            endpoint_url=settings.AWS_S3_DOMAIN_REPLACE,
+            config=botocore.client.Config(
+                region_name=settings.AWS_S3_REGION_NAME,
+                signature_version=settings.AWS_S3_SIGNATURE_VERSION,
+            ),
+        )
+    else:
+        s3_client = default_storage.connection.meta.client
+
+    return s3_client.generate_presigned_url(
+        ClientMethod="get_object",
+        Params={"Bucket": default_storage.bucket_name, "Key": key},
+        ExpiresIn=expires_in,
+    )
+
+
 @lru_cache(maxsize=1)
 def _format_telephony_phone_number(raw_number, default_country):
     """Parse a configured phone number and return (country, international_format).