junte/Model
1
0
Fork 0
mirror of https://github.com/Hestia-Homes/Model.git synced 2026-07-12 13:29:04 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions

Merge pull request #1515 from Hestia-Homes/feature/hubspot-planning-fields

Browse source 
Add planning/address-profiling fields to HubSpot deal ETL sync
This commit is contained in:
Jun-te Kim 2026-07-08 16:14:01 +01:00 • committed by GitHub
commit 8a04e9fca9
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
4 changed files with 105 additions and 0 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

9
backend/app/db/models/hubspot_deal_data.py
View file

@ -93,6 +93,15 @@ 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)
design_constraints: Optional[str] = Field(default=None)
planning_comments: Optional[str] = Field(default=None)
planning_status: Optional[str] = Field(default=None)
planning_suggested_approach: Optional[str] = Field(default=None)
created_at: Optional[datetime] = Field(
sa_column=Column(
DateTime(timezone=True),

20
etl/hubspot/hubspotDataTodB.py
View file

@ -302,6 +302,16 @@ 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"),
"design_constraints": deal_data.get("design_constraints"),
"planning_comments": deal_data.get("planning_comments"),
"planning_status": deal_data.get("planning_status"),
"planning_suggested_approach": deal_data.get(
"planning_suggested_approach"
),
}.items():
setattr(existing, attr, value)
@ -407,6 +417,16 @@ 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"),
design_constraints=deal_data.get("design_constraints"),
planning_comments=deal_data.get("planning_comments"),
planning_status=deal_data.get("planning_status"),
planning_suggested_approach=deal_data.get(
"planning_suggested_approach"
),
)
def _handle_existing_photo_upload(

8
etl/hubspot/hubspot_deal_differ.py
View file

@ -103,6 +103,14 @@ 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",
"design_constraints": "design_constraints",
"planning_comments": "planning_comments",
"planning_status": "planning_status",
"planning_suggested_approach": "planning_suggested_approach",
}
for hs_field, db_field in FIELD_MAP.items():

68
etl/hubspot/tests/test_hubspot_data_to_db.py
View file

@ -127,3 +127,71 @@ 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",
"design_constraints": "Sloped site",
"planning_comments": "Awaiting decision",
"planning_status": "Submitted",
"planning_suggested_approach": "Full application",
},
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"
assert new_deal.design_constraints == "Sloped site"
assert new_deal.planning_comments == "Awaiting decision"
assert new_deal.planning_status == "Submitted"
assert new_deal.planning_suggested_approach == "Full application"
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",
design_constraints="Old Constraints",
planning_comments="Old Comments",
planning_status="Old Status",
planning_suggested_approach="Old Approach",
)
deal_data = {
"planning_authority": "New Council",
"designated_area": "New Area",
"article_pd_rights": "New Rights",
"listed_building": "New Listing",
"design_constraints": "New Constraints",
"planning_comments": "New Comments",
"planning_status": "New Status",
"planning_suggested_approach": "New Approach",
}
_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"
assert existing.design_constraints == "New Constraints"
assert existing.planning_comments == "New Comments"
assert existing.planning_status == "New Status"
assert existing.planning_suggested_approach == "New Approach"