From 3970d70518f1432bc68f2af2532eec63308e5ff4 Mon Sep 17 00:00:00 2001 From: Jun-te Kim Date: Fri, 13 Mar 2026 14:36:53 +0000 Subject: [PATCH] its now perfect --- etl/hubspot/hubspotClient.py | 64 ++++++++++++++++-------------------- 1 file changed, 29 insertions(+), 35 deletions(-) diff --git a/etl/hubspot/hubspotClient.py b/etl/hubspot/hubspotClient.py index 9c1cd31e..b41d71f8 100644 --- a/etl/hubspot/hubspotClient.py +++ b/etl/hubspot/hubspotClient.py @@ -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