mirror of
https://github.com/Hestia-Homes/Model.git
synced 2026-07-27 23:35:01 +00:00
68 lines
2.5 KiB
Python
68 lines
2.5 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, field_validator, 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 requires no single field: it dates the cancel to whichever of the
|
|
# confirmed survey date or last submission date is present, so requiring either
|
|
# alone would reject valid messages; the orchestrator fails the flow if a deal
|
|
# carries neither. Its outcome is optional too, only feeding the reason seam.
|
|
_FIELDS_BY_FLOW: Dict[AbriFlow, Tuple[str, ...]] = {
|
|
"amend_job": ("confirmed_survey_date", "third_party_surveyor_identifier"),
|
|
"log_job": (
|
|
"place_ref",
|
|
"deal_name",
|
|
"confirmed_survey_date",
|
|
"third_party_surveyor_identifier",
|
|
),
|
|
"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
|
|
last_submission_date: Optional[date] = None
|
|
outcome: Optional[str] = None
|
|
# OpenHousing's bookable resource for the survey; required by the log and
|
|
# amend flows. Empty string is also rejected by Abri, so not allowed here
|
|
third_party_surveyor_identifier: Optional[str] = None
|
|
|
|
@field_validator("third_party_surveyor_identifier")
|
|
@classmethod
|
|
def _blank_surveyor_identifier_is_missing(
|
|
cls, value: Optional[str]
|
|
) -> Optional[str]:
|
|
if value is not None and not value.strip():
|
|
return None
|
|
return value
|
|
|
|
@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
|