The abri lambda dispatches queue messages through the Abri orchestrator 🟩

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

applications/abri/handler.py

@@ -0,0 +1,57 @@
+"""Lambda entry point for the Abri OpenHousing flows.
+
+Consumes the abri SQS queue (one trigger message per scrape, sent by the
+HubSpot scraper) and dispatches the fired flows through the AbriOrchestrator.
+Env-to-clients glue only; behaviour lives behind the dispatch seam.
+"""
+
+import os
+from typing import Any, Dict
+
+from hubspot.client import Client  # type: ignore[reportMissingTypeStubs]
+from sqlalchemy.pool import NullPool
+from sqlmodel import Session
+
+from applications.abri.abri_trigger_request import AbriTriggerRequest
+from applications.abri.dispatch import dispatch_abri_flows
+from domain.tasks.tasks import Source
+from infrastructure.abri.abri_client import AbriClient
+from infrastructure.abri.config import AbriConfig
+from infrastructure.hubspot.deal_contacts_client import HubspotDealContactsClient
+from infrastructure.hubspot.deal_properties_client import HubspotDealPropertiesClient
+from infrastructure.postgres.config import PostgresConfig
+from infrastructure.postgres.engine import make_engine
+from orchestration.abri_orchestrator import AbriOrchestrator
+from repositories.hubspot_deals.deal_database_postgres_gateway import (
+    DealDatabasePostgresGateway,
+)
+from utilities.aws_lambda.task_handler import task_handler
+from utilities.logger import setup_logger
+
+logger = setup_logger()
+
+
+@task_handler(task_source="abri", source=Source.HUBSPOT_DEAL)
+def handler(body: dict[str, Any], context: Any) -> Dict[str, str]:
+    request = AbriTriggerRequest.model_validate(body)
+
+    abri_client = AbriClient(config=AbriConfig.from_env(os.environ))
+    sdk_client: Client = Client.create(  # type: ignore[reportUnknownMemberType]
+        access_token=os.environ["HUBSPOT_API_KEY"]
+    )
+
+    # NullPool: a fresh engine per invocation, so pooling would only
+    # accumulate idle connections across warm Lambda containers.
+    engine = make_engine(PostgresConfig.from_env(dict(os.environ)), poolclass=NullPool)
+    with Session(engine) as session:
+        deal_database = DealDatabasePostgresGateway(session=session)
+        orchestrator = AbriOrchestrator(
+            abri_client=abri_client,
+            deal_contacts=HubspotDealContactsClient(sdk_client=sdk_client),
+            deal_properties=HubspotDealPropertiesClient(sdk_client=sdk_client),
+            deal_database=deal_database,
+        )
+        summary = dispatch_abri_flows(request, orchestrator, deal_database)
+
+    logger.info("Abri flows completed for deal %s: %s", request.hubspot_deal_id, summary)
+    return summary

applications/abri/handler/Dockerfile

@@ -0,0 +1,18 @@
+FROM public.ecr.aws/lambda/python:3.11
+
+WORKDIR /var/task
+
+COPY applications/abri/handler/requirements.txt .
+RUN pip install --no-cache-dir -r requirements.txt
+
+COPY utilities/ utilities/
+COPY utils/ utils/
+COPY backend/ backend/
+COPY applications/ applications/
+COPY domain/ domain/
+COPY datatypes/ datatypes/
+COPY orchestration/ orchestration/
+COPY repositories/ repositories/
+COPY infrastructure/ infrastructure/
+
+CMD ["applications.abri.handler.handler"]

applications/abri/handler/requirements.txt

@@ -0,0 +1,8 @@
+awslambdaric
+requests
+sqlalchemy==2.0.36
+sqlmodel
+psycopg2-binary==2.9.10
+pydantic-settings==2.6.0
+boto3==1.35.44
+hubspot-api-client