feat: export archived records and back-fill file content in team sync

apps/documents/tasks.py

@@ -234,7 +234,7 @@ def async_create_collection_version(self, collection_id: int):
 )
 def delete_collection_task(self, collection_id: int):
     try:
-        collection = Collection.objects.get(id=collection_id)
+        collection = Collection.objects.get_all().get(id=collection_id)
     except Collection.DoesNotExist:
         return
 

apps/documents/tests/test_tasks.py

@@ -15,6 +15,7 @@
 from apps.documents.tasks import (
     async_create_collection_version,
     create_collection_zip_task,
+    delete_collection_task,
     index_collection_files_task,
     migrate_vector_stores,
     sync_all_document_sources_task,
@@ -37,6 +38,24 @@ def collection(db):
     )
 
 
+@pytest.mark.django_db()
+def test_delete_collection_task_deletes_files_of_archived_collection():
+    """delete_collection_task runs after Collection.archive() has already set is_archived=True, so it
+    must load the archived row and delete its files. Regression: fetching via Collection.objects (which
+    hides archived rows) made the task a silent no-op, leaving CollectionFile rows orphaned."""
+    collection = CollectionFactory.create(is_index=False)
+    file = FileFactory.create(team=collection.team)
+    collection.files.add(file)
+    collection.is_archived = True
+    collection.save(update_fields=["is_archived"])
+    assert CollectionFile.objects.filter(collection=collection).exists()
+
+    with patch("apps.documents.utils.get_related_m2m_objects", return_value=[]):
+        delete_collection_task(collection.id)
+
+    assert not CollectionFile.objects.filter(collection=collection).exists()
+
+
 @pytest.mark.django_db()
 @patch("apps.documents.models.Collection.add_files_to_index")
 def test_collection_files_grouped_by_chunking_strategy(add_files_to_index_mock, collection):

apps/teams/export/client.py

@@ -9,6 +9,16 @@
 _API_KEY_HEADER = "X-Api-Key"
 
 
+class FileContentNotFound(Exception):
+    """The source's file content API returned 404 -- the source is missing the blob too, so it can't
+    be backfilled. Distinct from other HTTP errors so the importer can record it and carry on rather
+    than aborting the sync."""
+
+    def __init__(self, file_id: int) -> None:
+        self.file_id = file_id
+        super().__init__(f"file {file_id} has no content on the source")
+
+
 class ResourceFetcher:
     def __init__(self, base_url, api_key, *, session=None, sleep=time.sleep):
         self.base_url = base_url.rstrip("/")
@@ -49,7 +59,24 @@ def iter_rows(self, resource, start_cursor=None, limit=100):
                 break
             cursor = page["cursor"]
 
+    def get_file_content(self, file_id: int) -> bytes:
+        """The raw bytes of a file, from the source's file content API (``/api/files/<id>/content``).
+        Used to backfill a synced file whose blob is missing from this server's storage. A 404 means
+        the source is missing the blob too; raise ``FileContentNotFound`` so the caller can record it
+        and carry on. Other client errors surface as-is."""
+        try:
+            return self._request(f"/api/files/{file_id}/content").content
+        except requests.HTTPError as exc:
+            if exc.response is not None and exc.response.status_code == 404:
+                raise FileContentNotFound(file_id) from exc
+            raise
+
     def _get(self, path, params=None) -> dict:
+        return self._request(path, params).json()
+
+    def _request(self, path: str, params: dict | None = None) -> requests.Response:
+        """GET ``path`` with the API key header, retrying transient transport errors (timeouts, 5xx,
+        connection resets) with backoff and failing fast on 4xx. Returns the raw response."""
         url = self.base_url + path
         headers = {_API_KEY_HEADER: self.api_key}
         for attempt in range(self.max_retries):
@@ -68,5 +95,5 @@ def _get(self, path, params=None) -> dict:
                 continue
 
             response.raise_for_status()
-            return response.json()
+            return response
         raise RuntimeError("request retries exhausted without a response")  # unreachable with max_retries >= 1

apps/teams/export/importer.py

@@ -5,10 +5,11 @@
 import contextlib
 import copy
 import functools
-from collections.abc import Iterable
+from collections.abc import Callable, Iterable
 
 from django.contrib.auth.models import Group
 from django.contrib.contenttypes.models import ContentType
+from django.core.files.base import ContentFile
 from django.db import models, transaction
 from django.db.models.signals import m2m_changed, post_delete, post_save, pre_delete, pre_save
 from django.utils.dateparse import parse_datetime
@@ -18,6 +19,7 @@
 from apps.teams.utils import set_current_team
 from apps.utils.fields import as_int
 
+from .client import FileContentNotFound
 from .manifest import (
     EXCLUDE_REGISTRY,
     GLOBAL_CONFIG,
@@ -177,12 +179,22 @@ def _match_queue_aggregate(model, row: dict, store: FKTranslationStore, target_t
 
 
 class Importer:
-    def __init__(self, store: FKTranslationStore, private_key=None, on_user_created=None):
+    def __init__(
+        self,
+        store: FKTranslationStore,
+        private_key=None,
+        on_user_created=None,
+        fetch_file_content: Callable[[int], bytes] | None = None,
+    ):
         """``private_key`` unseals secret fields when supplied; ``on_user_created`` is called once
-        per newly created user (e.g. to send an invite)."""
+        per newly created user (e.g. to send an invite); ``fetch_file_content`` is a ``(source_pk) ->
+        bytes`` callable used to backfill a file whose blob is missing from this server's storage.
+        When it's None (the default), a missing blob surfaces as an error instead."""
         self.store = store
         self.private_key = private_key
         self.on_user_created = on_user_created
+        self.fetch_file_content = fetch_file_content
+        self.missing_files: list[str] = []
         # The single team every resource is imported into. Captured from the team row (first in the
         # manifest) and assigned to every team-scoped row, since the per-row team FK isn't exported.
         self.target_team = None
@@ -240,7 +252,7 @@ def _import_team_owned_row(
     ) -> tuple[models.Model, bool]:
         with transaction.atomic():
             field_values, m2m_values, timestamps = self._build_values(model_label, model, row)
-            instance, created = self._get_or_create(model_label, model, source_pk, row, field_values)
+            instance, created = self._create_or_update(model_label, model, source_pk, row, field_values)
 
             for name, target_pks in m2m_values.items():
                 getattr(instance, name).set([pk for pk in target_pks if pk is not None])
@@ -253,15 +265,60 @@ def _import_team_owned_row(
             self.store.record(model_label, source_pk, instance.pk)
         return instance, created
 
+    def _create_or_update(
+        self, model_label: str, model: type[models.Model], source_pk: int, row: dict, field_values: dict
+    ) -> tuple[models.Model, bool]:
+        """Create or update the row. Saving a ``files.file`` whose blob is missing from this server's
+        storage raises FileNotFoundError (File.save reads its size from storage); delegate that to
+        ``_handle_missing_object`` to recover, rather than let it abort the sync."""
+        try:
+            return self._get_or_create(model_label, model, source_pk, row, field_values)
+        except FileNotFoundError as exc:
+            return self._handle_missing_object(exc, model_label, model, source_pk, row, field_values)
+
+    def _handle_missing_object(
+        self,
+        exc: FileNotFoundError,
+        model_label: str,
+        model: type[models.Model],
+        source_pk: int,
+        row: dict,
+        field_values: dict,
+    ) -> tuple[models.Model, bool]:
+        """Recover a row whose backing storage object is missing. Only files are recoverable today:
+        backfill the blob from the source and retry, or on a 404 import the row without content and
+        record it. Anything else re-raises."""
+        fetch = self.fetch_file_content
+        if model_label != "files.file" or fetch is None:
+            raise exc
+        name = field_values["file"]
+        print(f"fetching content for missing file '{name}' from source")
+        try:
+            content = fetch(source_pk)
+        except FileContentNotFound:
+            print(f"file '{name}' has no content on the source; importing without it")
+            self.missing_files.append(name)
+            field_values["file"] = ""
+            return self._get_or_create(model_label, model, source_pk, row, field_values)
+        self._write_file_content(model, name, content)
+        return self._get_or_create(model_label, model, source_pk, row, field_values)
+
+    def _write_file_content(self, model: type[models.Model], name: str, content: bytes) -> None:
+        """Write a backfilled file's bytes to this server's storage at the path the row carries, so
+        the retried save finds the blob."""
+        model._meta.get_field("file").storage.save(name, ContentFile(content))
+
     def _get_or_create(
         self, model_label: str, model: type[models.Model], source_pk: int, row: dict, field_values: dict
     ) -> tuple[models.Model, bool]:
         """Find the existing target row (via the translation map, then a model-specific natural-key
-        match) or create it. Returns (instance, created)."""
+        match) or create it. Returns (instance, created). The lookup goes through ``_base_manager`` so
+        an archived or soft-deleted target row -- hidden by the default manager -- is still matched on
+        re-import rather than duplicated (which would break the unique external_id on channels)."""
         target_pk = self.store.get_target(model_label, source_pk)
         instance = None
-        if target_pk is not None and model.objects.filter(pk=target_pk).exists():
-            instance = model.objects.get(pk=target_pk)
+        if target_pk is not None and model._base_manager.filter(pk=target_pk).exists():
+            instance = model._base_manager.get(pk=target_pk)
         elif model_label in _MATCH_EXISTING:
             instance = _MATCH_EXISTING[model_label](model, row, self.store, self.target_team)
 

apps/teams/export/manifest.py

@@ -231,8 +231,14 @@ def model_has_team_field(model: type[Model]) -> bool:
 
 
 def team_scoped_queryset(entry: ManifestEntry, team) -> QuerySet:
-    """The model's rows for this team, including any shared global rows."""
+    """The model's rows for this team, including any shared global rows and any archived or
+    soft-deleted rows.
+
+    Some default managers hide rows: versioned models filter ``is_archived=False`` and
+    ``ExperimentChannel`` filters ``deleted=False``. ``_base_manager`` is Django's unfiltered manager
+    (docs: Model._base_manager), so those rows are shared across servers along with the live ones."""
     model = entry_model(entry.model)
+    base = model._base_manager
     paths = TEAM_PATH_REGISTRY.get(entry.model, "team")
     if isinstance(paths, str):
         paths = [paths]
@@ -242,7 +248,7 @@ def team_scoped_queryset(entry: ManifestEntry, team) -> QuerySet:
     spec = GLOBAL_CONFIG.get(entry.model)
     if spec:
         team_q |= Q(**{f"{spec.null_field}__isnull": True})
-    queryset = model.objects.filter(team_q)
+    queryset = base.filter(team_q)
     if len(paths) > 1:
         queryset = queryset.distinct()
     extra_filter = EXTRA_FILTERS.get(entry.model)

apps/teams/export/tests/test_client.py

@@ -1,21 +1,22 @@
 import pytest
 import requests
 
-from apps.teams.export.client import ResourceFetcher
+from apps.teams.export.client import FileContentNotFound, ResourceFetcher
 
 
 class FakeResponse:
-    def __init__(self, status_code=200, json_data=None):
+    def __init__(self, status_code=200, json_data=None, content=b""):
         self.status_code = status_code
         self._json = json_data or {}
+        self.content = content
         self.closed = False
 
     def json(self):
         return self._json
 
     def raise_for_status(self):
         if self.status_code >= 400:
-            raise requests.HTTPError(f"{self.status_code}")
+            raise requests.HTTPError(f"{self.status_code}", response=self)
 
     def close(self):
         self.closed = True
@@ -109,3 +110,27 @@ def test_client_error_is_not_retried():
     with pytest.raises(requests.HTTPError):
         _client(session).get_manifest()
     assert len(session.calls) == 1
+
+
+def test_get_file_content_hits_file_content_endpoint_and_returns_bytes():
+    session = FakeSession([FakeResponse(content=b"file-bytes")])
+    assert _client(session).get_file_content(42) == b"file-bytes"
+    call = session.calls[0]
+    assert call["url"] == "https://src.example/api/files/42/content"
+    assert call["headers"]["X-Api-Key"] == "secret-key"
+
+
+def test_get_file_content_missing_file_raises_file_content_not_found():
+    """A 404 means the source is also missing the blob -- raise a domain error (not retried) so the
+    importer can report it and carry on rather than aborting the whole sync."""
+    session = FakeSession([FakeResponse(404)])
+    with pytest.raises(FileContentNotFound):
+        _client(session).get_file_content(99)
+    assert len(session.calls) == 1
+
+
+def test_get_file_content_other_client_error_still_raises_http_error():
+    """A non-404 client error isn't a 'file is gone' signal, so it surfaces as-is."""
+    session = FakeSession([FakeResponse(403)])
+    with pytest.raises(requests.HTTPError):
+        _client(session).get_file_content(99)

apps/teams/export/tests/test_command.py

@@ -55,6 +55,9 @@ def iter_rows(self, resource, start_cursor=None, limit=100):
         self.iter_calls.append((resource, start_cursor))
         return list(self.rows_by_resource.get(resource, []))
 
+    def get_file_content(self, file_id):
+        return b""
+
 
 def _manifest(entries, checksum=None):
     return {"schema_checksum": checksum if checksum is not None else schema_checksum(), "entries": entries}