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 1

Merge pull request #1498 from Hestia-Homes/feature/bulk-document-download

Browse source 
Bulk Document Download: emailed ZIP of a property set's documents (ADR-0059/0060)
This commit is contained in:
KhalimCK 2026-07-08 12:49:46 +01:00 • committed by GitHub
commit bd24a789b9
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
47 changed files with 2252 additions and 4 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

38
.github/workflows/deploy_terraform.yml vendored
View file

@ -538,7 +538,7 @@ jobs:
# Deploy FastAPI Lambda
# ============================================================
fast_api_lambda:
needs: [determine_stage, ara_engine_lambda, categorisation_lambda, postcodeSplitter_lambda, bulk_address2uprn_combiner_lambda, bulkUploadFinaliser_lambda, modelling_e2e_lambda]
needs: [determine_stage, ara_engine_lambda, categorisation_lambda, postcodeSplitter_lambda, bulk_address2uprn_combiner_lambda, bulkUploadFinaliser_lambda, modelling_e2e_lambda, bulk_document_download_lambda]
uses: ./.github/workflows/_deploy_lambda.yml
with:
lambda_name: ara_fast_api
@ -816,6 +816,42 @@ jobs:
TF_VAR_open_epc_api_token: ${{ secrets.DEV_OPEN_EPC_API_TOKEN }}
TF_VAR_google_solar_api_key: ${{ secrets.DEV_GOOGLE_SOLAR_API_KEY }}
# ============================================================
# Build Bulk Document Download image and Push
# ============================================================
bulk_document_download_image:
needs: [determine_stage, shared_terraform]
uses: ./.github/workflows/_build_image.yml
with:
ecr_repo: bulk-document-download-${{ needs.determine_stage.outputs.stage }}
dockerfile_path: applications/bulk_document_download/Dockerfile
build_context: .
secrets:
AWS_ACCESS_KEY_ID: ${{ secrets.DEV_AWS_ACCESS_KEY_ID }}
AWS_SECRET_ACCESS_KEY: ${{ secrets.DEV_AWS_SECRET_ACCESS_KEY }}
AWS_REGION: ${{ secrets.DEV_AWS_REGION }}
# ============================================================
# Deploy Bulk Document Download Lambda
# ============================================================
bulk_document_download_lambda:
needs: [bulk_document_download_image, determine_stage]
uses: ./.github/workflows/_deploy_lambda.yml
with:
lambda_name: bulk_document_download
lambda_path: deployment/terraform/lambda/bulk_document_download
stage: ${{ needs.determine_stage.outputs.stage }}
ecr_repo: bulk-document-download-${{ needs.determine_stage.outputs.stage }}
image_digest: ${{ needs.bulk_document_download_image.outputs.image_digest }}
terraform_apply: ${{ needs.determine_stage.outputs.terraform_apply }}
secrets:
AWS_ACCESS_KEY_ID: ${{ secrets.DEV_AWS_ACCESS_KEY_ID }}
AWS_SECRET_ACCESS_KEY: ${{ secrets.DEV_AWS_SECRET_ACCESS_KEY }}
AWS_REGION: ${{ secrets.DEV_AWS_REGION }}
TF_VAR_db_host: ${{ secrets.DEV_DB_HOST }}
TF_VAR_db_name: ${{ secrets.DEV_DB_NAME }}
TF_VAR_db_port: ${{ secrets.DEV_DB_PORT }}
# ============================================================
# Deploy Hubspot ETL Lambda
# ============================================================

12
CONTEXT.md
View file

@ -99,6 +99,18 @@ _Avoid_: ventilation report, audit report, PAS ventilation (that is the external
An externally-uploaded ventilation survey document produced by a human assessor and ingested from an external source (e.g. Coordination Hub). Recorded in `uploaded_files` with `file_type = PAS_2023_VENTILATION`. Distinct from a **Ventilation Audit**, which is machine-generated from MagicPlan floor plan data.
_Avoid_: ventilation audit (that is the generated output)
**Document Type**:
The category of an uploaded document — the `uploaded_files.file_type` enum (photo_pack, site_note, the pas_2023_* / ecmk_* families, magic_plan_json, mcs_compliance_certificate, ventilation_audit, other, …). Nullable on the row. In a **Download Package** exactly one file per Document Type is included per property — the newest by `s3_upload_timestamp`; a row with a null Document Type is skipped and reported (ADR-0060).
_Avoid_: document category, file kind, doc class
**Download Package**:
The single ZIP archive a **Bulk Document Download** produces: one folder per property (named by human-readable address, enriched from the hubspot deals data, with `landlord_property_id` appended for uniqueness), each folder holding the latest file of each **Document Type** held for that property. Built best-effort — properties/documents that don't resolve are skipped and listed in the run's `sub_task.outputs`, not failed (ADR-0060). Streamed to a dedicated `DOCUMENT_EXPORTS_BUCKET` (never held whole in memory — a package can be several GB); delivered as a 60-minute presigned URL.
_Avoid_: export (that is the plan/scenario XLSX export), bundle, archive (ambiguous), zip (the format, not the concept)
**Bulk Document Download**:
The feature/job that assembles a **Download Package** for a chosen set of properties (the union of one or more whole HubSpot projects, selected by `project_code`, and a hand-picked `property_id` list) and emails the requester a link. Runs on the app-owned-task + attach-mode lane (ADR-0055): the FastAPI route resolves the selection to a distinct `landlord_property_id` set — project codes are expanded via `hubspot_deal_data` (which holds the project↔property grain) — pins that set and the recipient email onto the `sub_task`, and the `applications/bulk_document_download` Lambda builds the package, writes the URL to `sub_task.outputs`, and emails it (ADR-0059/0060). Files are matched by `landlord_property_id` (the missing `property_id` on `uploaded_files` is a known future gap). Selection is capped by property count at the trigger.
_Avoid_: document export, bulk export, file dump
### Source data
**Site Notes**:

35
applications/bulk_document_download/Dockerfile Normal file
View file

@ -0,0 +1,35 @@
FROM public.ecr.aws/lambda/python:3.11
# Postgres host/port/database are baked into the image at build time from the
# deploy workflow's --build-arg values (GitHub Actions DEV_DB_* secrets),
# mirroring the bulk_upload_finaliser Dockerfile. They map onto the POSTGRES_*
# names PostgresConfig.from_env reads. Username/password are NOT baked in --
# Terraform injects those (and the SES SMTP credentials) as Lambda env vars from
# Secrets Manager.
ARG DEV_DB_HOST
ARG DEV_DB_PORT
ARG DEV_DB_NAME
ENV POSTGRES_HOST=${DEV_DB_HOST}
ENV POSTGRES_PORT=${DEV_DB_PORT}
ENV POSTGRES_DATABASE=${DEV_DB_NAME}
WORKDIR /var/task
COPY applications/bulk_document_download/requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt
# DDD-shaped packages only -- no pandas, no legacy backend/.
COPY datatypes/ datatypes/
COPY domain/ domain/
COPY infrastructure/ infrastructure/
COPY orchestration/ orchestration/
COPY repositories/ repositories/
COPY utilities/ utilities/
COPY applications/ applications/
# Place the handler at the Lambda task root so the runtime resolves
# ``main.handler`` without an extra package prefix.
COPY applications/bulk_document_download/handler.py /var/task/main.py
CMD ["main.handler"]

19
applications/bulk_document_download/bulk_document_download_trigger_body.py Normal file
View file

@ -0,0 +1,19 @@
from uuid import UUID
from pydantic import BaseModel, ConfigDict
class BulkDocumentDownloadTriggerBody(BaseModel):
"""SQS trigger for the bulk_document_download Lambda (ADR-0055/0060).
Attach mode: the FastAPI route created the Task + SubTask and pinned the
recipe (``landlord_property_ids``, ``recipient_email``, ``package_name``)
onto the SubTask's ``inputs``. The message therefore carries only the
identifiers — the (potentially large) property set never travels through the
256 KB SQS body.
"""
model_config = ConfigDict(extra="allow")
task_id: UUID
subtask_id: UUID

100
applications/bulk_document_download/handler.py Normal file
View file

@ -0,0 +1,100 @@
"""bulk_document_download Lambda (ADR-0060).
Attach-mode (ADR-0055) handler. The FastAPI route created the app-owned Task +
SubTask and pinned the recipe onto the SubTask's ``inputs``; this reads that
recipe, assembles the Download Package through the orchestrator, and returns the
result (presigned URL + skipped-document summary), which `TaskOrchestrator`
records on ``sub_task.outputs``. A wholly-empty selection raises
``SubTaskFailure`` from the orchestrator — recorded, not retried.
PostgresConfig-only (POSTGRES_*), like the bulk_upload_finaliser Lambda. Source
documents are read from each row's own ``s3_file_bucket``; the ZIP is written to
the dedicated ``DOCUMENT_EXPORTS_BUCKET``. SES SMTP settings arrive as env vars
(terraform bakes the
Secrets Manager IAM-user credentials in).
"""
from __future__ import annotations
import os
from typing import Any
import boto3
from applications.bulk_document_download.bulk_document_download_trigger_body import (
BulkDocumentDownloadTriggerBody,
)
from domain.tasks.tasks import Source
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_downloader import S3DocumentDownloader
from orchestration.bulk_document_download_orchestrator import (
BulkDocumentDownloadOrchestrator,
)
from repositories.property.landlord_address_resolver import LandlordAddressResolver
from repositories.tasks.subtask_postgres_repository import SubTaskPostgresRepository
from repositories.uploaded_file.uploaded_file_postgres_repository import (
UploadedFilePostgresRepository,
)
from utilities.aws_lambda.task_handler import task_handler
# The Download Package link is valid for 60 minutes (ADR-0060).
_URL_TTL_SECONDS = 3600
def _email_sender() -> SesSmtpEmailSender:
return SesSmtpEmailSender(
host=os.environ["SES_SMTP_HOST"],
port=int(os.environ["SES_SMTP_PORT"]),
username=os.environ["SES_SMTP_USERNAME"],
password=os.environ["SES_SMTP_PASSWORD"],
from_address=os.environ["SES_SMTP_FROM_ADDRESS"],
)
@task_handler(task_source="bulk_document_download", source=Source.PORTFOLIO)
def handler(body: dict[str, Any], context: Any) -> dict[str, Any]:
trigger = BulkDocumentDownloadTriggerBody.model_validate(body)
engine = make_engine(PostgresConfig.from_env(os.environ))
session = make_session(engine)
try:
subtask = SubTaskPostgresRepository(session).get(trigger.subtask_id)
recipe: dict[str, Any] = subtask.inputs or {}
landlord_property_ids: list[str] = [
str(x) for x in recipe["landlord_property_ids"]
]
recipient_email: str = str(recipe["recipient_email"])
package_name: str = str(recipe["package_name"])
boto_s3: Any = boto3.client("s3") # pyright: ignore[reportUnknownMemberType]
orchestrator = BulkDocumentDownloadOrchestrator(
uploaded_files=UploadedFilePostgresRepository(session),
addresses=LandlordAddressResolver(session, landlord_property_ids),
documents=S3DocumentDownloader(boto_s3),
packages=S3Client(boto_s3, os.environ["DOCUMENT_EXPORTS_BUCKET"]),
email=_email_sender(),
url_ttl_seconds=_URL_TTL_SECONDS,
)
result = orchestrator.run(
landlord_property_ids=landlord_property_ids,
recipient_email=recipient_email,
package_name=package_name,
)
finally:
session.close()
return {
"presigned_url": result.presigned_url,
"package_s3_key": result.package_s3_key,
"included": result.included,
"skipped": [
{
"landlord_property_id": skip.landlord_property_id,
"s3_key": skip.s3_key,
"reason": skip.reason,
}
for skip in result.skipped
],
}

4
applications/bulk_document_download/requirements.txt Normal file
View file

@ -0,0 +1,4 @@
boto3
pydantic
sqlmodel
psycopg2-binary

1
backend/app/config.py
View file

@ -46,6 +46,7 @@ class Settings(BaseSettings):
LANDLORD_OVERRIDES_SQS_URL: str = "changeme"
FINALISER_SQS_URL: str = "changeme"
MODELLING_E2E_SQS_URL: str = "changeme"
BULK_DOCUMENT_DOWNLOAD_SQS_URL: str = "changeme"
# Third parties
EPC_AUTH_TOKEN: str = "changeme"

0
backend/app/documents/__init__.py Normal file
View file

122
backend/app/documents/download_tasks.py Normal file
View file

@ -0,0 +1,122 @@
"""The bulk-download route's view of the app-owned task's sub_task (ADR-0060).
Parameterised SQL rather than the SQLModel mirrors: importing the
``infrastructure.postgres`` mirror of ``sub_task`` into the FastAPI process
double-registers the table and crashes at import (the same constraint the
Modelling Run distributor works around — see ``modelling/run_tasks.py``).
"""
import json
from datetime import datetime, timezone
from typing import Any, cast
from uuid import UUID
from sqlalchemy import text
from sqlmodel import Session
class DocumentDownloadTasks:
def __init__(self, session: Session) -> None:
self._session = session
def read_selection_config(self, task_id: UUID) -> dict[str, Any]:
"""The task's selection config, read from the FE-owned ``task.inputs``
JSON (ADR-0060) — ``{portfolio_id, property_ids?, select_all?}``. Empty
dict when the task has no inputs."""
row = (
self._session.connection()
.execute(
text("SELECT inputs FROM tasks WHERE id = :task_id"),
{"task_id": task_id},
)
.first()
)
if row is None or row[0] is None:
return {}
parsed: Any = json.loads(row[0])
return cast(dict[str, Any], parsed) if isinstance(parsed, dict) else {}
def already_distributed(self, task_id: UUID) -> bool:
"""Whether the task already has a sub_task — the 409 guard against a
double-submit re-triggering the same download."""
row = (
self._session.connection()
.execute(
text("SELECT 1 FROM sub_task WHERE task_id = :task_id LIMIT 1"),
{"task_id": task_id},
)
.first()
)
return row is not None
def resolve_selection(
self,
project_codes: list[str],
property_ids: list[int],
) -> list[str]:
"""Resolve the selection to the distinct ``landlord_property_id`` set the
Download Package is built from (ADR-0060). Two independent sources,
unioned:
- **project codes** — every property in the named HubSpot projects, read
from ``hubspot_deal_data`` (the project↔property grain lives there, not
on ``property``).
- **hand-picked property ids** — the selected ``property`` rows.
Files are matched by ``landlord_property_id`` (the missing ``property_id``
on ``uploaded_files`` is a known future gap), so a source row with a null
``landlord_property_id`` contributes nothing and is dropped here. The
returned set size is the cap basis (property count, ADR-0060).
"""
landlord_property_ids: set[str] = set()
if project_codes:
rows = self._session.connection().execute(
text(
"SELECT DISTINCT landlord_property_id FROM hubspot_deal_data"
" WHERE project_code = ANY(:codes)"
" AND landlord_property_id IS NOT NULL"
),
{"codes": project_codes},
)
landlord_property_ids.update(row[0] for row in rows.all())
if property_ids:
rows = self._session.connection().execute(
text(
"SELECT landlord_property_id FROM property"
" WHERE id = ANY(:ids) AND landlord_property_id IS NOT NULL"
),
{"ids": property_ids},
)
landlord_property_ids.update(row[0] for row in rows.all())
return sorted(landlord_property_ids)
def create_download_subtask(
self, task_id: UUID, subtask_id: UUID, recipe: dict[str, Any]
) -> 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).
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)
result = self._session.connection().execute(
text(
"INSERT INTO sub_task (id, task_id, status, 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,
"task_id": task_id,
"inputs": json.dumps(recipe),
"updated_at": now,
},
)
self._session.commit()
return result.rowcount == 1

162
backend/app/documents/router.py Normal file
View file

@ -0,0 +1,162 @@
"""Bulk Document Download trigger: POST /v1/documents/bulk-download (ADR-0060).
Resolves the authenticated requester's email, resolves + caps the property
selection, pins the recipe onto a single pre-created sub_task under the
app-owned task, and enqueues one message to the bulk_document_download worker.
Never assembles the package synchronously — the worker emails the link.
"""
import json
from collections.abc import Callable, Iterator
from typing import Any, cast
from uuid import uuid4
import boto3
from fastapi import APIRouter, Depends, HTTPException
from sqlmodel import Session
from backend.app.config import get_settings
from backend.app.db.connection import db_engine
from backend.app.dependencies import validate_jwt_token, validate_token
from backend.app.documents.download_tasks import DocumentDownloadTasks
from backend.app.documents.schemas import BulkDownloadRequest
# The selection cap (ADR-0060) — rejected at the trigger so an oversized request
# never reaches the worker. A conservative, tunable starting value.
MAX_PROPERTIES = 500
MessageSender = Callable[[str], None]
def get_session() -> Iterator[Session]:
with Session(db_engine) as session:
yield session
def get_message_sender() -> MessageSender:
settings = get_settings()
client: Any = cast(Any, boto3.client("sqs", settings.AWS_DEFAULT_REGION)) # pyright: ignore[reportUnknownMemberType]
queue_url = settings.BULK_DOCUMENT_DOWNLOAD_SQS_URL
def send(body: str) -> None:
client.send_message(QueueUrl=queue_url, MessageBody=body)
return send
def get_requesting_user_email(user: Any = Depends(validate_jwt_token)) -> str:
"""The authenticated requester's email — the Download Package recipient.
ADR-0059: the route resolves the user (JWT ``dbId`` -> ``UserModel``) rather
than only gating the token, and threads the email into the job."""
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,
detail="Could not determine the requesting user's email address.",
)
return str(email)
router = APIRouter(
prefix="/documents",
tags=["documents"],
dependencies=[Depends(validate_token)],
)
@router.post("/bulk-download", status_code=202)
async def bulk_download(
body: BulkDownloadRequest,
session: Session = Depends(get_session),
send_message: MessageSender = Depends(get_message_sender),
recipient_email: str = Depends(get_requesting_user_email),
) -> dict[str, str]:
tasks = DocumentDownloadTasks(session)
if tasks.already_distributed(body.task_id):
raise HTTPException(
status_code=409,
detail=(
f"Task {body.task_id} already has a sub_task — a download has "
"already been started for it."
),
)
config = tasks.read_selection_config(body.task_id)
portfolio_id: Any = config.get("portfolio_id")
raw_project_codes: Any = config.get("project_codes") or []
raw_property_ids: Any = config.get("property_ids") or []
if not isinstance(raw_project_codes, list) or not all(
isinstance(code, str) for code in cast(list[Any], raw_project_codes)
):
raise HTTPException(
status_code=400,
detail="task.inputs project_codes must be a list of strings.",
)
if not isinstance(raw_property_ids, list) or not all(
isinstance(pid, int) for pid in cast(list[Any], raw_property_ids)
):
raise HTTPException(
status_code=400,
detail="task.inputs property_ids must be a list of integers.",
)
project_codes: list[str] = cast(list[str], raw_project_codes)
property_ids: list[int] = cast(list[int], raw_property_ids)
if not project_codes and not property_ids:
raise HTTPException(
status_code=400,
detail="task.inputs must provide project_codes or property_ids.",
)
landlord_property_ids = tasks.resolve_selection(project_codes, property_ids)
if not landlord_property_ids:
raise HTTPException(
status_code=400,
detail=(
"The selection resolves to no downloadable properties — none of "
"the selected projects or properties has a landlord_property_id."
),
)
if len(landlord_property_ids) > MAX_PROPERTIES:
raise HTTPException(
status_code=400,
detail=(
f"The selection has {len(landlord_property_ids)} properties, over "
f"the {MAX_PROPERTIES} limit — narrow your selection."
),
)
subtask_id = uuid4()
package_name = (
f"portfolio-{portfolio_id}-{body.task_id}"
if isinstance(portfolio_id, int)
else f"documents-{body.task_id}"
)
recipe = {
"landlord_property_ids": landlord_property_ids,
"recipient_email": recipient_email,
"package_name": package_name,
}
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)})
)
return {"message": "Bulk document download started"}

20
backend/app/documents/schemas.py Normal file
View file

@ -0,0 +1,20 @@
from uuid import UUID
from pydantic import BaseModel
class BulkDownloadRequest(BaseModel):
"""A request to assemble a Download Package (ADR-0060).
The front end creates the app-owned Task first and writes the **selection
config** into ``task.inputs`` (a JSON object of
``{project_codes?: str[], property_ids?: int[], portfolio_id?: int}`` — so a
large selection never travels in an HTTP body), then passes only the
``task_id`` here. The backend reads the selection from ``task.inputs`` and
resolves it to the distinct ``landlord_property_id`` set: the union of every
property in the named HubSpot ``project_codes`` (from ``hubspot_deal_data``)
and the hand-picked ``property_ids``. At least one of the two must be given;
``portfolio_id`` is optional and used only to name the package.
"""
task_id: UUID

2
backend/app/main.py
View file

@ -11,6 +11,7 @@ from backend.app.whlg import router as whlg_router
from backend.app.plan import router as plan_router
from backend.app.tasks import router as tasks_router
from backend.app.bulk_uploads import router as bulk_uploads_router
from backend.app.documents import router as documents_router
from backend.app.modelling import router as modelling_router
from backend.app.dependencies import validate_api_key
from backend.app.config import get_settings
@ -65,6 +66,7 @@ app.include_router(plan_router.router, prefix="/v1")
app.include_router(whlg_router.router, prefix="/v1")
app.include_router(bulk_uploads_router.router, prefix="/v1")
app.include_router(modelling_router.router, prefix="/v1")
app.include_router(documents_router.router, prefix="/v1")
if get_settings().ENVIRONMENT == "local":
from backend.app.local import router as local_router

111
deployment/terraform/lambda/bulk_document_download/main.tf Normal file
View file

@ -0,0 +1,111 @@
############################################
# Credentials from Secrets Manager (resolved at plan/apply time and baked into
# env vars). The SES SMTP credentials are an IAM *user* (see modules/ses), so
# the Lambda authenticates to SES over SMTP with them — the execution role needs
# neither ses:* nor runtime secretsmanager access.
############################################
data "aws_secretsmanager_secret_version" "db_credentials" {
secret_id = "${var.stage}/assessment_model/db_credentials"
}
data "aws_secretsmanager_secret_version" "ses_smtp" {
secret_id = "${var.stage}/ses/smtp_credentials"
}
locals {
db_credentials = jsondecode(data.aws_secretsmanager_secret_version.db_credentials.secret_string)
ses_smtp = jsondecode(data.aws_secretsmanager_secret_version.ses_smtp.secret_string)
document_exports_bucket = "retrofit-document-exports-${var.stage}"
# Source buckets the uploaded_files rows point at (read for packaging).
# OPEN QUESTION: confirm this is the closed set — if files can live in
# arbitrary buckets, widen to ["arn:aws:s3:::*"].
source_bucket_arns = [
"arn:aws:s3:::retrofit-data-${var.stage}",
"arn:aws:s3:::retrofit-energy-assessments-${var.stage}",
]
}
############################################
# Lambda + SQS queue + trigger
############################################
module "lambda" {
source = "../../modules/lambda_with_sqs"
name = var.lambda_name
stage = var.stage
image_uri = local.image_uri
reserved_concurrent_executions = var.reserved_concurrent_executions
batch_size = var.batch_size
maximum_concurrency = var.maximum_concurrency
timeout = 900
memory_size = 3008
# A Download Package can be several GB; it is streamed to /tmp before the
# multipart upload (ADR-0060), so raise ephemeral storage to the 10 GB max.
ephemeral_storage_size = 10240
environment = {
STAGE = var.stage
LOG_LEVEL = "info"
POSTGRES_USERNAME = local.db_credentials.db_assessment_model_username
POSTGRES_PASSWORD = local.db_credentials.db_assessment_model_password
POSTGRES_HOST = var.db_host
POSTGRES_DATABASE = var.db_name
POSTGRES_PORT = var.db_port
DOCUMENT_EXPORTS_BUCKET = local.document_exports_bucket
# SES SMTP — IAM-user credentials sourced from Secrets Manager (modules/ses).
SES_SMTP_HOST = "email-smtp.eu-west-2.amazonaws.com"
SES_SMTP_PORT = "587"
SES_SMTP_USERNAME = local.ses_smtp.username
SES_SMTP_PASSWORD = local.ses_smtp.password
SES_SMTP_FROM_ADDRESS = var.ses_from_address
}
}
############################################
# IAM: read uploaded documents from their source buckets
############################################
module "s3_read" {
source = "../../modules/s3_iam_policy"
policy_name = "BulkDocumentDownloadS3Read-${var.stage}"
policy_description = "Allow bulk_document_download Lambda to read uploaded_files objects from their source buckets"
bucket_arns = local.source_bucket_arns
actions = ["s3:GetObject", "s3:ListBucket"]
resource_paths = ["/*"]
}
resource "aws_iam_role_policy_attachment" "s3_read" {
role = module.lambda.role_name
policy_arn = module.s3_read.policy_arn
}
############################################
# IAM: write + read the assembled ZIP on the dedicated exports bucket.
# GetObject is required so the Lambda-signed presigned GET URL resolves.
############################################
module "s3_exports" {
source = "../../modules/s3_iam_policy"
policy_name = "BulkDocumentDownloadS3Exports-${var.stage}"
policy_description = "Allow bulk_document_download Lambda to write and presign-read Download Packages on the exports bucket"
bucket_arns = ["arn:aws:s3:::${local.document_exports_bucket}"]
actions = ["s3:PutObject", "s3:GetObject"]
resource_paths = ["/*"]
}
resource "aws_iam_role_policy_attachment" "s3_exports" {
role = module.lambda.role_name
policy_arn = module.s3_exports.policy_arn
}
# NOTE: no ses:* on the role — the handler sends over SMTP with the SES IAM-user
# credentials injected above.

9
deployment/terraform/lambda/bulk_document_download/outputs.tf Normal file
View file

@ -0,0 +1,9 @@
output "bulk_document_download_queue_url" {
value = module.lambda.queue_url
description = "URL of the bulk-document-download SQS queue (FastAPI enqueues jobs here)"
}
output "bulk_document_download_queue_arn" {
value = module.lambda.queue_arn
description = "ARN of the bulk-document-download SQS queue"
}

20
deployment/terraform/lambda/bulk_document_download/provider.tf Normal file
View file

@ -0,0 +1,20 @@
terraform {
required_providers {
aws = {
source = "hashicorp/aws"
version = ">= 5.0"
}
}
backend "s3" {
bucket = "bulk-document-download-terraform-state"
key = "terraform.tfstate"
region = "eu-west-2"
}
required_version = ">= 1.2.0"
}
provider "aws" {
region = "eu-west-2"
}

61
deployment/terraform/lambda/bulk_document_download/variables.tf Normal file
View file

@ -0,0 +1,61 @@
variable "lambda_name" {
type = string
description = "Logical name of the lambda"
}
variable "stage" {
description = "Deployment stage (e.g. dev, prod)"
type = string
}
variable "ecr_repo_url" {
type = string
description = "ECR repository URL (no tag, no digest)"
}
variable "image_digest" {
type = string
description = "Image digest (sha256:...)"
}
variable "reserved_concurrent_executions" {
type = number
default = -1
description = "Reserved concurrency for the Lambda. -1 = unreserved."
}
variable "maximum_concurrency" {
type = number
default = 5
description = "Maximum concurrent Lambda invocations from the SQS trigger."
}
variable "batch_size" {
type = number
default = 1
}
variable "db_host" {
type = string
sensitive = true
}
variable "db_name" {
type = string
sensitive = true
}
variable "db_port" {
type = string
sensitive = true
}
variable "ses_from_address" {
type = string
description = "Verified SES sender for Download Package notifications (ADR-0059)."
default = "noreply@domna.homes"
}
locals {
image_uri = "${var.ecr_repo_url}@${var.image_digest}"
}

14
deployment/terraform/lambda/fast-api/main.tf
View file

@ -73,6 +73,15 @@ data "terraform_remote_state" "modelling_e2e" {
}
}
data "terraform_remote_state" "bulk_document_download" {
backend = "s3"
config = {
bucket = "bulk-document-download-terraform-state",
key = "env:/${var.stage}/terraform.tfstate"
region = "eu-west-2"
}
}
############################################
# Load Credentials
############################################
@ -136,6 +145,8 @@ module "fastapi" {
FINALISER_SQS_URL = data.terraform_remote_state.bulk_upload_finaliser.outputs.bulk_upload_finaliser_queue_url
LANDLORD_OVERRIDES_SQS_URL = data.terraform_remote_state.landlord_description_overrides.outputs.landlord_description_overrides_queue_url
MODELLING_E2E_SQS_URL = data.terraform_remote_state.modelling_e2e.outputs.modelling_e2e_queue_url
BULK_DOCUMENT_DOWNLOAD_SQS_URL = data.terraform_remote_state.bulk_document_download.outputs.bulk_document_download_queue_url
}
}
@ -160,7 +171,8 @@ module "fastapi_sqs_policy" {
data.terraform_remote_state.bulk_address2uprn_combiner.outputs.bulk_address2uprn_combiner_queue_arn,
data.terraform_remote_state.bulk_upload_finaliser.outputs.bulk_upload_finaliser_queue_arn,
data.terraform_remote_state.landlord_description_overrides.outputs.landlord_description_overrides_queue_arn,
data.terraform_remote_state.modelling_e2e.outputs.modelling_e2e_queue_arn
data.terraform_remote_state.modelling_e2e.outputs.modelling_e2e_queue_arn,
data.terraform_remote_state.bulk_document_download.outputs.bulk_document_download_queue_arn
]
conditions = null

4
deployment/terraform/modules/lambda_service/main.tf
View file

@ -11,6 +11,10 @@ resource "aws_lambda_function" "this" {
reserved_concurrent_executions = var.reserved_concurrent_executions
ephemeral_storage {
size = var.ephemeral_storage_size
}
environment {
variables = var.environment
}

6
deployment/terraform/modules/lambda_service/variables.tf
View file

@ -12,6 +12,12 @@ variable "memory_size" {
default = 512
}
variable "ephemeral_storage_size" {
type = number
default = 512
description = "Lambda /tmp size in MB (512-10240). Default 512 keeps every existing Lambda unchanged; raised only where a large on-disk artifact is built (e.g. multi-GB Download Package ZIPs, ADR-0060)."
}
variable "environment" {
type = map(string)
default = {}

2
deployment/terraform/modules/lambda_with_sqs/main.tf
View file

@ -31,6 +31,8 @@ module "lambda" {
timeout = var.timeout
memory_size = var.memory_size
ephemeral_storage_size = var.ephemeral_storage_size
environment = var.environment
reserved_concurrent_executions = var.reserved_concurrent_executions
}

5
deployment/terraform/modules/lambda_with_sqs/variables.tf
View file

@ -25,6 +25,11 @@ variable "memory_size" {
default = 1024
}
variable "ephemeral_storage_size" {
type = number
default = 512
}
variable "environment" {
type = map(string)
default = {}

32
deployment/terraform/shared/main.tf
View file

@ -895,6 +895,38 @@ output "modelling_e2e_ecr_url" {
}
################################################
# Bulk Document Download – Lambda (ADR-0060)
################################################
module "bulk_document_download_state_bucket" {
source = "../modules/tf_state_bucket"
bucket_name = "bulk-document-download-terraform-state"
}
module "bulk_document_download_registry" {
source = "../modules/container_registry"
name = "bulk-document-download"
stage = var.stage
}
# Dedicated bucket for generated Download Packages — kept out of the shared data
# bucket so retention/IAM stay scoped to exports (ADR-0060).
module "document_exports" {
source = "../modules/s3"
bucketname = "retrofit-document-exports-${var.stage}"
allowed_origins = var.allowed_origins
}
output "document_exports_bucket_name" {
value = module.document_exports.bucket_name
description = "Name of the document exports bucket (Download Packages)"
}
output "bulk_document_download_ecr_url" {
value = module.bulk_document_download_registry.repository_url
}
################################################
# Abri OpenHousing – Lambda
################################################

79
docs/adr/0059-outbound-email-via-ses-smtp-adapter.md Normal file
View file

@ -0,0 +1,79 @@
---
status: accepted
---
# Outbound email is sent from the backend via an SES-SMTP adapter, with the recipient threaded from the trigger route
The Bulk Document Download feature (ADR-0060) must email a completed package's
link to the user who requested it. This is the **first outbound email in the
Python backend** — there is no `smtplib`/SES/SendGrid code anywhere today. Two
facts force a decision rather than a copy:
1. **Where the email is sent from.** The front end can send email, but the
package is produced **asynchronously in a Lambda**, long after the trigger
request has returned `202`. The completion event exists only on the backend.
2. **Who to email.** Auth today only *gates* the FastAPI routes
(`validate_token` returns the raw token, not the user); neither the handler
nor the `Task`/`SubTask` model captures a user. The requesting identity is
provable (the NextAuth JWT carries `dbId` → `UserModel.email`) but is not
currently threaded anywhere.
## Decision
**The backend sends the email, through a DDD port + SES-SMTP adapter, to a
recipient the trigger route resolves and pins into `tasks.inputs`.**
- **Send from the backend, on completion.** The job that finishes the work
owns the notification — the orchestrator calls an injected email port after
the package is uploaded and the URL minted. The front end is not asked to
poll-then-send.
- **Port + adapter (hexagonal).** A `repositories/email/` port
(`EmailSender` — `send(to, subject, body/html)`) with an
`infrastructure/email/ses_smtp_email_sender.py` adapter over `smtplib`, using
the **existing SES SMTP** setup (terraform `modules/ses`, domain
`domna.homes`, credentials already in Secrets Manager
`${stage}/ses/smtp_credentials`). SMTP, not the SES SendEmail API, because
that is what the provisioned infrastructure exposes. The port is injected into
the orchestrator so the send is testable behind an interface and the transport
is swappable.
- **Recipient threaded at the route, pinned in `inputs`.** The FastAPI trigger
route resolves the authenticated user (`dbId` → `UserModel.email`, via the
existing `get_user`) and writes the email into the `tasks.inputs` JSON at task
creation. The Lambda reads it from `inputs` and never touches auth. This
graduates the route from *gating* the token to *injecting* the user — a
deliberate, contained change to `backend/app/dependencies.py` usage at the one
new route.
- **Config follows the Lambda convention.** SMTP host/username/password reach
the Lambda as environment variables (sourced from the Secrets Manager entry in
terraform), read via `os.environ`/`PostgresConfig`-style config — not
`backend/app/config.py get_settings()`, which is the legacy app boundary.
## Considered options
- **Front end sends the email.** Rejected: the completion moment lives in the
Lambda; making the FE responsible means it must poll task status to
completion and then send — more moving parts, and the notification would lag
or be missed if the user closes the tab. Backend-sends keeps the trigger and
the notification on the same side of the queue.
- **SES SendEmail API (boto3 `ses`) instead of SMTP.** Rejected for now: the
provisioned infrastructure issues SMTP credentials; using the API would need
new IAM + identity wiring. Revisit if a templated/bulk API becomes worthwhile.
- **Resolve the recipient in the Lambda from `portfolio_id` → `PortfolioUsers`.**
Rejected: a portfolio has many users, so the join cannot identify *the
requester* — real risk of emailing the wrong person or several. The requester
is only unambiguous at the authenticated route.
- **FE passes the recipient email in the trigger body.** Rejected: trusts the
client for an identity the JWT already proves, and can drift from the
logged-in user.
## Consequences
- A reusable `EmailSender` capability now exists for future backend
notifications (not just this feature).
- The one new FastAPI route must inject the user (not merely gate) and persist
the email in `inputs`; `Task`/`SubTask` still gain no user column — the
recipient lives in the request payload, scoped to the job.
- Email content is plain for v1 (a link, a 60-minute expiry note, and the
skipped-document summary from `sub_task.outputs`); templating can follow.
- A new terraform env-var surface (SMTP creds from the existing Secrets Manager
entry) is added to the Lambda; no new SES infrastructure is required.

118
docs/adr/0060-bulk-document-download-package-model.md Normal file
View file

@ -0,0 +1,118 @@
---
status: accepted (builds on ADR-0055, ADR-0059)
---
# Bulk Document Download builds one capped, best-effort Download Package per request
Users need to pull the documents held in `uploaded_files` for many properties
at once — per property, the latest file of each **Document Type** — as a single
archive, without clicking through the UI file by file. The request is initiated
in the front end, can span a hand-picked set of properties or one or more whole
HubSpot projects (by project code), and finishes with a link emailed to the
requester (ADR-0059).
`uploaded_files` links to a property by `uprn` **or** `landlord_property_id`
(both nullable) and has **no `property_id`** column; `file_type` (the Document
Type) is itself nullable; and files live across arbitrary buckets
(`s3_file_bucket` per row).
## Decision
**A request produces exactly one Download Package: a ZIP of one folder per
property, each holding the latest file of each Document Type — built
best-effort, size-capped at the trigger, on the app-owned-task + attach-mode
lane (ADR-0055).**
- **Trigger & lifecycle (ADR-0055).** The **front end** creates the app-owned
`tasks` row and writes the **selection config** —
`{project_codes?: str[], property_ids?: int[], portfolio_id?: int}` — into
`tasks.inputs` (TEXT/JSON, FE-owned), then calls the FastAPI route with **only
the `task_id`**. A large selection therefore never travels in an HTTP body.
The route **reads the selection from `tasks.inputs`** and resolves it to the
distinct `landlord_property_id` set — the **union** of every property in the
named HubSpot `project_codes` (read from `hubspot_deal_data`, where the
project↔property grain lives — `property` carries only `portfolio_id`) and the
hand-picked `property_ids`. At least one of the two must be provided;
`portfolio_id` is optional and used only to name the package. The route then
caps the set, resolves the recipient email from the authenticated user
(ADR-0059), **pins the resolved recipe**
(`landlord_property_ids`, `recipient_email`, `package_name`) onto **one
pre-created `sub_task`'s `inputs`**, and drops one SQS message (`task_id`,
`sub_task_id`). The new `applications/bulk_document_download` Lambda runs in
**attach mode**, reading that recipe from the sub_task; `TaskOrchestrator`
owns status + roll-up. (`task.inputs` = the FE's selection; `sub_task.inputs`
= the backend's resolved recipe.)
- **One package = one sub_task, streamed.** No fan-out. A Download Package can
be **several GB**, so it is never held whole in memory: each document is read
from its own `s3_file_bucket`, written into an on-disk ZIP in `/tmp`, and
released; the finished archive is **multipart-uploaded from disk**
(`S3Client.upload_file`). The Lambda's ephemeral storage (`/tmp`) is raised
accordingly (up to 10 GB) — see Consequences. One artifact, one URL, one
email.
- **Selection cap at the route.** Reject `> N` properties synchronously with a
legible "narrow your selection" error (property count, not a document/byte
query — cheap, no S3 on the hot path). A `/tmp` size budget inside the Lambda
is a backstop that fails the sub_task with a clear reason if a pathological
selection still overflows. `N` starts at a conservative, tunable value.
- **Matching on `landlord_property_id`.** Files are gathered by
`landlord_property_id`; the missing `property_id` on `uploaded_files` is a
**known future gap**, out of scope here. Property display info (the folder
name) is enriched from the hubspot deals data
(`repositories/hubspot_deals/`).
- **Layout.** One folder per property named by human-readable **address**
(hubspot-enriched), with `landlord_property_id` appended for uniqueness;
inside, one file per Document Type = the newest by `s3_upload_timestamp`.
Rows with a **null Document Type are skipped** and listed in the run output.
- **Best-effort failure contract.** Build from whatever resolves. Skipped
properties (no documents) and skipped null-type files are recorded in
`sub_task.outputs`. The run **fails only** on an infrastructure error
(S3/DB/zip/upload/email) or when the whole selection yields **zero
documents**. The email carries an "N properties, M documents, X skipped"
summary.
- **Delivery — both channels.** The presigned URL (60-minute expiry) and the
skip-summary are written to `sub_task.outputs` (the FE already polls task
status and can show the link) **and** emailed (ADR-0059).
- **A dedicated exports bucket.** Packages are written to a **separate
`DOCUMENT_EXPORTS_BUCKET`**, not `DATA_BUCKET` — the generated archives are a
distinct class of artifact (transient, user-facing, wide-read source but
single-writer) and keeping them out of the shared data bucket keeps IAM and
any future retention policy cleanly scoped to exports. No lifecycle rule is
imposed on `DATA_BUCKET`. Retention on the exports bucket is an open decision
(the packages are transient, so an expiry is likely warranted, but it is left
to a follow-up rather than baked in here).
## Considered options
- **Fan out into per-batch sub_tasks / multiple ZIPs** (like the modelling
run). Rejected: the user would receive N links / N emails and "the download"
would stop being one file; the capped-single-package model keeps the artifact
and the notification singular.
- **Cap on resolved document count / bytes.** Rejected for v1: it adds
synchronous DB + S3 HEAD work to the trigger and couples the route to the
packaging logic; a property-count cap is cheap and predictable.
- **Strict "all-or-nothing" packaging.** Rejected: one unreadable row or a
property with no documents would deny the entire package; best-effort +
reporting matches "include documents and properties where they exist".
- **Match on `uprn`.** Deferred: `landlord_property_id` is the agreed key and
hubspot enrichment is keyed to the deal/property; revisit if/when the
`property_id` gap on `uploaded_files` is closed.
## Consequences
- New DDD pieces: `applications/bulk_document_download/` (thin handler +
trigger body), `orchestration/bulk_document_download_orchestrator.py`,
packaging rules in `domain/`, a `landlord_property_id`-keyed
"latest-per-Document-Type" query on the uploaded-file repository, a
`generate_presigned_url` + multipart upload on `S3Client`, and the email
port/adapter from ADR-0059. The only `backend/` touch is the trigger route.
- The stored, pinned `property_id` set makes a run reproducible even if the
portfolio changes between trigger and execution.
- `N`, the Lambda's ephemeral-storage (`/tmp`) size, and its memory are tunable
knobs, not load-bearing invariants. `/tmp` must be raised beyond the 512 MB
default (up to 10 GB) to hold a multi-GB archive — a new terraform knob on the
shared Lambda modules, backward-compatible (every other Lambda keeps 512 MB).
- A **new S3 bucket** (`DOCUMENT_EXPORTS_BUCKET`) is provisioned for the
packages, with its own IAM (single-writer, presign-read). Its retention policy
is a deliberate follow-up, not set here.
- Closing the `uploaded_files.property_id` gap later would let matching move off
`landlord_property_id` without changing the package model.

0
domain/bulk_document_download/__init__.py Normal file
View file

104
domain/bulk_document_download/package_plan.py Normal file
View file

@ -0,0 +1,104 @@
"""The Download Package plan (ADR-0060).
Pure logic: given the documents resolved for a set of properties, decide what
goes where in the ZIP — one folder per property, the latest file of each
Document Type — and which rows are skipped and reported. No I/O: the
orchestrator does the S3/DB work; this module only lays out the archive.
"""
from __future__ import annotations
import posixpath
from dataclasses import dataclass
from datetime import datetime
from typing import Optional, Sequence
@dataclass(frozen=True)
class ResolvedDocument:
"""One `uploaded_files` row resolved for packaging: its property identity
(matched by `landlord_property_id`, ADR-0060), the display address enriched
from the hubspot deals data, its Document Type (`file_type`, nullable), the
S3 location to read it from, and when it was uploaded (for latest-per-type).
"""
landlord_property_id: str
address: str
document_type: Optional[str]
s3_bucket: str
s3_key: str
uploaded_at: datetime
@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. 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)
class SkippedDocument:
"""A resolved row left out of the package, with the reason (surfaced in the
run's `sub_task.outputs`, ADR-0060)."""
landlord_property_id: str
s3_key: str
reason: str
@dataclass(frozen=True)
class DownloadPackagePlan:
"""The laid-out archive: the files to include and the rows skipped."""
entries: tuple[PackageEntry, ...]
skipped: tuple[SkippedDocument, ...]
def _extension_of(s3_key: str) -> str:
"""The file extension of an S3 key (``.pdf``, ``.zip``, …), empty when the
key's basename has none. Used to keep the packaged file openable while
naming it by Document Type."""
_root, ext = posixpath.splitext(posixpath.basename(s3_key))
return ext
def plan_download_package(
documents: Sequence[ResolvedDocument],
) -> DownloadPackagePlan:
"""Lay out a Download Package from resolved documents (ADR-0060): one folder
per property, the latest file of each Document Type, null-Document-Type rows
skipped and reported."""
latest_by_group: dict[tuple[str, str], ResolvedDocument] = {}
skipped: list[SkippedDocument] = []
for document in documents:
if document.document_type is None:
skipped.append(
SkippedDocument(
landlord_property_id=document.landlord_property_id,
s3_key=document.s3_key,
reason="null_document_type",
)
)
continue
group = (document.landlord_property_id, document.document_type)
incumbent = latest_by_group.get(group)
if incumbent is None or document.uploaded_at > incumbent.uploaded_at:
latest_by_group[group] = document
entries: list[PackageEntry] = [
PackageEntry(
zip_path=f"{document.address} ({document.landlord_property_id})/"
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()
]
return DownloadPackagePlan(entries=tuple(entries), skipped=tuple(skipped))

0
infrastructure/email/__init__.py Normal file
View file

71
infrastructure/email/ses_smtp_email_sender.py Normal file
View file

@ -0,0 +1,71 @@
"""SES-over-SMTP implementation of the outbound-email port (ADR-0059).
Sends via `smtplib` against the SES SMTP endpoint using the credentials
provisioned in terraform (`modules/ses`, Secrets Manager
`${stage}/ses/smtp_credentials`). The SMTP transport is injected so the sender
is testable without a live server.
"""
from __future__ import annotations
import smtplib
from email.message import EmailMessage
from types import TracebackType
from typing import Any, Callable, Optional, Protocol, Type, cast
class SmtpTransport(Protocol):
"""The minimal SMTP surface the sender uses — satisfied structurally by
`smtplib.SMTP` and by test fakes."""
def __enter__(self) -> "SmtpTransport": ...
def __exit__(
self,
exc_type: Optional[Type[BaseException]],
exc: Optional[BaseException],
tb: Optional[TracebackType],
) -> Optional[bool]: ...
def starttls(self) -> Any: ...
def login(self, user: str, password: str) -> Any: ...
def send_message(self, message: EmailMessage) -> Any: ...
def _default_smtp(host: str, port: int) -> SmtpTransport:
# `smtplib.SMTP` satisfies SmtpTransport structurally; its stdlib stubs
# declare wider signatures (extra optional args, an SMTP-typed context
# manager), so the cast bridges the real transport to the port's surface.
return cast(SmtpTransport, smtplib.SMTP(host, port))
class SesSmtpEmailSender:
def __init__(
self,
*,
host: str,
port: int,
username: str,
password: str,
from_address: str,
smtp_factory: Callable[[str, int], SmtpTransport] = _default_smtp,
) -> None:
self._host = host
self._port = port
self._username = username
self._password = password
self._from_address = from_address
self._smtp_factory = smtp_factory
def send(self, *, to: str, subject: str, body: str) -> None:
message = EmailMessage()
message["From"] = self._from_address
message["To"] = to
message["Subject"] = subject
message.set_content(body)
with self._smtp_factory(self._host, self._port) as smtp:
smtp.starttls()
smtp.login(self._username, self._password)
smtp.send_message(message)

7
infrastructure/postgres/task_table.py
View file

@ -2,7 +2,7 @@ from datetime import datetime, timezone
from typing import ClassVar, Optional
from uuid import UUID, uuid4
from sqlalchemy import Column
from sqlalchemy import Column, Text
from sqlalchemy import Enum as SAEnum
from sqlmodel import Field, SQLModel
@ -18,6 +18,11 @@ class TaskRow(SQLModel, table=True):
job_completed: Optional[datetime] = None
status: str = Field(default="waiting")
service: Optional[str] = None
# FE-owned (Drizzle) TEXT column holding the task's request as a JSON
# string. Declared here so the backend can read it (e.g. the Bulk Document
# Download route reads its property selection from it, ADR-0060) and so the
# test schema builds the column; prod owns the real column via FE migrations.
inputs: Optional[str] = Field(default=None, sa_column=Column(Text, nullable=True))
updated_at: datetime = Field(
default_factory=lambda: datetime.now(timezone.utc)
)

18
infrastructure/s3/s3_client.py
View file

@ -20,3 +20,21 @@ class S3Client:
def put_object(self, key: str, body: bytes) -> str:
self._client.put_object(Bucket=self._bucket, Key=key, Body=body)
return f"s3://{self._bucket}/{key}"
def upload_file(self, local_path: str, key: str) -> str:
"""Upload a file from local disk, using boto's managed transfer so a
large object (a multi-GB Download Package ZIP, ADR-0060) is streamed in
multipart from ``/tmp`` rather than held in memory."""
self._client.upload_file(Filename=local_path, Bucket=self._bucket, Key=key)
return f"s3://{self._bucket}/{key}"
def generate_presigned_url(self, key: str, expires_in: int) -> str:
"""A time-limited URL that lets the holder GET this object without AWS
credentials (ADR-0060 — how a Download Package link is delivered).
``expires_in`` is the validity window in seconds."""
url: str = self._client.generate_presigned_url(
"get_object",
Params={"Bucket": self._bucket, "Key": key},
ExpiresIn=expires_in,
)
return url

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)

176
orchestration/bulk_document_download_orchestrator.py Normal file
View file

@ -0,0 +1,176 @@
"""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 os
import tempfile
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 DocumentDownloader(Protocol):
def download(self, bucket: str, key: str, dest_path: str) -> None: ...
@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: DocumentDownloader,
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:
files = self._uploaded_files.by_landlord_property_ids(landlord_property_ids)
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:
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 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 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)
self._email.send(
to=recipient_email,
subject="Your document download is ready",
body=(
f"Your document download is ready. It contains "
f"{included} document(s).\n\n"
f"Download it here (link valid for "
f"{self._url_ttl_seconds // 60} minutes):\n{url}"
),
)
return DownloadPackageResult(
presigned_url=url,
package_s3_key=key,
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]:
if file.landlord_property_id is None:
return None
return ResolvedDocument(
landlord_property_id=file.landlord_property_id,
address=self._addresses.address_for(file.landlord_property_id),
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),
)

0
repositories/email/__init__.py Normal file
View file

15
repositories/email/email_sender.py Normal file
View file

@ -0,0 +1,15 @@
"""The outbound-email port (ADR-0059).
A narrow interface the orchestrator depends on to notify a user, so the
transport (SES over SMTP today) stays swappable and testable behind it.
"""
from __future__ import annotations
from typing import Protocol
class EmailSender(Protocol):
def send(self, *, to: str, subject: str, body: str) -> None:
"""Send a plain-text email to ``to``."""
...

37
repositories/property/landlord_address_resolver.py Normal file
View file

@ -0,0 +1,37 @@
"""Resolves a property's display address from its `landlord_property_id`
(ADR-0060), for naming the folders in a Download Package.
The address lives on the FE-owned `property` table (`uploaded_files` carries no
`property_id` — a known future gap). Addresses for the whole requested set are
loaded once at construction; `address_for` is then a lookup with a safe
fallback for a `landlord_property_id` that has no property row or address.
"""
from __future__ import annotations
from typing import Sequence
from sqlalchemy import select
from sqlmodel import Session, col
from infrastructure.postgres.property_table import PropertyRow
class LandlordAddressResolver:
_FALLBACK = "address unavailable"
def __init__(
self, session: Session, landlord_property_ids: Sequence[str]
) -> None:
stmt = select(PropertyRow).where(
col(PropertyRow.landlord_property_id).in_(list(landlord_property_ids))
)
rows = session.execute(stmt).scalars().all() # pyright: ignore[reportDeprecated]
self._by_id: dict[str, str] = {
row.landlord_property_id: row.address
for row in rows
if row.landlord_property_id is not None and row.address is not None
}
def address_for(self, landlord_property_id: str) -> str:
return self._by_id.get(landlord_property_id, self._FALLBACK)

14
repositories/uploaded_file/uploaded_file_postgres_repository.py
View file

@ -1,6 +1,6 @@
from __future__ import annotations
from typing import Optional
from typing import Optional, Sequence
from sqlalchemy import select
from sqlmodel import Session, col
@ -24,5 +24,17 @@ class UploadedFilePostgresRepository:
)
return self._session.execute(stmt).scalars().one_or_none() # pyright: ignore[reportDeprecated]
def by_landlord_property_ids(
self, landlord_property_ids: Sequence[str]
) -> list[UploadedFile]:
"""Every uploaded file held for the given properties, matched by
`landlord_property_id` (ADR-0060), across all Document Types. Returns
the raw rows — latest-per-Document-Type selection and null-type
skipping are the Download Package plan's job, not the query's."""
stmt = select(UploadedFile).where(
col(UploadedFile.landlord_property_id).in_(list(landlord_property_ids))
)
return list(self._session.execute(stmt).scalars().all()) # pyright: ignore[reportDeprecated]
def insert(self, uploaded_file: UploadedFile) -> None:
self._session.add(uploaded_file)

0
tests/backend/app/documents/__init__.py Normal file
View file

240
tests/backend/app/documents/test_bulk_download_router.py Normal file
View file

@ -0,0 +1,240 @@
"""Tests for the Bulk Document Download trigger endpoint (ADR-0060).
Session, SQS and requester-email seams are dependency-injected; tests run
against the ephemeral Postgres and record message bodies instead of calling AWS.
"""
import json
from collections.abc import Iterator
from dataclasses import dataclass, field
from typing import Any
from uuid import UUID
import pytest
from fastapi import FastAPI
from fastapi.testclient import TestClient
from sqlalchemy import Engine
from sqlmodel import Session
from backend.app.db.models.hubspot_deal_data import HubspotDealData
from backend.app.dependencies import validate_token
from backend.app.documents import router as documents_router
from backend.app.documents.router import (
get_message_sender,
get_requesting_user_email,
get_session,
)
from infrastructure.postgres.property_table import PropertyRow
from infrastructure.postgres.task_table import TaskRow
from repositories.tasks.subtask_postgres_repository import SubTaskPostgresRepository
PORTFOLIO_ID = 814
RECIPIENT = "tester@example.com"
@dataclass
class Api:
client: TestClient
engine: Engine
sent_bodies: list[str] = field(default_factory=list)
def seed_task(self, selection: dict[str, Any]) -> UUID:
"""Create the app-owned task with the FE-written selection config on
``task.inputs`` (ADR-0060)."""
with Session(self.engine) as session:
row = TaskRow(
task_source="app:bulk_document_download",
inputs=json.dumps(selection),
)
session.add(row)
session.commit()
session.refresh(row)
return row.id
def seed_property(self, landlord_property_id: str) -> int:
with Session(self.engine) as session:
row = PropertyRow(
portfolio_id=PORTFOLIO_ID, landlord_property_id=landlord_property_id
)
session.add(row)
session.commit()
session.refresh(row)
assert row.id is not None
return row.id
def seed_deal(self, project_code: str, landlord_property_id: str) -> None:
"""A hubspot_deal_data row carrying the project↔property link the
project-code selection resolves against (ADR-0060)."""
with Session(self.engine) as session:
session.add(
HubspotDealData(
deal_id=f"deal-{project_code}-{landlord_property_id}",
project_code=project_code,
landlord_property_id=landlord_property_id,
)
)
session.commit()
def subtask_inputs(self, task_id: UUID) -> list[dict[str, Any]]:
with Session(self.engine) as session:
subtasks = SubTaskPostgresRepository(session).list_by_task(task_id)
return [st.inputs or {} for st in subtasks]
@pytest.fixture
def api(db_engine: Engine) -> Api:
app = FastAPI()
app.include_router(documents_router.router, prefix="/v1")
harness = Api(client=TestClient(app), engine=db_engine)
def _session() -> Iterator[Session]:
with Session(db_engine) as session:
yield session
app.dependency_overrides[get_session] = _session
app.dependency_overrides[get_message_sender] = lambda: harness.sent_bodies.append
app.dependency_overrides[get_requesting_user_email] = lambda: RECIPIENT
app.dependency_overrides[validate_token] = lambda: "test-token"
return harness
def _trigger(api: Api, task_id: UUID) -> Any:
return api.client.post(
"/v1/documents/bulk-download", json={"task_id": str(task_id)}
)
def test_pins_the_recipe_and_enqueues_one_message(api: Api) -> None:
# arrange — two selected properties, pinned as the FE's task.inputs config.
p1 = api.seed_property("LP1")
p2 = api.seed_property("LP2")
task_id = api.seed_task(
{"portfolio_id": PORTFOLIO_ID, "property_ids": [p1, p2]}
)
# act
response = _trigger(api, task_id)
# assert — accepted; one sub_task pins the recipe; one message enqueued.
assert response.status_code == 202
inputs = api.subtask_inputs(task_id)
assert len(inputs) == 1
assert inputs[0]["landlord_property_ids"] == ["LP1", "LP2"]
assert inputs[0]["recipient_email"] == RECIPIENT
assert len(api.sent_bodies) == 1
message = json.loads(api.sent_bodies[0])
assert message["task_id"] == str(task_id)
assert "subtask_id" in message
def test_resolves_all_properties_for_the_given_project_codes(api: Api) -> None:
# arrange — two projects the user selects, plus one they don't; the
# project↔property grain lives on hubspot_deal_data, not `property`.
api.seed_deal("PROJ-A", "LP1")
api.seed_deal("PROJ-A", "LP2")
api.seed_deal("PROJ-B", "LP3")
api.seed_deal("PROJ-C", "LP9") # a different project — must not appear
task_id = api.seed_task({"project_codes": ["PROJ-A", "PROJ-B"]})
# act
response = _trigger(api, task_id)
# assert — the union of the selected projects' landlord_property_ids is pinned.
assert response.status_code == 202
inputs = api.subtask_inputs(task_id)
assert inputs[0]["landlord_property_ids"] == ["LP1", "LP2", "LP3"]
def test_unions_project_codes_with_hand_picked_property_ids(api: Api) -> None:
# arrange — one project plus a hand-picked property from outside it.
api.seed_deal("PROJ-A", "LP1")
api.seed_deal("PROJ-A", "LP2")
hand_picked = api.seed_property("LP5")
task_id = api.seed_task(
{"project_codes": ["PROJ-A"], "property_ids": [hand_picked]}
)
# act
response = _trigger(api, task_id)
# assert — both sources contribute; the set is de-duplicated and sorted.
assert response.status_code == 202
inputs = api.subtask_inputs(task_id)
assert inputs[0]["landlord_property_ids"] == ["LP1", "LP2", "LP5"]
def test_deal_rows_with_no_landlord_property_id_are_dropped(api: Api) -> None:
# arrange — a project whose deal rows carry no landlord_property_id can match
# no files (files are keyed on landlord_property_id), so the selection is empty.
api.seed_deal("PROJ-A", "LP1")
with Session(api.engine) as session:
session.add(HubspotDealData(deal_id="deal-null", project_code="PROJ-A"))
session.commit()
task_id = api.seed_task({"project_codes": ["PROJ-A"]})
# act
response = _trigger(api, task_id)
# assert — only the row that carries an id contributes.
assert response.status_code == 202
inputs = api.subtask_inputs(task_id)
assert inputs[0]["landlord_property_ids"] == ["LP1"]
def test_rejects_a_project_code_that_resolves_to_no_properties(api: Api) -> None:
# arrange — a project with no matching deal rows at all.
task_id = api.seed_task({"project_codes": ["PROJ-EMPTY"]})
# act
response = _trigger(api, task_id)
# assert — refused; nothing created or sent.
assert response.status_code == 400
assert api.subtask_inputs(task_id) == []
assert api.sent_bodies == []
def test_rejects_a_selection_over_the_cap(api: Api, monkeypatch: Any) -> None:
# arrange — cap of 1, two selected properties.
monkeypatch.setattr(documents_router, "MAX_PROPERTIES", 1)
p1 = api.seed_property("LP1")
p2 = api.seed_property("LP2")
task_id = api.seed_task(
{"portfolio_id": PORTFOLIO_ID, "property_ids": [p1, p2]}
)
# act
response = _trigger(api, task_id)
# assert — refused; nothing created or sent.
assert response.status_code == 400
assert api.subtask_inputs(task_id) == []
assert api.sent_bodies == []
def test_rejects_a_task_that_already_has_a_subtask(api: Api) -> None:
# arrange — a task already triggered once.
p1 = api.seed_property("LP1")
task_id = api.seed_task({"portfolio_id": PORTFOLIO_ID, "property_ids": [p1]})
assert _trigger(api, task_id).status_code == 202
# act
response = _trigger(api, task_id)
# assert — the double-submit is refused; still one sub_task, one message.
assert response.status_code == 409
assert len(api.subtask_inputs(task_id)) == 1
assert len(api.sent_bodies) == 1
def test_rejects_a_task_with_no_selection_config(api: Api) -> None:
# arrange — a task whose inputs carry no portfolio_id/selection.
api.seed_property("LP1")
task_id = api.seed_task({})
# act
response = _trigger(api, task_id)
# assert
assert response.status_code == 400
assert api.sent_bodies == []

0
tests/domain/bulk_document_download/__init__.py Normal file
View file

94
tests/domain/bulk_document_download/test_package_plan.py Normal file
View file

@ -0,0 +1,94 @@
"""The Download Package plan (ADR-0060) — pure layout of the ZIP from resolved
documents: one folder per property, latest file of each Document Type,
null-Document-Type rows skipped and reported."""
from datetime import datetime, timezone
from typing import Optional
from domain.bulk_document_download.package_plan import (
ResolvedDocument,
plan_download_package,
)
def _doc(
*,
landlord_property_id: str = "LP1",
address: str = "12 Oak Street",
document_type: Optional[str] = "site_note",
s3_key: str = "a.pdf",
uploaded_at: datetime = datetime(2026, 1, 1, tzinfo=timezone.utc),
) -> ResolvedDocument:
return ResolvedDocument(
landlord_property_id=landlord_property_id,
address=address,
document_type=document_type,
s3_bucket="src-bucket",
s3_key=s3_key,
uploaded_at=uploaded_at,
)
def test_keeps_only_the_latest_file_of_a_document_type_in_the_property_folder() -> None:
# Arrange — two site_note files for the same property, uploaded at
# different times; only the newest should be packaged.
older = _doc(
s3_key="old.pdf", uploaded_at=datetime(2026, 1, 1, tzinfo=timezone.utc)
)
newer = _doc(
s3_key="new.pdf", uploaded_at=datetime(2026, 6, 1, tzinfo=timezone.utc)
)
# Act
plan = plan_download_package([older, newer])
# Assert — one entry, the newer file, foldered under the property.
assert len(plan.entries) == 1
entry = plan.entries[0]
assert entry.s3_key == "new.pdf"
assert entry.zip_path.startswith("12 Oak Street (LP1)/")
def test_skips_and_reports_a_document_with_no_document_type() -> None:
# Arrange — one packable file and one row whose Document Type is null
# (file_type is nullable on uploaded_files).
packable = _doc(document_type="site_note", s3_key="note.pdf")
untyped = _doc(document_type=None, s3_key="mystery.bin")
# Act
plan = plan_download_package([packable, untyped])
# Assert — the untyped row is left out of the archive and reported instead.
assert [e.s3_key for e in plan.entries] == ["note.pdf"]
assert len(plan.skipped) == 1
skipped = plan.skipped[0]
assert skipped.s3_key == "mystery.bin"
assert skipped.landlord_property_id == "LP1"
def test_packaged_filename_is_the_document_type_with_the_original_extension() -> None:
# Arrange — a photo pack stored under a prefixed key with a .zip extension;
# the file inside the archive should be named by type but stay openable.
doc = _doc(document_type="photo_pack", s3_key="surveys/2026/pack.zip")
# Act
plan = plan_download_package([doc])
# Assert
assert plan.entries[0].zip_path == "12 Oak Street (LP1)/photo_pack.zip"
def test_latest_per_type_is_scoped_per_property_each_gets_its_own_folder() -> None:
# Arrange — two DIFFERENT properties, each with a site_note. Latest-per-type
# must not dedupe across properties: both survive, in their own folders.
first = _doc(landlord_property_id="LP1", address="12 Oak Street", s3_key="a.pdf")
second = _doc(landlord_property_id="LP2", address="9 Elm Road", s3_key="b.pdf")
# Act
plan = plan_download_package([first, second])
# Assert
assert {e.zip_path for e in plan.entries} == {
"12 Oak Street (LP1)/site_note.pdf",
"9 Elm Road (LP2)/site_note.pdf",
}

0
tests/infrastructure/email/__init__.py Normal file
View file

104
tests/infrastructure/email/test_ses_smtp_email_sender.py Normal file
View file

@ -0,0 +1,104 @@
"""The SES-over-SMTP email sender (ADR-0059) — delivers a plain-text message to
the requesting user via an injected SMTP transport."""
from __future__ import annotations
from email.message import EmailMessage
from types import TracebackType
from typing import Any, Optional, Type
from infrastructure.email.ses_smtp_email_sender import SesSmtpEmailSender
class _FakeSmtp:
"""Records what the sender hands the transport, standing in for a live SES
SMTP connection."""
def __init__(self, host: str, port: int) -> None:
self.host = host
self.port = port
self.messages: list[EmailMessage] = []
self.started_tls = False
self.login_args: Optional[tuple[str, str]] = None
def __enter__(self) -> "_FakeSmtp":
return self
def __exit__(
self,
exc_type: Optional[Type[BaseException]],
exc: Optional[BaseException],
tb: Optional[TracebackType],
) -> Optional[bool]:
return None
def starttls(self) -> Any:
self.started_tls = True
def login(self, user: str, password: str) -> Any:
self.login_args = (user, password)
def send_message(self, message: EmailMessage) -> Any:
self.messages.append(message)
def test_send_delivers_a_plaintext_message_with_the_right_envelope() -> None:
# Arrange
created: list[_FakeSmtp] = []
def factory(host: str, port: int) -> _FakeSmtp:
smtp = _FakeSmtp(host, port)
created.append(smtp)
return smtp
sender = SesSmtpEmailSender(
host="email-smtp.eu-west-2.amazonaws.com",
port=587,
username="AKIA_SMTP_USER",
password="smtp-password",
from_address="noreply@domna.homes",
smtp_factory=factory,
)
# Act
sender.send(
to="user@example.com",
subject="Your document download is ready",
body="Download it here: https://signed-url",
)
# Assert — one message, correctly addressed, body carried through.
assert len(created) == 1
assert len(created[0].messages) == 1
message = created[0].messages[0]
assert message["To"] == "user@example.com"
assert message["From"] == "noreply@domna.homes"
assert message["Subject"] == "Your document download is ready"
assert "https://signed-url" in message.get_content()
def test_send_starts_tls_and_authenticates_before_sending() -> None:
# Arrange — SES SMTP rejects unauthenticated, cleartext senders, so the
# transport must STARTTLS and log in with the SMTP credentials.
created: list[_FakeSmtp] = []
def factory(host: str, port: int) -> _FakeSmtp:
smtp = _FakeSmtp(host, port)
created.append(smtp)
return smtp
sender = SesSmtpEmailSender(
host="email-smtp.eu-west-2.amazonaws.com",
port=587,
username="AKIA_SMTP_USER",
password="smtp-password",
from_address="noreply@domna.homes",
smtp_factory=factory,
)
# Act
sender.send(to="user@example.com", subject="s", body="b")
# Assert
assert created[0].started_tls is True
assert created[0].login_args == ("AKIA_SMTP_USER", "smtp-password")

30
tests/infrastructure/test_s3_client.py
View file

@ -34,3 +34,33 @@ def test_get_object_returns_bytes_written_by_put_object(s3_client: S3Client) ->
def test_bucket_property_exposes_configured_bucket(s3_client: S3Client) -> None:
# act / assert
assert s3_client.bucket == BUCKET
def test_generate_presigned_url_signs_a_get_for_the_object(s3_client: S3Client) -> None:
# arrange
s3_client.put_object("bulk-downloads/pkg.zip", b"zip-bytes")
# act
url = s3_client.generate_presigned_url("bulk-downloads/pkg.zip", expires_in=3600)
# assert — a signed GET URL for this bucket + key
assert BUCKET in url
assert "bulk-downloads/pkg.zip" in url
assert "Signature" in url
def test_upload_file_streams_a_local_file_to_s3(
s3_client: S3Client, tmp_path: object
) -> None:
# arrange — a file on local disk (stands in for the /tmp ZIP)
from pathlib import Path
local = Path(str(tmp_path)) / "package.zip"
local.write_bytes(b"ZIP-CONTENT")
# act
uri = s3_client.upload_file(str(local), "bulk-downloads/package.zip")
# assert — the object landed with the file's bytes
assert uri == f"s3://{BUCKET}/bulk-downloads/package.zip"
assert s3_client.get_object("bulk-downloads/package.zip") == b"ZIP-CONTENT"

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"

244
tests/orchestration/test_bulk_document_download_orchestrator.py Normal file
View file

@ -0,0 +1,244 @@
"""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 domain.tasks.subtasks import SubTaskFailure
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 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:
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
def test_a_selection_with_no_documents_is_a_recorded_failure(
packages: S3Client,
) -> None:
# Arrange — the selection resolves to no packageable documents.
email = _RecordingEmail()
orchestrator = BulkDocumentDownloadOrchestrator(
uploaded_files=_FakeUploadedFiles([]),
addresses=_FakeAddresses({}),
documents=_FakeDocuments({}),
packages=packages,
email=email,
url_ttl_seconds=3600,
)
# Act / Assert — a recorded SubTask failure, and nobody is emailed an empty
# package.
with pytest.raises(SubTaskFailure):
orchestrator.run(
landlord_property_ids=["LP1"],
recipient_email="user@example.com",
package_name="empty",
)
assert email.sent == []
def test_packages_the_good_documents_and_reports_the_skipped_ones(
packages: S3Client,
) -> None:
# Arrange — one packable site note and one row with no Document Type.
good = _uploaded_file("LP1", "src-bucket", "surveys/a.pdf")
untyped = UploadedFile(
s3_file_bucket="src-bucket",
s3_file_key="mystery.bin",
s3_upload_timestamp=datetime(2026, 1, 1, tzinfo=timezone.utc),
landlord_property_id="LP1",
file_type=None,
)
email = _RecordingEmail()
orchestrator = BulkDocumentDownloadOrchestrator(
uploaded_files=_FakeUploadedFiles([good, untyped]),
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="mixed",
)
# Assert — the good file is packaged and delivered; the untyped one is
# reported, not fatal.
assert result.included == 1
assert [s.s3_key for s in result.skipped] == ["mystery.bin"]
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_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"]

35
tests/repositories/property/test_landlord_address_resolver.py Normal file
View file

@ -0,0 +1,35 @@
from collections.abc import Iterator
import pytest
from sqlalchemy import Engine
from sqlmodel import Session
from infrastructure.postgres.property_table import PropertyRow
from repositories.property.landlord_address_resolver import LandlordAddressResolver
@pytest.fixture
def session(db_engine: Engine) -> Iterator[Session]:
with Session(db_engine) as s:
yield s
def test_resolves_known_addresses_and_falls_back_for_unknowns(
session: Session,
) -> None:
# arrange — two properties with addresses; a third id has no property row.
session.add(
PropertyRow(portfolio_id=1, landlord_property_id="LP1", address="12 Oak Street")
)
session.add(
PropertyRow(portfolio_id=1, landlord_property_id="LP2", address="9 Elm Road")
)
session.flush()
# act
resolver = LandlordAddressResolver(session, ["LP1", "LP2", "LP3"])
# assert
assert resolver.address_for("LP1") == "12 Oak Street"
assert resolver.address_for("LP2") == "9 Elm Road"
assert resolver.address_for("LP3") == "address unavailable"

35
tests/repositories/uploaded_file/test_uploaded_file_postgres_repository.py
View file

@ -78,3 +78,38 @@ def test_does_not_return_row_with_different_file_type(db_engine: Engine) -> None
# Assert
assert result is None
def _landlord_file(
landlord_property_id: str,
s3_file_key: str,
file_type: FileTypeEnum = FileTypeEnum.SITE_NOTE,
) -> UploadedFile:
return UploadedFile(
s3_file_bucket=_BUCKET,
s3_file_key=s3_file_key,
s3_upload_timestamp=datetime(2026, 1, 1, 12, 0, 0, tzinfo=timezone.utc),
landlord_property_id=landlord_property_id,
file_type=file_type.value,
)
def test_by_landlord_property_ids_returns_only_requested_properties(
db_engine: Engine,
) -> None:
# Arrange — files for three properties; only two are requested.
with Session(db_engine) as session:
session.add(_landlord_file("LP1", "one.pdf"))
session.add(_landlord_file("LP2", "two.pdf"))
session.add(_landlord_file("LP3", "three.pdf"))
session.commit()
# Act
with Session(db_engine) as session:
found = UploadedFilePostgresRepository(session).by_landlord_property_ids(
["LP1", "LP2"]
)
result = {(f.landlord_property_id, f.s3_file_key) for f in found}
# Assert
assert result == {("LP1", "one.pdf"), ("LP2", "two.pdf")}