Add planning-authority HubSpot deal properties to ETL sync

Sync Planning Authority, Designated Area, Article 4 PD Rights, and
Listed Building from HubSpot into hubspot_deal_data so downstream
consumers of the existing HubSpot ETL process can read them.

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
This commit is contained in:
Jun-te Kim 2026-07-08 14:45:03 +00:00
parent a76e49e133
commit b8316418e1
4 changed files with 65 additions and 0 deletions

View file

@ -93,6 +93,11 @@ class HubspotDealData(SQLModel, table=True):
last_outbound_email: Optional[datetime] = Field(default=None)
last_submission_date: Optional[datetime] = Field(default=None)
planning_authority: Optional[str] = Field(default=None)
designated_area: Optional[str] = Field(default=None)
article_4_pd_rights: Optional[str] = Field(default=None)
listed_building: Optional[str] = Field(default=None)
created_at: Optional[datetime] = Field(
sa_column=Column(
DateTime(timezone=True),

View file

@ -302,6 +302,10 @@ class HubspotDataToDb:
"last_submission_date": parse_hs_date(
deal_data.get("last_submission_date")
),
"planning_authority": deal_data.get("planning_authority"),
"designated_area": deal_data.get("designated_area"),
"article_4_pd_rights": deal_data.get("article_pd_rights"),
"listed_building": deal_data.get("listed_building"),
}.items():
setattr(existing, attr, value)
@ -407,6 +411,10 @@ class HubspotDataToDb:
last_outbound_call=parse_hs_date(deal_data.get("last_outbound_call")),
last_outbound_email=parse_hs_date(deal_data.get("last_outbound_email")),
last_submission_date=parse_hs_date(deal_data.get("last_submission_date")),
planning_authority=deal_data.get("planning_authority"),
designated_area=deal_data.get("designated_area"),
article_4_pd_rights=deal_data.get("article_pd_rights"),
listed_building=deal_data.get("listed_building"),
)
def _handle_existing_photo_upload(

View file

@ -109,6 +109,10 @@ class HubspotDealDiffer:
"measures_for_pibi_ordered": "measures_for_pibi_ordered",
"property_halted_reason": "property_halted_reason",
"technical_approved_measures_for_install": "technical_approved_measures_for_install",
"planning_authority": "planning_authority",
"designated_area": "designated_area",
"article_pd_rights": "article_4_pd_rights",
"listed_building": "listed_building",
}
for hs_field, db_field in FIELD_MAP.items():

View file

@ -127,3 +127,51 @@ def test_build_new_deal__client_booking_reference_maps_to_job_no() -> None:
)
assert new_deal.client_booking_reference == "AD0226519"
def test_build_new_deal__planning_fields_mapped() -> None:
new_deal = _make_instance()._build_new_deal(
deal_id="MOCK_DEAL_ID",
deal_data={
"planning_authority": "Test Council",
"designated_area": "Conservation Area",
"article_pd_rights": "Removed",
"listed_building": "Grade II",
},
listing=None,
company=None,
project=None,
)
assert new_deal.planning_authority == "Test Council"
assert new_deal.designated_area == "Conservation Area"
assert new_deal.article_4_pd_rights == "Removed"
assert new_deal.listed_building == "Grade II"
def test_update_existing_deal__planning_fields_overwritten() -> None:
existing = HubspotDealData(
deal_id="MOCK_DEAL_ID",
planning_authority="Old Council",
designated_area="Old Area",
article_4_pd_rights="Old Rights",
listed_building="Old Listing",
)
deal_data = {
"planning_authority": "New Council",
"designated_area": "New Area",
"article_pd_rights": "New Rights",
"listed_building": "New Listing",
}
_make_instance()._update_existing_deal(
existing=existing,
deal_data=deal_data,
listing=None,
company=None,
)
assert existing.planning_authority == "New Council"
assert existing.designated_area == "New Area"
assert existing.article_4_pd_rights == "New Rights"
assert existing.listed_building == "New Listing"