mirror of
https://github.com/Hestia-Homes/Model.git
synced 2026-06-08 11:17:27 +00:00
34 lines
1 KiB
Python
34 lines
1 KiB
Python
import json
|
|
import time
|
|
from typing import Any, Mapping
|
|
|
|
from backend.app.db.functions.tasks.Tasks import SubTaskInterface
|
|
from backend.app.plan.utils import build_cloudwatch_log_url
|
|
from backend.categorisation.categorisation_trigger_request import (
|
|
CategorisationTriggerRequest,
|
|
)
|
|
from backend.categorisation.processor import process_portfolio
|
|
from utils.logger import setup_logger
|
|
|
|
|
|
logger = setup_logger()
|
|
|
|
|
|
def handler(event: Mapping[str, Any], context: Any) -> None:
|
|
|
|
logger.info("Received message")
|
|
|
|
logger.info(f"Number of events: {len(event.get('Records', []))}")
|
|
|
|
for record in event.get("Records", []):
|
|
try:
|
|
body_dict = json.loads(record["body"])
|
|
logger.debug("Validating request body")
|
|
payload = CategorisationTriggerRequest.model_validate(body_dict)
|
|
|
|
logger.debug("Successfully validated request body")
|
|
|
|
process_portfolio(payload)
|
|
except Exception as e:
|
|
logger.info("Handler exception")
|
|
logger.error(f"Failed to process record: {e}")
|