junte/Model
1
0
Fork 0
mirror of https://github.com/Hestia-Homes/Model.git synced 2026-07-12 13:29:04 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions

PR review: best-effort read path, true per-member streaming, race-safe 409, local email 🟩

Browse source 
Addresses PR #1498 review:
- A per-file read failure (missing/deleted object, or a bucket the role can't
  reach) is now a SkippedDocument(reason=unreadable), not a whole-run abort;
  an all-unreadable selection still fails (no empty package). Best-effort on
  the read path (ADR-0060).
- Each ZIP member is streamed to a temp file (S3DocumentDownloader.download,
  boto download_file) and added from disk, so a single multi-GB member never
  hits the heap — the 'never held whole in memory' claim is now true.
- The 409 double-submit guard is DB-arbitrated: the sub_task insert is
  conditional on the task having none, so a race creates only one.
- Local env gets a placeholder recipient email so the route is exercisable.
- PackageEntry carries landlord_property_id so a read failure can be reported.
- TODO noting the UploadedFile.s3_upload_timestamp typing fix.
Two new tests: per-file read failure skips-and-reports; address fallback flows
to the folder name.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
This commit is contained in:
Khalim Conn-Kowlessar 2026-07-08 11:08:46 +00:00
parent 0b52a28808
commit 9ec7987e97
10 changed files with 218 additions and 89 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

4
applications/bulk_document_download/handler.py
View file

@ -29,7 +29,7 @@ from infrastructure.email.ses_smtp_email_sender import SesSmtpEmailSender
from infrastructure.postgres.config import PostgresConfig
from infrastructure.postgres.engine import make_engine, make_session
from infrastructure.s3.s3_client import S3Client
from infrastructure.s3.s3_document_bytes_reader import S3DocumentBytesReader
from infrastructure.s3.s3_document_downloader import S3DocumentDownloader
from orchestration.bulk_document_download_orchestrator import (
BulkDocumentDownloadOrchestrator,
)
@ -72,7 +72,7 @@ def handler(body: dict[str, Any], context: Any) -> dict[str, Any]:
orchestrator = BulkDocumentDownloadOrchestrator(
uploaded_files=UploadedFilePostgresRepository(session),
addresses=LandlordAddressResolver(session, landlord_property_ids),
documents=S3DocumentBytesReader(boto_s3),
documents=S3DocumentDownloader(boto_s3),
packages=S3Client(boto_s3, os.environ["DOCUMENT_EXPORTS_BUCKET"]),
email=_email_sender(),
url_ttl_seconds=_URL_TTL_SECONDS,

17
backend/app/documents/download_tasks.py
View file

@ -87,16 +87,24 @@ class DocumentDownloadTasks:
def create_download_subtask(
self, task_id: UUID, subtask_id: UUID, recipe: dict[str, Any]
) -> None:
) -> bool:
"""Pre-create the single ``waiting`` sub_task under the app-owned task,
pinning the recipe (landlord_property_ids, recipient_email,
package_name) onto its inputs — the worker's reproducible instructions
(ADR-0055/0060)."""
(ADR-0055/0060).
The insert is **conditional on the task having no sub_task yet**, so the
DB — not a prior read — arbitrates a double-submit race: two concurrent
requests can both pass the ``already_distributed`` check, but only one
insert lands. Returns whether this call created the sub_task."""
now = datetime.now(timezone.utc)
self._session.connection().execute(
result = self._session.connection().execute(
text(
"INSERT INTO sub_task (id, task_id, status, inputs, updated_at)"
" VALUES (:id, :task_id, 'waiting', :inputs, :updated_at)"
" SELECT :id, :task_id, 'waiting', :inputs, :updated_at"
" WHERE NOT EXISTS ("
" SELECT 1 FROM sub_task WHERE task_id = :task_id"
" )"
),
{
"id": subtask_id,
@ -106,3 +114,4 @@ class DocumentDownloadTasks:
},
)
self._session.commit()
return result.rowcount == 1

17
backend/app/documents/router.py
View file

@ -52,6 +52,11 @@ def get_requesting_user_email(user: Any = Depends(validate_jwt_token)) -> str:
email: Any = getattr(user, "email", None)
if email is None and isinstance(user, dict):
email = cast(dict[str, Any], user).get("email")
if not email and get_settings().ENVIRONMENT == "local":
# Local `get_user` returns a dummy dict with no email; a placeholder lets
# the route be exercised end-to-end on a dev machine (email is a no-op
# locally anyway).
email = "local-dev@example.com"
if not email:
raise HTTPException(
status_code=400,
@ -133,7 +138,17 @@ async def bulk_download(
"recipient_email": recipient_email,
"package_name": f"portfolio-{portfolio_id}-{body.task_id}",
}
tasks.create_download_subtask(body.task_id, subtask_id, recipe)
created = tasks.create_download_subtask(body.task_id, subtask_id, recipe)
if not created:
# Lost a double-submit race — another request already created the
# sub_task. The DB arbitrated; don't enqueue a second job.
raise HTTPException(
status_code=409,
detail=(
f"Task {body.task_id} already has a sub_task — a download has "
"already been started for it."
),
)
send_message(
json.dumps({"task_id": str(body.task_id), "subtask_id": str(subtask_id)})

5
domain/bulk_document_download/package_plan.py
View file

@ -33,11 +33,13 @@ class ResolvedDocument:
@dataclass(frozen=True)
class PackageEntry:
"""One file to place in the Download Package: where it lands in the ZIP and
the S3 object to stream in from."""
the S3 object to stream in from. Carries the property identity so a
read-time failure can be reported as a SkippedDocument (ADR-0060)."""
zip_path: str
s3_bucket: str
s3_key: str
landlord_property_id: str
@dataclass(frozen=True)
@ -95,6 +97,7 @@ def plan_download_package(
f"{document.document_type}{_extension_of(document.s3_key)}",
s3_bucket=document.s3_bucket,
s3_key=document.s3_key,
landlord_property_id=document.landlord_property_id,
)
for document in latest_by_group.values()
]

21
infrastructure/s3/s3_document_bytes_reader.py
View file

@ -1,21 +0,0 @@
"""Reads document bytes from arbitrary S3 buckets (ADR-0060).
Uploaded files live across many source buckets (each `uploaded_files` row
carries its own `s3_file_bucket`), so — unlike the bucket-bound `S3Client` —
this reader takes the bucket per call. Used by the Bulk Document Download
orchestrator to stream each chosen file into the ZIP.
"""
from __future__ import annotations
from typing import Any
class S3DocumentBytesReader:
def __init__(self, boto_s3_client: Any) -> None:
self._client = boto_s3_client
def read(self, bucket: str, key: str) -> bytes:
response: dict[str, Any] = self._client.get_object(Bucket=bucket, Key=key)
body: bytes = response["Body"].read()
return body

20
infrastructure/s3/s3_document_downloader.py Normal file
View file

@ -0,0 +1,20 @@
"""Streams uploaded documents from arbitrary S3 buckets to local disk (ADR-0060).
Uploaded files live across many source buckets (each `uploaded_files` row
carries its own `s3_file_bucket`), so — unlike the bucket-bound `S3Client` —
this takes the bucket per call. Downloading to a file (rather than reading the
object into memory) keeps a single multi-GB member off the heap when it is added
to the on-disk ZIP.
"""
from __future__ import annotations
from typing import Any
class S3DocumentDownloader:
def __init__(self, boto_s3_client: Any) -> None:
self._client = boto_s3_client
def download(self, bucket: str, key: str, dest_path: str) -> None:
self._client.download_file(Bucket=bucket, Key=key, Filename=dest_path)

88
orchestration/bulk_document_download_orchestrator.py
View file

@ -38,8 +38,8 @@ class AddressResolver(Protocol):
def address_for(self, landlord_property_id: str) -> str: ...
class DocumentBytesReader(Protocol):
def read(self, bucket: str, key: str) -> bytes: ...
class DocumentDownloader(Protocol):
def download(self, bucket: str, key: str, dest_path: str) -> None: ...
@dataclass(frozen=True)
@ -58,7 +58,7 @@ class BulkDocumentDownloadOrchestrator:
*,
uploaded_files: UploadedFileReader,
addresses: AddressResolver,
documents: DocumentBytesReader,
documents: DocumentDownloader,
packages: S3Client,
email: EmailSender,
url_ttl_seconds: int,
@ -83,36 +83,44 @@ class BulkDocumentDownloadOrchestrator:
resolved = [self._resolve(file) for file in files]
plan = plan_download_package([r for r in resolved if r is not None])
skipped: list[SkippedDocument] = list(plan.skipped)
if not plan.entries:
# Nothing to package (ADR-0060): a recorded failure, not an empty
# ZIP emailed to the user. The skip detail rides on the failure.
raise SubTaskFailure(
"no documents resolved for the selection",
details={
"skipped": [
{
"landlord_property_id": s.landlord_property_id,
"s3_key": s.s3_key,
"reason": s.reason,
}
for s in plan.skipped
]
},
)
raise self._empty_failure(skipped)
# Stream to a /tmp file and multipart-upload it (ADR-0060): a Download
# Package can be several GB, so it is never held whole in memory — each
# source file is read, written into the on-disk ZIP, and released; the
# finished archive is streamed up from disk.
# Package can be several GB, so it is never held whole in memory. Each
# source file is streamed to its own temp file, added to the on-disk ZIP,
# then deleted; the finished archive is streamed up from disk. Best-effort
# (ADR-0060): a file that can't be read — deleted, or in a bucket the role
# can't reach — becomes a SkippedDocument, it does not fail the package.
key = f"{self._package_key_prefix}/{package_name}.zip"
included = 0
with tempfile.TemporaryDirectory() as tmp_dir:
zip_path = os.path.join(tmp_dir, "package.zip")
with zipfile.ZipFile(zip_path, "w", zipfile.ZIP_DEFLATED) as archive:
for entry in plan.entries:
archive.writestr(
entry.zip_path,
self._documents.read(entry.s3_bucket, entry.s3_key),
)
for index, entry in enumerate(plan.entries):
member_path = os.path.join(tmp_dir, f"member-{index}")
try:
self._documents.download(
entry.s3_bucket, entry.s3_key, member_path
)
except Exception:
skipped.append(
SkippedDocument(
landlord_property_id=entry.landlord_property_id,
s3_key=entry.s3_key,
reason="unreadable",
)
)
continue
archive.write(member_path, arcname=entry.zip_path)
os.remove(member_path)
included += 1
if included == 0:
# Every file that resolved failed to read — no package to send.
raise self._empty_failure(skipped)
self._packages.upload_file(zip_path, key)
url = self._packages.generate_presigned_url(key, self._url_ttl_seconds)
@ -121,7 +129,7 @@ class BulkDocumentDownloadOrchestrator:
subject="Your document download is ready",
body=(
f"Your document download is ready. It contains "
f"{len(plan.entries)} document(s).\n\n"
f"{included} document(s).\n\n"
f"Download it here (link valid for "
f"{self._url_ttl_seconds // 60} minutes):\n{url}"
),
@ -129,8 +137,26 @@ class BulkDocumentDownloadOrchestrator:
return DownloadPackageResult(
presigned_url=url,
package_s3_key=key,
included=len(plan.entries),
skipped=plan.skipped,
included=included,
skipped=tuple(skipped),
)
def _empty_failure(self, skipped: list[SkippedDocument]) -> SubTaskFailure:
"""A recorded failure when the selection yields no readable documents
(ADR-0060) — never an empty ZIP emailed to the user. The skip detail
rides on the failure."""
return SubTaskFailure(
"no documents could be packaged for the selection",
details={
"skipped": [
{
"landlord_property_id": s.landlord_property_id,
"s3_key": s.s3_key,
"reason": s.reason,
}
for s in skipped
]
},
)
def _resolve(self, file: UploadedFile) -> Optional[ResolvedDocument]:
@ -142,5 +168,9 @@ class BulkDocumentDownloadOrchestrator:
document_type=file.file_type,
s3_bucket=file.s3_file_bucket,
s3_key=file.s3_file_key,
# TODO: the cast is only needed because UploadedFile types
# s3_upload_timestamp as `object` (though it is NOT NULL datetime in
# the DB). Type that column `datetime` on the table so callers don't
# each cast — see PR #1498 review.
uploaded_at=cast(datetime, file.s3_upload_timestamp),
)

29
tests/infrastructure/test_s3_document_bytes_reader.py
View file

@ -1,29 +0,0 @@
"""The multi-bucket document reader (ADR-0060) — reads object bytes from
whichever source bucket an uploaded file lives in."""
from collections.abc import Iterator
import pytest
from moto import mock_aws
from infrastructure.s3.s3_document_bytes_reader import S3DocumentBytesReader
from tests.infrastructure import make_boto_client
@pytest.fixture
def reader() -> Iterator[S3DocumentBytesReader]:
with mock_aws():
boto_client = make_boto_client("s3")
boto_client.create_bucket(Bucket="src-a")
boto_client.create_bucket(Bucket="src-b")
boto_client.put_object(Bucket="src-a", Key="surveys/one.pdf", Body=b"AAA")
boto_client.put_object(Bucket="src-b", Key="photos/two.zip", Body=b"BBB")
yield S3DocumentBytesReader(boto_client)
def test_reads_bytes_from_the_named_source_bucket(
reader: S3DocumentBytesReader,
) -> None:
# act / assert — each object is read from its own bucket.
assert reader.read("src-a", "surveys/one.pdf") == b"AAA"
assert reader.read("src-b", "photos/two.zip") == b"BBB"

36
tests/infrastructure/test_s3_document_downloader.py Normal file
View file

@ -0,0 +1,36 @@
"""The multi-bucket document downloader (ADR-0060) — streams object bytes from
whichever source bucket an uploaded file lives in, to local disk."""
from collections.abc import Iterator
from pathlib import Path
import pytest
from moto import mock_aws
from infrastructure.s3.s3_document_downloader import S3DocumentDownloader
from tests.infrastructure import make_boto_client
@pytest.fixture
def downloader() -> Iterator[S3DocumentDownloader]:
with mock_aws():
boto_client = make_boto_client("s3")
boto_client.create_bucket(Bucket="src-a")
boto_client.create_bucket(Bucket="src-b")
boto_client.put_object(Bucket="src-a", Key="surveys/one.pdf", Body=b"AAA")
boto_client.put_object(Bucket="src-b", Key="photos/two.zip", Body=b"BBB")
yield S3DocumentDownloader(boto_client)
def test_downloads_bytes_from_the_named_source_bucket(
downloader: S3DocumentDownloader, tmp_path: Path
) -> None:
# act — each object streams from its own bucket to a local file.
a = tmp_path / "a"
b = tmp_path / "b"
downloader.download("src-a", "surveys/one.pdf", str(a))
downloader.download("src-b", "photos/two.zip", str(b))
# assert
assert a.read_bytes() == b"AAA"
assert b.read_bytes() == b"BBB"

70
tests/orchestration/test_bulk_document_download_orchestrator.py
View file

@ -45,8 +45,11 @@ 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)]
def download(self, bucket: str, key: str, dest_path: str) -> None:
# KeyError for an object we don't hold — stands in for a missing/
# unreadable source object.
with open(dest_path, "wb") as handle:
handle.write(self._blobs[(bucket, key)])
class _RecordingEmail:
@ -176,3 +179,66 @@ def test_packages_the_good_documents_and_reports_the_skipped_ones(
archive = packages.get_object(result.package_s3_key)
with zipfile.ZipFile(io.BytesIO(archive)) as zf:
assert zf.namelist() == ["12 Oak Street (LP1)/site_note.pdf"]
def test_a_file_that_cannot_be_read_is_skipped_not_fatal(packages: S3Client) -> None:
# Arrange — two documents for one property; the photo pack's object is
# missing (download raises), the site note reads fine. Best-effort (ADR-0060):
# the run must still deliver the readable file and report the other.
good = _uploaded_file("LP1", "src-bucket", "surveys/a.pdf")
missing = _uploaded_file(
"LP1", "src-bucket", "surveys/gone.pdf", file_type=FileTypeEnum.PHOTO_PACK
)
email = _RecordingEmail()
orchestrator = BulkDocumentDownloadOrchestrator(
uploaded_files=_FakeUploadedFiles([good, missing]),
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="partial",
)
# Assert — the readable file is delivered; the missing one is reported.
assert result.included == 1
assert [(s.s3_key, s.reason) for s in result.skipped] == [
("surveys/gone.pdf", "unreadable")
]
assert len(email.sent) == 1
archive = packages.get_object(result.package_s3_key)
with zipfile.ZipFile(io.BytesIO(archive)) as zf:
assert zf.namelist() == ["12 Oak Street (LP1)/site_note.pdf"]
def test_the_address_fallback_string_becomes_the_folder_name(
packages: S3Client,
) -> None:
# Arrange — a property whose address resolved to the fallback string.
doc = _uploaded_file("LP9", "src-bucket", "surveys/a.pdf")
orchestrator = BulkDocumentDownloadOrchestrator(
uploaded_files=_FakeUploadedFiles([doc]),
addresses=_FakeAddresses({"LP9": "address unavailable"}),
documents=_FakeDocuments({("src-bucket", "surveys/a.pdf"): b"PDF-BYTES"}),
packages=packages,
email=_RecordingEmail(),
url_ttl_seconds=3600,
)
# Act
result = orchestrator.run(
landlord_property_ids=["LP9"],
recipient_email="user@example.com",
package_name="fallback",
)
# Assert — the fallback flows through to a usable folder name.
archive = packages.get_object(result.package_s3_key)
with zipfile.ZipFile(io.BytesIO(archive)) as zf:
assert zf.namelist() == ["address unavailable (LP9)/site_note.pdf"]