Build, upload and email a Download Package for a property set 🟥

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

orchestration/bulk_document_download_orchestrator.py

@@ -0,0 +1,81 @@
+"""The Bulk Document Download orchestrator (ADR-0060).
+
+Ties the pieces together for one run: gather a property set's uploaded files
+(by `landlord_property_id`), enrich each with its display address, lay them out
+with the Download Package plan, stream the chosen files into a single ZIP,
+upload it, mint a presigned URL, and email the requester. Best-effort — skipped
+rows are reported, not fatal; a wholly-empty selection is a recorded failure
+(ADR-0060).
+"""
+
+from __future__ import annotations
+
+import io
+import zipfile
+from dataclasses import dataclass
+from datetime import datetime
+from typing import Optional, Protocol, Sequence, cast
+
+from domain.bulk_document_download.package_plan import (
+    ResolvedDocument,
+    SkippedDocument,
+    plan_download_package,
+)
+from domain.tasks.subtasks import SubTaskFailure
+from infrastructure.postgres.uploaded_file_table import UploadedFile
+from infrastructure.s3.s3_client import S3Client
+from repositories.email.email_sender import EmailSender
+
+
+class UploadedFileReader(Protocol):
+    def by_landlord_property_ids(
+        self, landlord_property_ids: Sequence[str]
+    ) -> list[UploadedFile]: ...
+
+
+class AddressResolver(Protocol):
+    def address_for(self, landlord_property_id: str) -> str: ...
+
+
+class DocumentBytesReader(Protocol):
+    def read(self, bucket: str, key: str) -> bytes: ...
+
+
+@dataclass(frozen=True)
+class DownloadPackageResult:
+    """What a completed run reports (lands in `sub_task.outputs`, ADR-0060)."""
+
+    presigned_url: str
+    package_s3_key: str
+    included: int
+    skipped: tuple[SkippedDocument, ...]
+
+
+class BulkDocumentDownloadOrchestrator:
+    def __init__(
+        self,
+        *,
+        uploaded_files: UploadedFileReader,
+        addresses: AddressResolver,
+        documents: DocumentBytesReader,
+        packages: S3Client,
+        email: EmailSender,
+        url_ttl_seconds: int,
+        package_key_prefix: str = "bulk-downloads",
+    ) -> None:
+        self._uploaded_files = uploaded_files
+        self._addresses = addresses
+        self._documents = documents
+        self._packages = packages
+        self._email = email
+        self._url_ttl_seconds = url_ttl_seconds
+        self._package_key_prefix = package_key_prefix
+
+    def run(
+        self,
+        *,
+        landlord_property_ids: Sequence[str],
+        recipient_email: str,
+        package_name: str,
+    ) -> DownloadPackageResult:
+        raise NotImplementedError

tests/orchestration/test_bulk_document_download_orchestrator.py

@@ -0,0 +1,113 @@
+"""The Bulk Document Download orchestrator (ADR-0060): gather a property set's
+files, lay them out, zip + upload, presign, and email the requester."""
+
+from __future__ import annotations
+
+import io
+import zipfile
+from collections.abc import Iterator
+from datetime import datetime, timezone
+
+import pytest
+from moto import mock_aws
+
+from infrastructure.postgres.uploaded_file_table import FileTypeEnum, UploadedFile
+from infrastructure.s3.s3_client import S3Client
+from orchestration.bulk_document_download_orchestrator import (
+    BulkDocumentDownloadOrchestrator,
+)
+from tests.infrastructure import make_boto_client
+
+DATA_BUCKET = "data-bucket"
+
+
+class _FakeUploadedFiles:
+    def __init__(self, files: list[UploadedFile]) -> None:
+        self._files = files
+
+    def by_landlord_property_ids(
+        self, landlord_property_ids: object
+    ) -> list[UploadedFile]:
+        wanted = set(landlord_property_ids)  # type: ignore[call-overload]
+        return [f for f in self._files if f.landlord_property_id in wanted]
+
+
+class _FakeAddresses:
+    def __init__(self, mapping: dict[str, str]) -> None:
+        self._mapping = mapping
+
+    def address_for(self, landlord_property_id: str) -> str:
+        return self._mapping[landlord_property_id]
+
+
+class _FakeDocuments:
+    def __init__(self, blobs: dict[tuple[str, str], bytes]) -> None:
+        self._blobs = blobs
+
+    def read(self, bucket: str, key: str) -> bytes:
+        return self._blobs[(bucket, key)]
+
+
+class _RecordingEmail:
+    def __init__(self) -> None:
+        self.sent: list[tuple[str, str, str]] = []
+
+    def send(self, *, to: str, subject: str, body: str) -> None:
+        self.sent.append((to, subject, body))
+
+
+def _uploaded_file(
+    landlord_property_id: str,
+    s3_file_bucket: str,
+    s3_file_key: str,
+    file_type: FileTypeEnum = FileTypeEnum.SITE_NOTE,
+) -> UploadedFile:
+    return UploadedFile(
+        s3_file_bucket=s3_file_bucket,
+        s3_file_key=s3_file_key,
+        s3_upload_timestamp=datetime(2026, 1, 1, tzinfo=timezone.utc),
+        landlord_property_id=landlord_property_id,
+        file_type=file_type.value,
+    )
+
+
+@pytest.fixture
+def packages() -> Iterator[S3Client]:
+    with mock_aws():
+        boto_client = make_boto_client("s3")
+        boto_client.create_bucket(Bucket=DATA_BUCKET)
+        yield S3Client(boto_client, DATA_BUCKET)
+
+
+def test_builds_uploads_and_emails_a_download_package(packages: S3Client) -> None:
+    # Arrange — one property with one site note.
+    files = [_uploaded_file("LP1", "src-bucket", "surveys/a.pdf")]
+    email = _RecordingEmail()
+    orchestrator = BulkDocumentDownloadOrchestrator(
+        uploaded_files=_FakeUploadedFiles(files),
+        addresses=_FakeAddresses({"LP1": "12 Oak Street"}),
+        documents=_FakeDocuments({("src-bucket", "surveys/a.pdf"): b"PDF-BYTES"}),
+        packages=packages,
+        email=email,
+        url_ttl_seconds=3600,
+    )
+
+    # Act
+    result = orchestrator.run(
+        landlord_property_ids=["LP1"],
+        recipient_email="user@example.com",
+        package_name="portfolio-42",
+    )
+
+    # Assert — one file packaged, at the planned path with the source bytes.
+    assert result.included == 1
+    archive = packages.get_object(result.package_s3_key)
+    with zipfile.ZipFile(io.BytesIO(archive)) as zf:
+        assert zf.read("12 Oak Street (LP1)/site_note.pdf") == b"PDF-BYTES"
+
+    # ...the presigned URL points at that object, and the requester is emailed it.
+    assert result.package_s3_key in result.presigned_url
+    assert len(email.sent) == 1
+    to, _subject, body = email.sent[0]
+    assert to == "user@example.com"
+    assert result.presigned_url in body