diff --git a/applications/ara_export/__init__.py b/applications/ara_export/__init__.py new file mode 100644 index 000000000..e69de29bb diff --git a/applications/ara_export/ara_export_trigger_body.py b/applications/ara_export/ara_export_trigger_body.py new file mode 100644 index 000000000..cf12f13be --- /dev/null +++ b/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 diff --git a/applications/ara_export/handler.py b/applications/ara_export/handler.py new file mode 100644 index 000000000..9e78e2a7b --- /dev/null +++ b/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, + } diff --git a/backend/app/config.py b/backend/app/config.py index 8f0d551f1..c5ec42948 100644 --- a/backend/app/config.py +++ b/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" diff --git a/backend/app/exports/router.py b/backend/app/exports/router.py new file mode 100644 index 000000000..e4d99b9d0 --- /dev/null +++ b/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) + ) diff --git a/backend/app/exports/schemas.py b/backend/app/exports/schemas.py new file mode 100644 index 000000000..bd3554d07 --- /dev/null +++ b/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 diff --git a/backend/app/main.py b/backend/app/main.py index 270917f63..2060d0604 100644 --- a/backend/app/main.py +++ b/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 diff --git a/tests/applications/ara_export/__init__.py b/tests/applications/ara_export/__init__.py new file mode 100644 index 000000000..e69de29bb diff --git a/tests/applications/ara_export/test_ara_export_trigger_body.py b/tests/applications/ara_export/test_ara_export_trigger_body.py new file mode 100644 index 000000000..74c9cbfc2 --- /dev/null +++ b/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")