mirror of
https://github.com/Hestia-Homes/Model.git
synced 2026-07-12 13:29:04 +00:00
45 lines
1.5 KiB
Python
45 lines
1.5 KiB
Python
from __future__ import annotations
|
|
|
|
import os
|
|
from typing import Any
|
|
|
|
import boto3
|
|
|
|
from applications.audit_generator.audit_generator_trigger_request import (
|
|
AuditGeneratorTriggerRequest,
|
|
)
|
|
from infrastructure.postgres.config import PostgresConfig
|
|
from infrastructure.postgres.engine import make_engine, make_session
|
|
from infrastructure.s3.s3_client import S3Client
|
|
from orchestration.audit_generator_orchestrator import AuditGeneratorOrchestrator
|
|
from orchestration.audit_generator_unit_of_work import AuditGeneratorUnitOfWork
|
|
from orchestration.task_orchestrator import TaskOrchestrator
|
|
from utilities.aws_lambda.subtask_handler import subtask_handler
|
|
|
|
|
|
@subtask_handler()
|
|
def handler(
|
|
body: dict[str, Any], context: Any, task_orchestrator: TaskOrchestrator
|
|
) -> None:
|
|
trigger = AuditGeneratorTriggerRequest.model_validate(body)
|
|
|
|
boto3_client: Any = (
|
|
boto3.client
|
|
) # pyright: ignore[reportUnknownMemberType, reportUnknownVariableType]
|
|
boto_s3: Any = boto3_client("s3")
|
|
bucket = os.environ["S3_BUCKET_NAME"]
|
|
s3_client = S3Client(boto_s3_client=boto_s3, bucket=bucket)
|
|
|
|
engine = make_engine(PostgresConfig.from_env(os.environ))
|
|
|
|
def session_factory() -> Any:
|
|
return make_session(engine)
|
|
|
|
def uow_factory() -> AuditGeneratorUnitOfWork:
|
|
return AuditGeneratorUnitOfWork(session_factory)
|
|
|
|
AuditGeneratorOrchestrator(
|
|
hubspot_deal_id=trigger.hubspot_deal_id,
|
|
s3_client=s3_client,
|
|
uow_factory=uow_factory,
|
|
).run()
|