mirror of
https://github.com/Hestia-Homes/Model.git
synced 2026-07-12 13:29:04 +00:00
48 lines
1.7 KiB
Python
48 lines
1.7 KiB
Python
"""The message contract between the HubSpot scraper and the abri lambda.
|
|
|
|
One message per scrape, naming every Abri flow that fired plus the deal
|
|
fields those flows need. A message missing a field its flow needs fails
|
|
validation — the resulting failed task / DLQ entry is the agreed
|
|
visibility surface.
|
|
"""
|
|
|
|
from datetime import date
|
|
from typing import Dict, List, Literal, Optional, Tuple
|
|
|
|
from pydantic import BaseModel, ConfigDict, Field, model_validator
|
|
|
|
AbriFlow = Literal["amend_job", "log_job", "sync_tenant_data", "abandon_job"]
|
|
|
|
# The deal fields each flow reads; hubspot_deal_id is always required.
|
|
# abandon_job needs no field beyond the deal id (the outcome is optional, only
|
|
# feeding the reason-mapping seam).
|
|
_FIELDS_BY_FLOW: Dict[AbriFlow, Tuple[str, ...]] = {
|
|
"amend_job": ("confirmed_survey_date",),
|
|
"log_job": ("place_ref", "deal_name", "confirmed_survey_date"),
|
|
"sync_tenant_data": ("place_ref",),
|
|
"abandon_job": (),
|
|
}
|
|
|
|
|
|
class AbriTriggerRequest(BaseModel):
|
|
model_config = ConfigDict(extra="ignore")
|
|
|
|
hubspot_deal_id: str
|
|
flows: List[AbriFlow] = Field(min_length=1)
|
|
place_ref: Optional[str] = None
|
|
deal_name: Optional[str] = None
|
|
confirmed_survey_date: Optional[date] = None
|
|
confirmed_survey_time: Optional[str] = None
|
|
outcome: Optional[str] = None
|
|
|
|
@model_validator(mode="after")
|
|
def _each_fired_flow_has_its_fields(self) -> "AbriTriggerRequest":
|
|
missing = [
|
|
f"flow {flow} requires {field}"
|
|
for flow in self.flows
|
|
for field in _FIELDS_BY_FLOW[flow]
|
|
if getattr(self, field) is None
|
|
]
|
|
if missing:
|
|
raise ValueError("; ".join(missing))
|
|
return self
|