refactor(files): simplify asset and delta responses

src/govoplan_files/backend/router.py

@@ -991,38 +991,8 @@ def _asset_response(session: Session, asset: FileAsset, *, include_shares: bool
     )
 
 
-def _asset_list_response(session: Session, assets: list[FileAsset], *, include_shares: bool = False) -> list[FileAssetResponse]:
+def _file_share_response(row: FileShare) -> FileShareResponse:
-    if not assets:
+    return FileShareResponse(
-        return []
-
-    asset_ids = [asset.id for asset in assets]
-    version_ids = [asset.current_version_id for asset in assets if asset.current_version_id]
-    if len(version_ids) != len(assets):
-        raise FileStorageError("File has no current version")
-
-    version_blob_rows: list[tuple[FileVersion, FileBlob]] = []
-    for chunk in _chunks(version_ids):
-        version_blob_rows.extend(
-            session.query(FileVersion, FileBlob)
-            .join(FileBlob, FileBlob.id == FileVersion.blob_id)
-            .filter(FileVersion.id.in_(chunk))
-            .all()
-        )
-    versions_by_id = {version.id: version for version, _blob in version_blob_rows}
-    blobs_by_version_id = {version.id: blob for version, blob in version_blob_rows}
-
-    shares_by_asset_id: dict[str, list[FileShareResponse]] = {asset_id: [] for asset_id in asset_ids}
-    if include_shares:
-        for chunk in _chunks(asset_ids):
-            share_rows = (
-                session.query(FileShare)
-                .filter(FileShare.file_asset_id.in_(chunk))
-                .order_by(FileShare.created_at.desc())
-                .all()
-            )
-            for row in share_rows:
-                shares_by_asset_id.setdefault(row.file_asset_id, []).append(
-                    FileShareResponse(
         id=row.id,
         target_type=row.target_type,
         target_id=row.target_id,
@@ -1030,27 +1000,62 @@ def _asset_list_response(session: Session, assets: list[FileAsset], *, include_s
         created_at=row.created_at.isoformat(),
         revoked_at=row.revoked_at.isoformat() if row.revoked_at else None,
     )
-                )
 
+
+def _asset_versions_and_blobs(session: Session, assets: list[FileAsset]) -> dict[str, tuple[FileVersion, FileBlob]]:
+    version_ids = [asset.current_version_id for asset in assets if asset.current_version_id]
+    if len(version_ids) != len(assets):
+        raise FileStorageError("File has no current version")
+    rows: list[tuple[FileVersion, FileBlob]] = []
+    for chunk in _chunks(version_ids):
+        rows.extend(
+            session.query(FileVersion, FileBlob)
+            .join(FileBlob, FileBlob.id == FileVersion.blob_id)
+            .filter(FileVersion.id.in_(chunk))
+            .all()
+        )
+    return {version.id: (version, blob) for version, blob in rows}
+
+
+def _asset_shares_by_id(session: Session, asset_ids: list[str], *, include_shares: bool) -> dict[str, list[FileShareResponse]]:
+    shares_by_asset_id: dict[str, list[FileShareResponse]] = {asset_id: [] for asset_id in asset_ids}
+    if not include_shares:
+        return shares_by_asset_id
+    for chunk in _chunks(asset_ids):
+        rows = (
+            session.query(FileShare)
+            .filter(FileShare.file_asset_id.in_(chunk))
+            .order_by(FileShare.created_at.desc())
+            .all()
+        )
+        for row in rows:
+            shares_by_asset_id.setdefault(row.file_asset_id, []).append(_file_share_response(row))
+    return shares_by_asset_id
+
+
+def _sent_campaign_asset_ids(session: Session, asset_ids: list[str]) -> set[str]:
     sent_asset_ids: set[str] = set()
     for chunk in _chunks(asset_ids):
-        sent_rows = (
+        rows = (
             session.query(CampaignAttachmentUse.file_asset_id)
             .filter(CampaignAttachmentUse.file_asset_id.in_(chunk), CampaignAttachmentUse.use_stage == "sent")
             .distinct()
             .all()
         )
-        sent_asset_ids.update(row[0] for row in sent_rows)
+        sent_asset_ids.update(row[0] for row in rows)
+    return sent_asset_ids
 
-    responses: list[FileAssetResponse] = []
+
-    for asset in assets:
+def _loaded_asset_response(
-        version = versions_by_id.get(asset.current_version_id or "")
+    asset: FileAsset,
-        blob = blobs_by_version_id.get(asset.current_version_id or "")
+    *,
-        if not version or not blob:
+    version: FileVersion,
-            raise FileStorageError("File version not found")
+    blob: FileBlob,
+    shares: list[FileShareResponse],
+    sent_asset_ids: set[str],
+) -> FileAssetResponse:
     metadata = asset.metadata_ or {}
-        responses.append(
+    return FileAssetResponse(
-            FileAssetResponse(
         id=asset.id,
         tenant_id=asset.tenant_id,
         owner_type=asset.owner_type,
@@ -1069,7 +1074,32 @@ def _asset_list_response(session: Session, assets: list[FileAsset], *, include_s
         metadata=metadata,
         source_provenance=source_provenance_from_metadata(metadata),
         source_revision=source_revision_from_metadata(metadata),
+        shares=shares,
+    )
+
+
+def _asset_list_response(session: Session, assets: list[FileAsset], *, include_shares: bool = False) -> list[FileAssetResponse]:
+    if not assets:
+        return []
+
+    asset_ids = [asset.id for asset in assets]
+    versions_and_blobs = _asset_versions_and_blobs(session, assets)
+    shares_by_asset_id = _asset_shares_by_id(session, asset_ids, include_shares=include_shares)
+    sent_asset_ids = _sent_campaign_asset_ids(session, asset_ids)
+
+    responses: list[FileAssetResponse] = []
+    for asset in assets:
+        version_and_blob = versions_and_blobs.get(asset.current_version_id or "")
+        if version_and_blob is None:
+            raise FileStorageError("File version not found")
+        version, blob = version_and_blob
+        responses.append(
+            _loaded_asset_response(
+                asset,
+                version=version,
+                blob=blob,
                 shares=shares_by_asset_id.get(asset.id, []),
+                sent_asset_ids=sent_asset_ids,
             )
         )
     return responses
@@ -1499,6 +1529,104 @@ def _entry_matches_delta_scope(
     )
 
 
+def _decode_files_delta_watermark(since: str) -> int:
+    try:
+        return decode_sequence_watermark(since)
+    except ValueError as exc:
+        raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail=str(exc)) from exc
+
+
+def _changed_delta_resource_ids(entries: list[ChangeSequenceEntry]) -> tuple[list[str], list[str]]:
+    file_ids = list(dict.fromkeys(entry.resource_id for entry in entries if entry.resource_type == "file"))
+    folder_ids = list(dict.fromkeys(entry.resource_id for entry in entries if entry.resource_type == "folder"))
+    return file_ids, folder_ids
+
+
+def _visible_assets_for_delta(
+    session: Session,
+    *,
+    principal: ApiPrincipal,
+    owner_type: Literal["user", "group"] | None,
+    owner_id: str | None,
+    campaign_id: str | None,
+    path_prefix: str | None,
+    changed_file_ids: list[str],
+) -> dict[str, FileAsset]:
+    return {
+        asset.id: asset
+        for asset in list_assets_for_user(
+            session,
+            tenant_id=principal.tenant_id,
+            user_id=principal.user.id,
+            owner_type=owner_type,
+            owner_id=owner_id,
+            campaign_id=campaign_id,
+            path_prefix=path_prefix,
+            is_admin=_is_admin(principal),
+        )
+        if asset.id in changed_file_ids
+    }
+
+
+def _changed_visible_folders_for_delta(
+    session: Session,
+    *,
+    principal: ApiPrincipal,
+    owner_type: Literal["user", "group"] | None,
+    owner_id: str | None,
+    path_prefix: str | None,
+    changed_folder_ids: list[str],
+) -> dict[str, FileFolder]:
+    return {
+        folder.id: folder
+        for folder in _visible_folders_for_delta(
+            session,
+            principal=principal,
+            owner_type=owner_type,
+            owner_id=owner_id,
+            path_prefix=path_prefix,
+        )
+        if folder.id in changed_folder_ids
+    }
+
+
+def _deleted_delta_items(
+    session: Session,
+    *,
+    principal: ApiPrincipal,
+    entries: list[ChangeSequenceEntry],
+    visible_assets: dict[str, FileAsset],
+    visible_folders: dict[str, FileFolder],
+    owner_type: Literal["user", "group"] | None,
+    owner_id: str | None,
+    campaign_id: str | None,
+    path_prefix: str | None,
+) -> list[DeltaDeletedItem]:
+    deleted: dict[tuple[str, str], DeltaDeletedItem] = {}
+    for entry in entries:
+        if entry.resource_type == "file" and entry.resource_id in visible_assets:
+            continue
+        if entry.resource_type == "folder" and entry.resource_id in visible_folders:
+            continue
+        if not _entry_matches_delta_scope(
+            session,
+            principal,
+            entry,
+            owner_type=owner_type,
+            owner_id=owner_id,
+            campaign_id=campaign_id,
+            path_prefix=path_prefix,
+        ):
+            continue
+        deleted[(entry.resource_type, entry.resource_id)] = DeltaDeletedItem(
+            id=entry.resource_id,
+            resource_type=entry.resource_type,
+            revision=encode_sequence_watermark(entry.id),
+            deleted_at=entry.created_at if entry.operation == "deleted" else None,
+        )
+    return list(deleted.values())
+
+
 def _files_delta_response(
     session: Session,
     *,
@@ -1510,10 +1638,7 @@ def _files_delta_response(
     since: str,
     limit: int,
 ) -> FileDeltaResponse:
-    try:
+    since_sequence = _decode_files_delta_watermark(since)
-        since_sequence = decode_sequence_watermark(since)
-    except ValueError as exc:
-        raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail=str(exc)) from exc
     if sequence_watermark_is_expired(
         session,
         since=since_sequence,
@@ -1541,65 +1666,41 @@ def _files_delta_response(
     has_more = len(entries_plus_one) > limit
     entries = entries_plus_one[:limit]
 
-    changed_file_ids = list(dict.fromkeys(entry.resource_id for entry in entries if entry.resource_type == "file"))
+    changed_file_ids, changed_folder_ids = _changed_delta_resource_ids(entries)
-    changed_folder_ids = list(dict.fromkeys(entry.resource_id for entry in entries if entry.resource_type == "folder"))
+    visible_assets = _visible_assets_for_delta(
-
-    visible_assets = {
-        asset.id: asset
-        for asset in list_assets_for_user(
         session,
-            tenant_id=principal.tenant_id,
+        principal=principal,
-            user_id=principal.user.id,
         owner_type=owner_type,
         owner_id=owner_id,
         campaign_id=campaign_id,
         path_prefix=path_prefix,
-            is_admin=_is_admin(principal),
+        changed_file_ids=changed_file_ids,
     )
-        if asset.id in changed_file_ids
+    visible_folders = _changed_visible_folders_for_delta(
-    }
-    visible_folders = {
-        folder.id: folder
-        for folder in _visible_folders_for_delta(
         session,
         principal=principal,
         owner_type=owner_type,
         owner_id=owner_id,
         path_prefix=path_prefix,
+        changed_folder_ids=changed_folder_ids,
     )
-        if folder.id in changed_folder_ids
+    deleted = _deleted_delta_items(
-    }
-
-    deleted: dict[tuple[str, str], DeltaDeletedItem] = {}
-    for entry in entries:
-        is_visible = (
-            (entry.resource_type == "file" and entry.resource_id in visible_assets)
-            or (entry.resource_type == "folder" and entry.resource_id in visible_folders)
-        )
-        if is_visible:
-            continue
-        if not _entry_matches_delta_scope(
         session,
-            principal,
+        principal=principal,
-            entry,
+        entries=entries,
+        visible_assets=visible_assets,
+        visible_folders=visible_folders,
         owner_type=owner_type,
         owner_id=owner_id,
         campaign_id=campaign_id,
         path_prefix=path_prefix,
-        ):
-            continue
-        deleted[(entry.resource_type, entry.resource_id)] = DeltaDeletedItem(
-            id=entry.resource_id,
-            resource_type=entry.resource_type,
-            revision=encode_sequence_watermark(entry.id),
-            deleted_at=entry.created_at if entry.operation == "deleted" else None,
     )
 
     watermark = encode_sequence_watermark(entries[-1].id) if has_more and entries else _files_delta_watermark(session, principal.tenant_id)
     return FileDeltaResponse(
         files=_asset_list_response(session, list(visible_assets.values()), include_shares=True),
         folders=[_folder_response(folder) for folder in visible_folders.values()],
-        deleted=list(deleted.values()),
+        deleted=deleted,
         watermark=watermark,
         has_more=has_more,
         full=False,