junte/Model
1
0
Fork 0
mirror of https://github.com/Hestia-Homes/Model.git synced 2026-06-08 11:17:27 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions

its now perfect

Browse source
This commit is contained in:
Jun-te Kim 2026-03-13 14:36:53 +00:00
parent 76dbde602b
commit 3970d70518
1 changed files with 29 additions and 35 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

64
etl/hubspot/hubspotClient.py
View file

@ -2,24 +2,22 @@ import os
from enum import Enum
from typing import Optional, cast
from hubspot.client import Client # type: ignore[reportMissingTypeStubs]
from hubspot.crm.associations import ApiException # type: ignore[reportMissingTypeStubs]
from hubspot.crm.objects import SimplePublicObjectInput # type: ignore[reportMissingTypeStubs]
from hubspot.crm.objects.api.basic_api import BasicApi as ObjectsBasicApi # type: ignore[reportMissingTypeStubs]
from hubspot.crm.deals.api.basic_api import BasicApi as DealsBasicApi # type: ignore[reportMissingTypeStubs]
from hubspot.crm.companies.api.basic_api import BasicApi as CompaniesBasicApi # type: ignore[reportMissingTypeStubs]
from hubspot.crm.products.api.basic_api import BasicApi as ProductsBasicApi # type: ignore[reportMissingTypeStubs]
from hubspot.crm.line_items.api.basic_api import BasicApi as LineItemsBasicApi # type: ignore[reportMissingTypeStubs]
from hubspot.crm.pipelines.api.pipelines_api import PipelinesApi # type: ignore[reportMissingTypeStubs]
from hubspot.crm.pipelines.models import ( # type: ignore[reportMissingTypeStubs]
from hubspot.client import Client
from hubspot.crm.associations import ApiException
from hubspot.crm.objects import SimplePublicObjectInput
from hubspot.crm.objects.api.basic_api import BasicApi as ObjectsBasicApi
from hubspot.crm.deals.api.basic_api import BasicApi as DealsBasicApi
from hubspot.crm.companies.api.basic_api import BasicApi as CompaniesBasicApi
from hubspot.crm.pipelines.api.pipelines_api import PipelinesApi
from hubspot.crm.pipelines.models import (
CollectionResponsePipelineNoPaging as PipelinesResponse,
)
from hubspot.crm.pipelines.models import Pipeline as HubspotPipeline # type: ignore[reportMissingTypeStubs]
from hubspot.crm.pipelines.models import PipelineStage as HubspotPipelineStage # type: ignore[reportMissingTypeStubs]
from hubspot.crm.objects.models import SimplePublicObject as HubspotObject # type: ignore[reportMissingTypeStubs]
from hubspot.crm.associations.v4 import AssociationSpec # type: ignore[reportMissingTypeStubs]
from hubspot.crm.associations.v4.api.basic_api import BasicApi as AssociationsBasicApi # type: ignore[reportMissingTypeStubs]
from hubspot.crm.associations.v4.models import ( # type: ignore[reportMissingTypeStubs]
from hubspot.crm.pipelines.models import Pipeline as HubspotPipeline
from hubspot.crm.pipelines.models import PipelineStage as HubspotPipelineStage
from hubspot.crm.objects.models import SimplePublicObject as HubspotObject
from hubspot.crm.associations.v4 import AssociationSpec
from hubspot.crm.associations.v4.api.basic_api import BasicApi as AssociationsBasicApi
from hubspot.crm.associations.v4.models import (
CollectionResponseMultiAssociatedObjectWithLabelForwardPaging as AssociationsPageResponse,
MultiAssociatedObjectWithLabel as AssociationsResult,
ForwardPaging as AssociationsPaging,
@ -364,17 +362,16 @@ class HubspotClient:
self.logger.error(f"Failed to download file from HubSpot: {e}")
raise
def create_line_item_from_product(self, product_id: str, quantity: int = 1) -> str:
products_api: ProductsBasicApi = self.client.crm.products.basic_api # type: ignore[reportUnknownMemberType]
def create_line_item_from_product(self, product_id: str, quantity: int = 1):
# Fetch product mapping
product: HubspotObject = products_api.get_by_id( # type: ignore[reportUnknownMemberType]
product = self.client.crm.products.basic_api.get_by_id(
product_id, properties=["name", "price", "hs_price"]
)
product_properties: dict[str, str] = cast(dict[str, str], product.properties) # type: ignore[reportUnknownMemberType]
name: Optional[str] = product_properties.get("name")
price: str = product_properties.get("price") or product_properties.get("hs_price") or "0"
name = product.properties.get("name")
price = (
product.properties.get("price") or product.properties.get("hs_price") or "0"
)
# Build line item payload
line_item_input = SimplePublicObjectInput(
@ -388,18 +385,16 @@ class HubspotClient:
}
)
line_items_api: LineItemsBasicApi = self.client.crm.line_items.basic_api # type: ignore[reportUnknownMemberType]
# Create line item
line_item: HubspotObject = line_items_api.create(line_item_input) # type: ignore[reportUnknownMemberType]
return cast(str, line_item.id) # type: ignore[reportUnknownMemberType]
line_item = self.client.crm.line_items.basic_api.create(line_item_input)
return line_item.id
def associate_line_item_to_deal(self, line_item_id: str, deal_id: str) -> None:
def associate_line_item_to_deal(self, line_item_id: str, deal_id: str):
self.logger.info(f"Associating line item {line_item_id} → deal {deal_id}")
association_api: AssociationsBasicApi = self.client.crm.associations.v4.basic_api # type: ignore[reportUnknownMemberType]
association_api = self.client.crm.associations.v4.basic_api
association_api.create( # type: ignore[reportUnknownMemberType]
association_api.create(
"0-3", # to object type
deal_id, # to object id
"line_items", # from object type
@ -414,24 +409,23 @@ class HubspotClient:
def add_product_line_item_to_deal(
self, deal_id: str, product_id: str, quantity: int = 1
) -> str:
):
# Step 1: Create the line item from product mapping
line_item_id: str = self.create_line_item_from_product(product_id, quantity)
line_item_id = self.create_line_item_from_product(product_id, quantity)
# Step 2: Associate the created line item to the deal
self.associate_line_item_to_deal(line_item_id, deal_id)
return line_item_id
def delete_line_item(self, line_item_id: str) -> bool:
def delete_line_item(self, line_item_id: str):
"""
Delete (archive) a line item in HubSpot by its ID.
"""
try:
self.logger.info(f"Deleting line item {line_item_id}...")
line_items_api: LineItemsBasicApi = self.client.crm.line_items.basic_api # type: ignore[reportUnknownMemberType]
line_items_api.archive(line_item_id) # type: ignore[reportUnknownMemberType]
self.client.crm.line_items.basic_api.archive(line_item_id)
self.logger.info(f"Line item {line_item_id} deleted successfully.")
return True