Add the scenario export route, worker handler and SQS trigger 🟩

Wires POST /v1/exports/scenario -> tasks.inputs recipe -> pinned sub_task ->
ARA_EXPORT_SQS_URL -> ara_export Lambda -> orchestrator. Route resolution and the
trigger body are covered by tests; the Lambda handler mirrors bulk_document_download.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

applications/ara_export/ara_export_trigger_body.py

@@ -0,0 +1,19 @@
+from uuid import UUID
+
+from pydantic import BaseModel, ConfigDict
+
+
+class AraExportTriggerBody(BaseModel):
+    """SQS trigger for the ara_export Lambda (ADR-0055/0065).
+
+    Attach mode: the FastAPI route created the Task + SubTask and pinned the
+    recipe (``portfolio_id``, ``scenario_ids``, ``property_ids``,
+    ``recipient_email``, ``export_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

applications/ara_export/handler.py

@@ -0,0 +1,121 @@
+"""ara_export Lambda (ADR-0065).
+
+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, builds the sheet-per-scenario branded workbook through the orchestrator,
+and returns the result (presigned URL + counts), which ``TaskOrchestrator``
+records on ``sub_task.outputs``. A selection that yields no rows raises
+``SubTaskFailure`` from the orchestrator — recorded, not retried.
+
+PostgresConfig-only (POSTGRES_*), like the bulk_document_download Lambda. The
+workbook is written to the dedicated ``DOCUMENT_EXPORTS_BUCKET``; SES SMTP
+settings arrive as env vars.
+"""
+
+from __future__ import annotations
+
+import logging
+import os
+from typing import Any
+
+import boto3
+from sqlalchemy import text
+from sqlmodel import Session
+
+from applications.ara_export.ara_export_trigger_body import AraExportTriggerBody
+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 orchestration.scenario_export_orchestrator import ScenarioExportOrchestrator
+from repositories.scenario_export.scenario_export_postgres_repository import (
+    ScenarioExportPostgresRepository,
+)
+from repositories.tasks.subtask_postgres_repository import SubTaskPostgresRepository
+from utilities.aws_lambda.task_handler import task_handler
+
+logging.getLogger().setLevel(logging.INFO)
+logger = logging.getLogger(__name__)
+
+# The export link is valid for 60 minutes (ADR-0065).
+_URL_TTL_SECONDS = 3600
+
+
+class _ScenarioNames:
+    """Scenario id → display name, read from the ``scenario`` table (the domain
+    Scenario aggregate does not carry the name)."""
+
+    def __init__(self, session: Session, scenario_ids: list[int]) -> None:
+        rows = (
+            session.connection()
+            .execute(
+                text("SELECT id, name FROM scenario WHERE id = ANY(:ids)"),
+                {"ids": scenario_ids},
+            )
+            .all()
+        )
+        self._names: dict[int, str] = {
+            row[0]: (row[1] or f"scenario_{row[0]}") for row in rows
+        }
+
+    def name_for(self, scenario_id: int) -> str:
+        return self._names.get(scenario_id, f"scenario_{scenario_id}")
+
+
+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="ara_export", source=Source.PORTFOLIO)
+def handler(body: dict[str, Any], context: Any) -> dict[str, Any]:
+    trigger = AraExportTriggerBody.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 {}
+        portfolio_id = int(recipe["portfolio_id"])
+        scenario_ids: list[int] = [int(x) for x in recipe["scenario_ids"]]
+        property_ids: list[int] = [int(x) for x in recipe["property_ids"]]
+        recipient_email = str(recipe["recipient_email"])
+        export_name = str(recipe["export_name"])
+
+        logger.info(
+            "ara_export: starting export for %d properties across %d scenario(s) "
+            "(subtask=%s)",
+            len(property_ids),
+            len(scenario_ids),
+            trigger.subtask_id,
+        )
+
+        boto_s3: Any = boto3.client("s3")  # pyright: ignore[reportUnknownMemberType]
+        orchestrator = ScenarioExportOrchestrator(
+            rows=ScenarioExportPostgresRepository(session),
+            scenarios=_ScenarioNames(session, scenario_ids),
+            exports=S3Client(boto_s3, os.environ["DOCUMENT_EXPORTS_BUCKET"]),
+            email=_email_sender(),
+            url_ttl_seconds=_URL_TTL_SECONDS,
+        )
+        result = orchestrator.run(
+            portfolio_id=portfolio_id,
+            scenario_ids=scenario_ids,
+            property_ids=property_ids,
+            recipient_email=recipient_email,
+            export_name=export_name,
+        )
+    finally:
+        session.close()
+
+    return {
+        "presigned_url": result.presigned_url,
+        "export_s3_key": result.export_s3_key,
+        "sheet_count": result.sheet_count,
+        "row_count": result.row_count,
+    }

backend/app/config.py

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

backend/app/exports/router.py

@@ -0,0 +1,170 @@
+"""Scenario Export trigger: POST /v1/exports/scenario (ADR-0065).
+
+Resolves the authenticated requester's email, reads the FE-written selection
+recipe from ``task.inputs``, resolves + caps the property selection (an explicit
+``property_ids`` list overrides the filters; otherwise the shared Modelling Run
+filter resolver runs), pins the resolved recipe onto a single pre-created
+sub_task under the app-owned task, and enqueues one message to the ara_export
+worker. Never builds the workbook 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.exports.export_tasks import ScenarioExportTasks
+from backend.app.exports.schemas import ScenarioExportRequest
+from backend.app.modelling.property_filters import PropertyGroupFilters
+
+# The selection cap (ADR-0065) — rejected at the trigger so an oversized export
+# never reaches the worker. Configurable ceiling; the JSON-column mechanism has
+# no limit of its own.
+MAX_PROPERTIES = 100_000
+
+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.ARA_EXPORT_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 export recipient (ADR-0059)."""
+    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":
+        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="/exports",
+    tags=["exports"],
+    dependencies=[Depends(validate_token)],
+)
+
+
+@router.post("/scenario", status_code=202)
+async def scenario_export(
+    body: ScenarioExportRequest,
+    session: Session = Depends(get_session),
+    send_message: MessageSender = Depends(get_message_sender),
+    recipient_email: str = Depends(get_requesting_user_email),
+) -> dict[str, str]:
+    tasks = ScenarioExportTasks(session)
+    if tasks.already_distributed(body.task_id):
+        raise HTTPException(
+            status_code=409,
+            detail=(
+                f"Task {body.task_id} already has a sub_task — an export has "
+                "already been started for it."
+            ),
+        )
+
+    config = tasks.read_selection_config(body.task_id)
+    portfolio_id: Any = config.get("portfolio_id")
+    raw_scenario_ids: Any = config.get("scenario_ids") or []
+    raw_property_ids: Any = config.get("property_ids") or []
+    raw_filters: Any = config.get("filters") or {}
+
+    if not isinstance(portfolio_id, int):
+        raise HTTPException(
+            status_code=400, detail="task.inputs must provide an integer portfolio_id."
+        )
+    if not _is_int_list(raw_scenario_ids) or not raw_scenario_ids:
+        raise HTTPException(
+            status_code=400,
+            detail="task.inputs scenario_ids must be a non-empty list of integers.",
+        )
+    if not _is_int_list(raw_property_ids):
+        raise HTTPException(
+            status_code=400,
+            detail="task.inputs property_ids must be a list of integers.",
+        )
+    if not isinstance(raw_filters, dict):
+        raise HTTPException(
+            status_code=400, detail="task.inputs filters must be an object."
+        )
+
+    scenario_ids: list[int] = cast(list[int], raw_scenario_ids)
+    property_ids: list[int] = cast(list[int], raw_property_ids)
+    try:
+        filters = PropertyGroupFilters(**cast(dict[str, Any], raw_filters))
+    except Exception as exc:  # pragma: no cover - pydantic validation detail
+        raise HTTPException(
+            status_code=400, detail=f"task.inputs filters are invalid: {exc}"
+        ) from exc
+
+    resolved_property_ids = tasks.resolve_property_ids(
+        portfolio_id=portfolio_id, filters=filters, property_ids=property_ids
+    )
+    if not resolved_property_ids:
+        raise HTTPException(
+            status_code=400,
+            detail="The selection resolves to no properties.",
+        )
+    if len(resolved_property_ids) > MAX_PROPERTIES:
+        raise HTTPException(
+            status_code=400,
+            detail=(
+                f"The selection has {len(resolved_property_ids)} properties, over "
+                f"the {MAX_PROPERTIES} limit — narrow your selection."
+            ),
+        )
+
+    subtask_id = uuid4()
+    recipe = {
+        "portfolio_id": portfolio_id,
+        "scenario_ids": scenario_ids,
+        "property_ids": resolved_property_ids,
+        "recipient_email": recipient_email,
+        "export_name": f"portfolio-{portfolio_id}-{body.task_id}",
+    }
+    created = tasks.create_export_subtask(body.task_id, subtask_id, recipe)
+    if not created:
+        raise HTTPException(
+            status_code=409,
+            detail=(
+                f"Task {body.task_id} already has a sub_task — an export has "
+                "already been started for it."
+            ),
+        )
+
+    send_message(
+        json.dumps({"task_id": str(body.task_id), "subtask_id": str(subtask_id)})
+    )
+    return {"message": "Scenario export started"}
+
+
+def _is_int_list(value: Any) -> bool:
+    # bool is an int subclass; exclude it so a stray True/False isn't a valid id.
+    return isinstance(value, list) and all(
+        isinstance(item, int) and not isinstance(item, bool)
+        for item in cast(list[Any], value)
+    )

backend/app/exports/schemas.py

@@ -0,0 +1,12 @@
+from uuid import UUID
+
+from pydantic import BaseModel
+
+
+class ScenarioExportRequest(BaseModel):
+    """The Scenario Export trigger body (ADR-0065). Carries only the app-owned
+    ``task_id``: the FE wrote the selection recipe (portfolio, scenarios,
+    filters, optional property_ids) into ``task.inputs``, so the (potentially
+    large) selection never travels in the HTTP body."""
+
+    task_id: UUID

backend/app/main.py

@@ -12,6 +12,7 @@ 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.exports import router as exports_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
@@ -67,6 +68,7 @@ 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")
+app.include_router(exports_router.router, prefix="/v1")
 
 if get_settings().ENVIRONMENT == "local":
     from backend.app.local import router as local_router

tests/applications/ara_export/test_ara_export_trigger_body.py

@@ -0,0 +1,24 @@
+"""The ara_export SQS trigger body (ADR-0065): carries only the task/subtask
+identifiers (the selection rides sub_task.inputs), tolerating extra fields."""
+
+from __future__ import annotations
+
+from uuid import UUID
+
+from applications.ara_export.ara_export_trigger_body import AraExportTriggerBody
+
+
+def test_parses_the_task_and_subtask_ids_and_tolerates_extra_fields() -> None:
+    # arrange — a message body with the two ids and an extra AWS-added field.
+    raw = {
+        "task_id": "11111111-1111-1111-1111-111111111111",
+        "subtask_id": "22222222-2222-2222-2222-222222222222",
+        "Records": "ignored",
+    }
+
+    # act
+    body = AraExportTriggerBody.model_validate(raw)
+
+    # assert — the ids parse as UUIDs; the extra field is tolerated, not fatal.
+    assert body.task_id == UUID("11111111-1111-1111-1111-111111111111")
+    assert body.subtask_id == UUID("22222222-2222-2222-2222-222222222222")