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

added hubspot company data

Browse source
This commit is contained in:
Jun-te Kim 2026-03-24 15:43:10 +00:00
parent a362e1dd99
commit 29ab9ecfd7
3 changed files with 18 additions and 12 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

5
etl/hubspot/hubspotClient.py
View file

@ -25,6 +25,7 @@ from hubspot.crm.associations.v4.models import ( # type: ignore[reportMissingTy
ForwardPaging as AssociationsPaging, ForwardPaging as AssociationsPaging,
NextPage as AssociationsPagingNext, NextPage as AssociationsPagingNext,
) )
from etl.hubspot.hubspotDataTodB import CompanyData
from backend.app.config import get_settings from backend.app.config import get_settings
@ -223,7 +224,7 @@ class HubspotClient:
return deal, company, listing return deal, company, listing
def get_company_information(self, company_id: str) -> dict[str, str]: def get_company_information(self, company_id: str) -> CompanyData:
companies_api: CompaniesBasicApi = self.client.crm.companies.basic_api # type: ignore[reportUnknownMemberType] companies_api: CompaniesBasicApi = self.client.crm.companies.basic_api # type: ignore[reportUnknownMemberType]
company: HubspotObject = companies_api.get_by_id( # type: ignore[reportUnknownMemberType] company: HubspotObject = companies_api.get_by_id( # type: ignore[reportUnknownMemberType]
@ -233,7 +234,7 @@ class HubspotClient:
], ],
) )
company_info: dict[str, str] = cast(dict[str, str], company.properties) # type: ignore[reportUnknownMemberType] company_info: CompanyData = company.properties # type: ignore[reportUnknownMemberType]
return company_info return company_info
def get_all_pipelines(self) -> list[dict[str, str]]: def get_all_pipelines(self) -> list[dict[str, str]]:

11
etl/hubspot/hubspotDataTodB.py
View file

@ -1,4 +1,4 @@
from backend.app.db.connection import db_session from backend.app.db.connection import db_read_session
from backend.app.db.models.organisation import Organisation from backend.app.db.models.organisation import Organisation
from sqlmodel import select from sqlmodel import select
from datetime import datetime, timezone from datetime import datetime, timezone
@ -15,13 +15,18 @@ class HubspotDataToDb:
pass pass
def read_org_table(self, limit: int = 10): def read_org_table(self, limit: int = 10):
with db_session() as session: with db_read_session() as session:
records = session.exec(select(Organisation).limit(limit)).all() records = session.exec(select(Organisation).limit(limit)).all()
return records return records
def get_org_names(self, limit: int = 10) -> list[str]:
"""Returns a list of organisation names."""
records = self.read_org_table(limit)
return [org.name for org in records if org.name]
def upsert_company(self, company_data: CompanyData) -> Organisation: def upsert_company(self, company_data: CompanyData) -> Organisation:
"""Upserts a company record. Updates if hubspot_company_id exists, otherwise creates new.""" """Upserts a company record. Updates if hubspot_company_id exists, otherwise creates new."""
with db_session() as session: with db_read_session() as session:
hubspot_id = company_data.get("hs_object_id") hubspot_id = company_data.get("hs_object_id")
company_name = company_data.get("name") company_name = company_data.get("name")

14
etl/hubspot/scripts/onboarding/new_organisation.py
View file

@ -1,19 +1,19 @@
from etl.hubspot.hubspotClient import HubspotClient, Companies from etl.hubspot.hubspotClient import HubspotClient, Companies
from etl.hubspot.hubspotDataTodB import HubspotDataToDb from etl.hubspot.hubspotDataTodB import HubspotDataToDb, CompanyData
hubspot = HubspotClient() hubspot = HubspotClient()
dbRead = HubspotDataToDb()
companies_to_add_or_ensure_it_exists = [ companies_to_add_or_ensure_it_exists = [
Companies.THE_GUINESS_PARTNERSHIP, Companies.THE_GUINESS_PARTNERSHIP,
Companies.SOUTHERN_HOUSING_GROUP, Companies.SOUTHERN_HOUSING_GROUP,
] ]
for company in companies_to_add_or_ensure_it_exists: for company in companies_to_add_or_ensure_it_exists:
company_info = hubspot.get_company_information(company.value) company_info: CompanyData = hubspot.get_company_information(company.value)
company_info dbRead.upsert_company(company_info)
break
dbRead = HubspotDataToDb() dbRead = HubspotDataToDb()
names = dbRead.get_org_names()
dbRead.read_org_table() print(f"Organisations in database: {names}")