chore(user files): introduced management command to permanently delete user files (baserow#4178)

bram2w · silvestrid · web-flow · commit b4926d6bbc55 · 2025-12-24T11:05:06.000+01:00
* Introduced management command to permanently delete user files uploaded by user

* added tests

* Fix copilot feedback

* allow reuploading file

* fixed migration

* Fix tests

---------

Co-authored-by: Davide Silvestri &lt;silvestri.eng@gmail.com&gt;
diff --git a/backend/src/baserow/core/management/commands/delete_files_uploaded_by_user.py b/backend/src/baserow/core/management/commands/delete_files_uploaded_by_user.py
@@ -0,0 +1,96 @@
+from django.conf import settings
+from django.core.management.base import BaseCommand
+from django.utils.timezone import now
+
+from baserow.core.models import UserFile
+from baserow.core.storage import get_default_storage
+from baserow.core.user_files.handler import UserFileHandler
+
+
+class Command(BaseCommand):
+    help = (
+        "This command permanently deletes all the user files uploaded by the provided "
+        "user. It deletes the files where the `uploaded_by` matches the given `user_id`. "
+        "Please note that this could lead to a situation where a file is deleted that "
+        "another user also uploaded, if it's exactly the same file and the original "
+        "filename was the same."
+    )
+
+    def add_arguments(self, parser):
+        parser.add_argument(
+            "user_id",
+            type=int,
+            help="The ID of the user whose uploaded files should be permanently deleted.",
+        )
+        parser.add_argument(
+            "--yes",
+            action="store_true",
+            help="Skip confirmation prompt and delete immediately.",
+        )
+
+    def handle(self, *args, **options):
+        user_id = options["user_id"]
+        skip_confirmation = options["yes"]
+
+        storage = get_default_storage()
+        handler = UserFileHandler()
+
+        user_files = UserFile.objects.filter(
+            uploaded_by_id=user_id, deleted_at__isnull=True
+        )
+        count = user_files.count()
+
+        self.stdout.write(f"\nFound {count} file(s) uploaded by user ID {user_id}\n")
+
+        if count == 0:
+            return
+
+        for user_file in user_files:
+            self.stdout.write(
+                f"- {user_file.name} ({user_file.size} bytes, "
+                f"uploaded {user_file.uploaded_at})"
+            )
+
+        if not skip_confirmation:
+            confirm = input(
+                "\nAre you sure you want to permanently delete these files from storage? [y/N]: "
+            )
+            if confirm.lower() not in ["y", "yes"]:
+                self.stdout.write(self.style.WARNING("Aborted by user."))
+                return
+
+        deleted_count = 0
+        for user_file in user_files:
+            try:
+                file_path = handler.user_file_path(user_file)
+                if storage.exists(file_path):
+                    storage.delete(file_path)
+                    self.stdout.write(f"Deleted: {file_path}")
+                else:
+                    self.stdout.write(
+                        self.style.WARNING(f"File not found in storage: {file_path}")
+                    )
+
+                if getattr(settings, "USER_THUMBNAILS_DIRECTORY", None):
+                    for thumb_name in getattr(settings, "USER_THUMBNAILS", {}).keys():
+                        thumb_path = handler.user_file_thumbnail_path(
+                            user_file, thumb_name
+                        )
+                        if storage.exists(thumb_path):
+                            storage.delete(thumb_path)
+                            self.stdout.write(f"Deleted thumbnail: {thumb_path}")
+
+                user_file.deleted_at = now()
+                user_file.save(update_fields=["deleted_at"])
+                deleted_count += 1
+
+            except Exception as e:
+                self.stderr.write(
+                    self.style.ERROR(f"Error deleting file {user_file.name}: {e}")
+                )
+
+        self.stdout.write(
+            self.style.SUCCESS(
+                f"\nSuccessfully deleted {deleted_count} file(s) for user ID {user_id}."
+            )
+        )
diff --git a/backend/src/baserow/core/migrations/0109_userfile_deleted_at.py b/backend/src/baserow/core/migrations/0109_userfile_deleted_at.py
@@ -0,0 +1,17 @@
+# Generated by Django 5.0.14 on 2025-11-06 21:29
+
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+    dependencies = [
+        ("core", "0108_alter_userfile_unique"),
+    ]
+
+    operations = [
+        migrations.AddField(
+            model_name="userfile",
+            name="deleted_at",
+            field=models.DateTimeField(blank=True, default=None, null=True),
+        ),
+    ]
diff --git a/backend/src/baserow/core/user_files/handler.py b/backend/src/baserow/core/user_files/handler.py
@@ -268,7 +268,9 @@ def upload_user_file(self, user, file_name, stream, storage=None):
         file_name = truncate_middle(file_name, 64)
 
         existing_user_file = UserFile.objects.filter(
-            original_name=file_name, sha256_hash=stream_hash
+            original_name=file_name,
+            sha256_hash=stream_hash,
+            deleted_at__isnull=True,
         ).first()
 
         if existing_user_file:
diff --git a/backend/src/baserow/core/user_files/models.py b/backend/src/baserow/core/user_files/models.py
@@ -22,6 +22,7 @@ class UserFile(models.Model):
     uploaded_at = models.DateTimeField(auto_now_add=True)
     uploaded_by = models.ForeignKey(User, on_delete=models.SET_NULL, null=True)
     sha256_hash = models.CharField(max_length=64, db_index=True)
+    deleted_at = models.DateTimeField(null=True, blank=True, default=None)
 
     objects = UserFileQuerySet.as_manager()
 
diff --git a/backend/tests/baserow/core/management/test_delete_files_uploaded_by_user.py b/backend/tests/baserow/core/management/test_delete_files_uploaded_by_user.py
@@ -0,0 +1,152 @@
+from io import StringIO
+from unittest.mock import MagicMock, patch
+
+from django.core.management import call_command
+
+import pytest
+
+
+@pytest.mark.django_db
+def test_delete_files_uploaded_by_user_no_files(data_fixture):
+    user = data_fixture.create_user()
+    out = StringIO()
+
+    call_command("delete_files_uploaded_by_user", user.id, stdout=out)
+
+    output = out.getvalue()
+    assert f"Found 0 file(s) uploaded by user ID {user.id}" in output
+
+
+@pytest.mark.django_db
+def test_delete_files_uploaded_by_user_aborts_on_no_confirmation(data_fixture):
+    user = data_fixture.create_user()
+    user_file = data_fixture.create_user_file(uploaded_by=user)
+
+    with patch("builtins.input", return_value="n"):
+        out = StringIO()
+        call_command("delete_files_uploaded_by_user", user.id, stdout=out)
+
+    output = out.getvalue()
+    assert "Aborted by user." in output
+    user_file.refresh_from_db()
+    assert user_file.deleted_at is None
+
+
+@pytest.mark.django_db
+def test_delete_files_uploaded_by_user_deletes_files_with_yes_flag(data_fixture):
+    user = data_fixture.create_user()
+    user_file = data_fixture.create_user_file(uploaded_by=user)
+
+    mock_storage = MagicMock()
+    mock_storage.exists.return_value = True
+
+    with patch(
+        "baserow.core.management.commands.delete_files_uploaded_by_user.get_default_storage",
+        return_value=mock_storage,
+    ):
+        out = StringIO()
+        call_command("delete_files_uploaded_by_user", user.id, "--yes", stdout=out)
+
+    output = out.getvalue()
+    user_file.refresh_from_db()
+
+    assert "Successfully deleted 1 file(s)" in output
+    assert user_file.deleted_at is not None
+    assert mock_storage.delete.called
+
+
+@pytest.mark.django_db
+def test_delete_files_uploaded_by_user_deletes_thumbnails(data_fixture, settings):
+    user = data_fixture.create_user()
+    data_fixture.create_user_file(uploaded_by=user)
+
+    settings.USER_THUMBNAILS_DIRECTORY = "thumbnails"
+    settings.USER_THUMBNAILS = {"tiny": [32, 32], "small": [64, 64]}
+
+    mock_storage = MagicMock()
+    mock_storage.exists.return_value = True
+
+    with patch(
+        "baserow.core.management.commands.delete_files_uploaded_by_user.get_default_storage",
+        return_value=mock_storage,
+    ):
+        out = StringIO()
+        call_command("delete_files_uploaded_by_user", user.id, "--yes", stdout=out)
+
+    output = out.getvalue()
+
+    # Verify main file was deleted
+    assert "Successfully deleted 1 file(s)" in output
+    # Verify thumbnails were also deleted
+    delete_calls = [call[0][0] for call in mock_storage.delete.call_args_list]
+    assert any("thumbnails/tiny" in path for path in delete_calls)
+    assert any("thumbnails/small" in path for path in delete_calls)
+
+
+@pytest.mark.django_db
+def test_delete_files_uploaded_by_user_handles_missing_files(data_fixture):
+    user = data_fixture.create_user()
+    user_file = data_fixture.create_user_file(uploaded_by=user)
+
+    mock_storage = MagicMock()
+    mock_storage.exists.return_value = False
+
+    with patch(
+        "baserow.core.management.commands.delete_files_uploaded_by_user.get_default_storage",
+        return_value=mock_storage,
+    ):
+        out = StringIO()
+        call_command("delete_files_uploaded_by_user", user.id, "--yes", stdout=out)
+
+    output = out.getvalue()
+    user_file.refresh_from_db()
+
+    assert "File not found in storage" in output
+    assert user_file.deleted_at is not None
+
+
+@pytest.mark.django_db
+def test_delete_files_uploaded_by_user_logs_exceptions_but_continues(data_fixture):
+    user = data_fixture.create_user()
+    file1 = data_fixture.create_user_file(uploaded_by=user)
+    file2 = data_fixture.create_user_file(uploaded_by=user)
+
+    mock_storage = MagicMock()
+    mock_storage.exists.return_value = True
+
+    calls = {"count": 0}
+
+    def delete_side_effect(*args, **kwargs):
+        if calls["count"] == 0:
+            calls["count"] += 1
+            raise Exception("delete failed")
+        return None
+
+    mock_storage.delete.side_effect = delete_side_effect
+
+    with patch(
+        "baserow.core.management.commands.delete_files_uploaded_by_user.get_default_storage",
+        return_value=mock_storage,
+    ):
+        out = StringIO()
+        err = StringIO()
+        call_command(
+            "delete_files_uploaded_by_user",
+            user.id,
+            "--yes",
+            stdout=out,
+            stderr=err,
+        )
+
+    error_output = err.getvalue()
+    file1.refresh_from_db()
+    file2.refresh_from_db()
+
+    # The error should be logged for the first file
+    assert "Error deleting file" in error_output
+
+    # file1 failed before marking deleted
+    assert file1.deleted_at is None
+
+    # file2 should succeed and be marked deleted
+    assert file2.deleted_at is not None
diff --git a/changelog/entries/unreleased/feature/delete_user_files_uploaded_by.json b/changelog/entries/unreleased/feature/delete_user_files_uploaded_by.json
@@ -0,0 +1,8 @@
+{
+  "type": "feature",
+  "message": "Introduced management command to permanently delete user files uploaded by user.",
+  "domain": "core",
+  "issue_number": null,
+  "bullet_points": [],
+  "created_at": "2025-11-06"
+}
