# Hubspot API > Implements HubSpot API integration (CRM, Contacts, Companies, Deals, Tickets, using hubspot-api-client Python SDK with OAuth 2.0, private apps, CRM objects, associations, search, and HubSpot REST API patterns. - Skill: `paulpas/hubspot-api` (Agent Skill) - Install (CLI): `npx skillmds@latest add paulpas/hubspot-api` - Raw SKILL.md: https://api.skillmd.com/api/skills/paulpas/hubspot-api/raw - Safety review: pending - Works with: Claude Code, Claude.ai, OpenAI Codex - Category: Integrations & APIs - License: MIT - Author: paulpas (https://skillmd.com/u/paulpas) - Updated: 2026-09-17 - Page: https://skillmd.com/skills/paulpas/hubspot-api --- # HubSpot API Integration Implements production-grade HubSpot integration using the `hubspot-api-client` Python SDK and HubSpot REST API. When loaded, this skill makes the model implement CRUD operations on HubSpot CRM objects (Contacts, Companies, Deals, Tickets, Products, Line Items, Custom Objects), associations between objects, CRM search, HubSpot Forms, Engagements, and HubSpot's Batch API. All implementations follow HubSpot best practices: use `HUBSPOT_ACCESS_TOKEN` from environment for Private Apps, implement OAuth 2.0 refresh token flow for user-facing apps, use v4 API for CRM objects, handle rate limits with exponential backoff, and properly paginate through list results. --- ## Constraints ### MUST DO - Implement structured error responses with consistent format: {error_code, message, details, request_id} - Add rate limiting per client/API key with configurable burst and sustained limits using a token bucket algorithm - Validate all incoming requests against a schema before processing — reject malformed input with clear error messages - Include correlation/request IDs in all log entries for end-to-end request tracing across service boundaries ### MUST NOT DO - Do not expose internal implementation details, stack traces, or database queries in error responses - Avoid accepting unbounded request bodies — set maximum payload sizes and timeout limits - Never trust client-supplied authentication tokens without validation (signature verification, expiration check) - Do not log request/response bodies containing PII, API keys, or other sensitive data ## TL;DR Checklist - [ ] Use `hubspot-api-client` SDK v8+ with Private App access token - [ ] Use `HUBSPOT_ACCESS_TOKEN` environment variable for auth - [ ] OAuth 2.0: Store refresh tokens securely, implement auto-refresh - [ ] Use `crm.contacts.basic_api` for Contact CRUD operations - [ ] Use `crm.companies.basic_api` for Company CRUD operations - [ ] Use `crm.deals.basic_api` for Deal CRUD operations - [ ] Use `crm.associations.v4` API for object associations - [ ] Use `crm.objects.search` for CRM search with filters - [ ] Use Batch API (`crm.contacts.batch_api`) for bulk operations - [ ] Implement pagination with `after` cursor for list endpoints - [ ] Handle 429 rate limit errors with exponential backoff - [ ] Never log or expose access tokens or refresh tokens --- ## When to Use Use this skill when: - Managing HubSpot Contacts, Companies, Deals, or Tickets - Creating associations between CRM objects - Searching CRM data with complex filters - Syncing data between your app and HubSpot CRM - Building integrations with HubSpot marketing automation - Importing/exporting data to/from HubSpot - Automating business processes in HubSpot - Working with custom objects and properties - Managing HubSpot Forms and submissions - Tracking Engagements (calls, emails, meetings, notes) - Processing HubSpot webhooks - Using HubSpot's Batch API for bulk operations --- ## When NOT to Use - For Salesforce-specific CRM — use `coding-salesforce-api` instead - For Marketo marketing automation — use `coding-marketo-api` instead - For Zendesk support tickets — use `coding-zendesk-api` instead - For simple HTTP-only use cases when hubspot-api-client is overkill - When you need real-time events only (consider HubSpot Webhooks + Pub/Sub) - For read-only reporting (use HubSpot Analytics API directly) --- ## Core Workflow 1. **Initialize Client** — Create HubSpot client using `hubspot-api-client`: - Private App: `HubSpot(access_token=os.getenv("HUBSPOT_ACCESS_TOKEN"))` - OAuth: `HubSpot(access_token=access_token)` with token refresh logic **Checkpoint:** Validate connection with `crm.contacts.basic_api.get_page(limit=1)`. 2. **CRM Object Operations** — Perform CRUD on standard objects: - Contacts: `crm.contacts.basic_api.create()`, `read()`, `update()`, `archive()` - Companies: `crm.companies.basic_api.create()`, etc. - Deals: `crm.deals.basic_api.create()`, etc. - Tickets: `crm.tickets.basic_api.create()`, etc. **Checkpoint:** All operations include `properties` parameter to return needed fields. 3. **Associations** — Link objects together using v4 Associations API: - `crm.associations.v4.basic_api.create(contact_to_company, contact_id, company_id, association_type)` - Use predefined association type IDs (e.g., "279" for Contact-to-Company primary) **Checkpoint:** Association operations use correct type IDs and object types. 4. **CRM Search** — Query objects with filters using Search API: - `crm.objects.contacts.search_api.do_search(filter_groups=..., properties=...)` - Support for EQ, NEQ, LT, GT, BETWEEN, CONTAINS_TOKEN, etc. **Checkpoint:** Search queries include `limit` and handle `after` for pagination. 5. **Batch Operations** — Use Batch API for bulk CRUD: - `crm.contacts.batch_api.create(inputs=[...])` - `crm.contacts.batch_api.read(inputs=[...], properties=["email", "firstname"])` - `crm.contacts.batch_api.update(inputs=[...])` - `crm.contacts.batch_api.archive(inputs=[...])` **Checkpoint:** Batch operations respect HubSpot's batch size limits (max 100 per batch). 6. **Handle Limits & Rate Limiting** — Implement exponential backoff: - 429 status code = rate limit exceeded - Check `Retry-After` header for wait time - Use tenacity or backoff library for retries **Checkpoint:** Rate limit handling implemented with jittered exponential backoff. --- ## Implementation Patterns ### Pattern 1: HubSpot Client Initialization (BAD vs GOOD) ```python """HubSpot client initialization patterns. Key concepts: - hubspot-api-client: Official HubSpot Python SDK (v8+) - Private Apps: Access token from HubSpot App Marketplace - OAuth 2.0: User-specific access with refresh tokens - API versions: v3 for most operations, v4 for associations - Rate limits: 100 requests/10 seconds (private apps), varies by tier Environment variables: HUBSPOT_ACCESS_TOKEN: Private App access token (recommended) HUBSPOT_CLIENT_ID: OAuth client ID HUBSPOT_CLIENT_SECRET: OAuth client secret HUBSPOT_REFRESH_TOKEN: OAuth refresh token (for server-to-server) """ from __future__ import annotations import os import json import time import logging from typing import Any, Optional, List, Dict, TypeVar, Callable from dataclasses import dataclass, field from datetime import datetime, timedelta, timezone from functools import wraps logger = logging.getLogger(__name__) # Try to import hubspot-api-client try: import hubspot from hubspot.crm.contacts import ( BasicApi as ContactsBasicApi, BatchApi as ContactsBatchApi, SearchApi as ContactsSearchApi, SimplePublicObjectInput, BatchReadInputSimplePublicObjectId, BatchInputSimplePublicObjectBatchInput, ) from hubspot.crm.companies import ( BasicApi as CompaniesBasicApi, BatchApi as CompaniesBatchApi, SearchApi as CompaniesSearchApi, ) from hubspot.crm.deals import ( BasicApi as DealsBasicApi, BatchApi as DealsBatchApi, SearchApi as DealsSearchApi, ) from hubspot.crm.tickets import ( BasicApi as TicketsBasicApi, ) from hubspot.crm.associations.v4 import ( BasicApi as AssociationsBasicApi, BatchApi as AssociationsBatchApi, ) from hubspot.exceptions import ApiException, NotFoundException, ForbiddenException HUBSPOT_AVAILABLE = True except ImportError: HUBSPOT_AVAILABLE = False logger.warning("hubspot-api-client not installed. Run: pip install hubspot-api-client") # =================================================================== # ❌ BAD — hardcoded token, no error handling, no retry logic # =================================================================== def bad_hubspot_init() -> Any: """❌ BAD: Don't do any of these things.""" if not HUBSPOT_AVAILABLE: raise ImportError("hubspot-api-client required") # ❌ Hardcoded access token! Never commit this! client = hubspot.HubSpot(access_token="pat-na1-abc123-def456-ghi789") # ❌ No validation # ❌ No error handling # ❌ No rate limit handling # ❌ No token refresh for OAuth # ❌ Using deprecated v1/v2 API patterns return client # =================================================================== # ✅ GOOD — env-based config, validation, OAuth refresh, retries # =================================================================== class HubSpotError(Exception): """Base exception for HubSpot integration errors.""" pass class HubSpotAuthError(HubSpotError): """Authentication failed or token invalid/expired.""" pass class HubSpotRateLimitError(HubSpotError): """Rate limit exceeded.""" def __init__(self, message: str, retry_after: Optional[int] = None): super().__init__(message) self.retry_after = retry_after class HubSpotNotFoundError(HubSpotError): """Resource not found (404).""" pass @dataclass class HubSpotTokenStore: """Stores and manages HubSpot OAuth tokens. For production, use a secure database or key vault. This is an in-memory implementation for reference. """ access_token: Optional[str] = None refresh_token: Optional[str] = None expires_at: Optional[datetime] = None # When access_token expires def is_access_token_valid(self) -> bool: """Check if access token is still valid (has > 60 seconds left).""" if not self.access_token or not self.expires_at: return False now = datetime.now(timezone.utc) buffer = timedelta(seconds=60) # Refresh 60 seconds before expiry return self.expires_at > (now + buffer) def set_tokens( self, access_token: str, refresh_token: Optional[str] = None, expires_in_seconds: int = 1800, ) -> None: """Set tokens from OAuth response. Args: access_token: New access token refresh_token: New refresh token (if provided) expires_in_seconds: TTL in seconds (default 30 min = 1800s) """ self.access_token = access_token if refresh_token: self.refresh_token = refresh_token self.expires_at = datetime.now(timezone.utc) + timedelta(seconds=expires_in_seconds) @dataclass class HubSpotConfig: """HubSpot configuration from environment variables. Environment variables: HUBSPOT_ACCESS_TOKEN: Private App access token (simpler, recommended) HUBSPOT_CLIENT_ID: OAuth client ID HUBSPOT_CLIENT_SECRET: OAuth client secret HUBSPOT_REFRESH_TOKEN: OAuth refresh token HUBSPOT_REDIRECT_URI: OAuth redirect URI """ # Private App auth (simpler, server-to-server) access_token: Optional[str] = None # OAuth 2.0 auth (user-facing apps) client_id: Optional[str] = None client_secret: Optional[str] = None refresh_token: Optional[str] = None redirect_uri: Optional[str] = None # Request config timeout: float = 30.0 max_retries: int = 5 initial_retry_delay: float = 1.0 max_retry_delay: float = 60.0 @classmethod def from_env(cls) -> "HubSpotConfig": """Load configuration from environment variables.""" def parse_float(env_var: str, default: float) -> float: val = os.environ.get(env_var) if val is None: return default try: return float(val) except ValueError: return default def parse_int(env_var: str, default: int) -> int: val = os.environ.get(env_var) if val is None: return default try: return int(val) except ValueError: return default return cls( # Private App access_token=os.environ.get("HUBSPOT_ACCESS_TOKEN"), # OAuth client_id=os.environ.get("HUBSPOT_CLIENT_ID"), client_secret=os.environ.get("HUBSPOT_CLIENT_SECRET"), refresh_token=os.environ.get("HUBSPOT_REFRESH_TOKEN"), redirect_uri=os.environ.get("HUBSPOT_REDIRECT_URI"), # Config timeout=parse_float("HUBSPOT_TIMEOUT", 30.0), max_retries=parse_int("HUBSPOT_MAX_RETRIES", 5), ) def is_enabled(self) -> bool: """Check if HubSpot is configured.""" # Private App token if self.access_token: return True # OAuth with refresh token if self.client_id and self.client_secret and self.refresh_token: return True return False def validate(self) -> bool: """Validate configuration. Returns: True if valid Raises: ValueError: If invalid when enabled """ if not self.is_enabled(): logger.info("HubSpot not configured") return True # Private App is valid if token present if self.access_token: return True # OAuth needs client_id, client_secret, and refresh_token if not self.client_id: raise ValueError("HUBSPOT_CLIENT_ID required for OAuth") if not self.client_secret: raise ValueError("HUBSPOT_CLIENT_SECRET required for OAuth") if not self.refresh_token: raise ValueError("HUBSPOT_REFRESH_TOKEN required for OAuth (or use HUBSPOT_ACCESS_TOKEN)") return True class HubSpotClient: """Production-grade HubSpot client with OAuth refresh and retry handling. Features: - Config from environment - Automatic OAuth token refresh - Exponential backoff for rate limits - Helper methods for common operations - Unified error handling """ def __init__(self, config: HubSpotConfig) -> None: self._config = config self._token_store = HubSpotTokenStore() self._client: Any = None # Lazy-loaded HubSpot client self._api_client: Any = None # Underlying API client # If using OAuth with refresh token, initialize store if config.refresh_token: self._token_store.refresh_token = config.refresh_token def _refresh_oauth_token(self) -> str: """Refresh OAuth access token using refresh token. Returns: New access token Raises: HubSpotAuthError: If refresh fails """ import requests if not self._config.client_id or not self._config.client_secret: raise HubSpotAuthError("OAuth client ID/secret not configured") if not self._token_store.refresh_token: raise HubSpotAuthError("No refresh token available") url = "https://api.hubapi.com/oauth/v1/token" data = { "grant_type": "refresh_token", "client_id": self._config.client_id, "client_secret": self._config.client_secret, "refresh_token": self._token_store.refresh_token, } try: response = requests.post(url, data=data, timeout=self._config.timeout) response.raise_for_status() token_data = response.json() access_token = token_data["access_token"] refresh_token = token_data.get("refresh_token") # May be rotated expires_in = token_data.get("expires_in", 1800) # Update token store self._token_store.set_tokens( access_token=access_token, refresh_token=refresh_token, expires_in_seconds=expires_in, ) logger.info("OAuth token refreshed successfully") # If refresh token was rotated, you may want to persist it here # self._persist_refresh_token(self._token_store.refresh_token) return access_token except requests.exceptions.RequestException as e: raise HubSpotAuthError(f"Failed to refresh OAuth token: {e}") from e def _get_valid_access_token(self) -> str: """Get a valid access token, refreshing OAuth if needed. Returns: Valid access token """ # Private App token (simple, no refresh needed) if self._config.access_token: return self._config.access_token # OAuth token - check if valid, refresh if needed if self._token_store.is_access_token_valid(): return self._token_store.access_token # Need to refresh logger.info("OAuth token expired or missing, refreshing...") return self._refresh_oauth_token() def _get_client(self) -> Any: """Get or create the HubSpot client with fresh token. Lazy-loaded and token-refreshing. """ if not HUBSPOT_AVAILABLE: raise ImportError( "hubspot-api-client not installed. " "Run: pip install hubspot-api-client" ) # Always get fresh token (handles OAuth refresh automatically) access_token = self._get_valid_access_token() # Create new client if first time or token changed if self._client is None or self._current_token != access_token: self._client = hubspot.HubSpot(access_token=access_token) self._current_token = access_token logger.info("HubSpot client (re)initialized with fresh token") return self._client @property def _current_token(self) -> str: """Get the token the current client was created with.""" return getattr(self, '_client_token', None) @_current_token.setter def _current_token(self, value: str) -> None: self._client_token = value # =================================================================== # Retry Decorator for Rate Limits # =================================================================== def _execute_with_retry( self, operation: Callable[[], Any], operation_name: str = "operation", ) -> Any: """Execute a HubSpot API operation with retry for rate limits. Implements exponential backoff with jitter. Args: operation: Callable that makes the API call operation_name: Name for logging Returns: API response Raises: HubSpotError: Wrapped API exception HubSpotRateLimitError: If retries exhausted """ import random delay = self._config.initial_retry_delay last_exception: Optional[Exception] = None for attempt in range(self._config.max_retries): try: return operation() except ApiException as e: status = e.status if hasattr(e, 'status') else getattr(e, 'code', None) # 429 = rate limit exceeded if status == 429: last_exception = e # Try to get Retry-After header retry_after = None if hasattr(e, 'headers'): headers = e.headers or {} retry_after_str = headers.get('Retry-After') if retry_after_str: try: retry_after = int(retry_after_str) except ValueError: pass if retry_after: wait_time = float(retry_after) logger.warning( "Rate limited on %s (attempt %d/%d). Retry-After: %ds", operation_name, attempt + 1, self._config.max_retries, retry_after ) else: # Exponential backoff with jitter wait_time = min( delay * (2 ** attempt) + random.uniform(0, 1), self._config.max_retry_delay ) logger.warning( "Rate limited on %s (attempt %d/%d). Waiting %.1fs", operation_name, attempt + 1, self._config.max_retries, wait_time ) time.sleep(wait_time) continue # 401 = unauthorized - token may be expired elif status == 401: # If OAuth, try refreshing token once if self._config.refresh_token and attempt == 0: logger.warning("401 received, attempting token refresh...") self._refresh_oauth_token() self._client = None # Force re-init with new token continue raise HubSpotAuthError(f"Authentication failed: {e}") from e # 404 = not found elif status == 404: raise HubSpotNotFoundError(f"Resource not found: {e}") from e # 403 = forbidden elif status == 403: raise HubSpotAuthError(f"Forbidden (check scopes): {e}") from e # Other errors else: raise HubSpotError(f"HubSpot API error ({status}): {e}") from e except Exception as e: # Non-API exceptions raise HubSpotError(f"HubSpot operation failed: {e}") from e # All retries exhausted raise HubSpotRateLimitError( f"Rate limit retries exhausted after {self._config.max_retries} attempts: {last_exception}" ) from last_exception # =================================================================== # API Access Properties # =================================================================== @property def contacts(self) -> Any: """Contacts API (basic, batch, search).""" client = self._get_client() return client.crm.contacts @property def companies(self) -> Any: """Companies API (basic, batch, search).""" client = self._get_client() return client.crm.companies @property def deals(self) -> Any: """Deals API (basic, batch, search).""" client = self._get_client() return client.crm.deals @property def tickets(self) -> Any: """Tickets API (basic, batch, search).""" client = self._get_client() return client.crm.tickets @property def associations(self) -> Any: """Associations v4 API.""" client = self._get_client() return client.crm.associations.v4 # =================================================================== # Common Operations # =================================================================== def get_contact_by_id( self, contact_id: str, properties: Optional[List[str]] = None, properties_with_history: Optional[List[str]] = None, ) -> Dict[str, Any]: """Get a Contact by ID. Args: contact_id: HubSpot contact ID (numeric string) properties: List of property names to return properties_with_history: Properties with change history Returns: Contact dict with properties """ default_props = ["email", "firstname", "lastname", "phone", "company", "createdate", "lastmodifieddate"] props = properties if properties else default_props def _op(): return self.contacts.basic_api.get_by_id( contact_id=contact_id, properties=props, properties_with_history=properties_with_history, ) result = self._execute_with_retry(_op, f"get_contact({contact_id})") # Convert to dict return self._simple_object_to_dict(result) def get_contact_by_email( self, email: str, properties: Optional[List[str]] = None, ) -> Optional[Dict[str, Any]]: """Get a Contact by email address (using Search API). Args: email: Email address to look up properties: Properties to return Returns: Contact dict or None if not found """ default_props = ["email", "firstname", "lastname", "phone", "company"] props = properties if properties else default_props # Search by email filter_group = { "filters": [ { "propertyName": "email", "operator": "EQ", "value": email.lower(), } ] } def _op(): return self.contacts.search_api.do_search( filter_groups=[filter_group], properties=props, limit=1, ) result = self._execute_with_retry(_op, f"search_contact(email={email})") results = getattr(result, 'results', []) or [] if not results: return None return self._simple_object_to_dict(results[0]) def create_contact( self, properties: Dict[str, Any], ) -> Dict[str, Any]: """Create a new Contact. Args: properties: Contact properties (email, firstname, lastname, etc.) Returns: Created contact dict """ simple_input = SimplePublicObjectInput(properties=properties) def _op(): return self.contacts.basic_api.create(simple_public_object_input=simple_input) result = self._execute_with_retry(_op, "create_contact") return self._simple_object_to_dict(result) def update_contact( self, contact_id: str, properties: Dict[str, Any], ) -> Dict[str, Any]: """Update an existing Contact. Args: contact_id: HubSpot contact ID properties: Properties to update Returns: Updated contact dict """ simple_input = SimplePublicObjectInput(properties=properties) def _op(): return self.contacts.basic_api.update( contact_id=contact_id, simple_public_object_input=simple_input, ) result = self._execute_with_retry(_op, f"update_contact({contact_id})") return self._simple_object_to_dict(result) def delete_contact(self, contact_id: str) -> None: """Delete (archive) a Contact. Args: contact_id: HubSpot contact ID """ def _op(): return self.contacts.basic_api.archive(contact_id=contact_id) self._execute_with_retry(_op, f"delete_contact({contact_id})") logger.info("Deleted contact: %s", contact_id) def list_contacts( self, limit: int = 100, after: Optional[str] = None, properties: Optional[List[str]] = None, ) -> Dict[str, Any]: """List Contacts with pagination. Args: limit: Max per page (max 100) after: Pagination cursor properties: Properties to return Returns: Dict with 'results', 'paging' """ default_props = ["email", "firstname", "lastname", "phone", "company"] props = properties if properties else default_props def _op(): return self.contacts.basic_api.get_page( limit=min(limit, 100), after=after, properties=props, ) result = self._execute_with_retry(_op, f"list_contacts(after={after})") return { "results": [self._simple_object_to_dict(r) for r in (getattr(result, 'results', []) or [])], "paging": getattr(result, 'paging', None), } def search_contacts( self, filter_groups: List[Dict[str, Any]], properties: Optional[List[str]] = None, limit: int = 100, after: Optional[str] = None, sorts: Optional[List[str]] = None, ) -> Dict[str, Any]: """Search Contacts using CRM Search API. Filter operators: - EQ, NEQ: Equals / Not equals - LT, LTE, GT, GTE: Less than / Greater than - BETWEEN: Range (requires highValue) - IN, NOT_IN: Value in list - HAS_PROPERTY, NOT_HAS_PROPERTY: Property exists - CONTAINS_TOKEN, NOT_CONTAINS_TOKEN: String contains Args: filter_groups: List of filter groups (OR between groups, AND within) properties: Properties to return limit: Max per page after: Pagination cursor sorts: Sort properties (e.g., ["-createdate"] for descending) Returns: Dict with 'results', 'total', 'paging' """ default_props = ["email", "firstname", "lastname", "phone", "company", "createdate"] props = properties if properties else default_props def _op(): return self.contacts.search_api.do_search( filter_groups=filter_groups, properties=props, limit=min(limit, 100), after=after, sorts=sorts, ) result = self._execute_with_retry(_op, "search_contacts") return { "total": getattr(result, 'total', 0), "results": [self._simple_object_to_dict(r) for r in (getattr(result, 'results', []) or [])], "paging": getattr(result, 'paging', None), } def batch_create_contacts( self, contacts_list: List[Dict[str, Any]], ) -> List[Dict[str, Any]]: """Create multiple Contacts in batch. Args: contacts_list: List of property dicts Returns: List of created contact dicts """ if not contacts_list: return [] # HubSpot batch limit is 100 batch_size = 100 all_results = [] for i in range(0, len(contacts_list), batch_size): batch = contacts_list[i:i + batch_size] inputs = [{"properties": props} for props in batch] def _op(batch_inputs=inputs): return self.contacts.batch_api.create( batch_input_simple_public_object_input_for_create={"inputs": batch_inputs} ) result = self._execute_with_retry(_op, f"batch_create_contacts[{i}:{i+batch_size}]") results = getattr(result, 'results', []) or [] all_results.extend(self._simple_object_to_dict(r) for r in results) logger.info("Batch created %d contacts", len(results)) return all_results # =================================================================== # Associations # =================================================================== def associate_contact_to_company( self, contact_id: str, company_id: str, association_type: str = "279", # Primary Contact-to-Company ) -> None: """Associate a Contact to a Company. Common association type IDs: - Contact to Company: "279" (primary), "280" (secondary) - Company to Contact: "281" (primary), "282" (secondary) - Deal to Company: "341" - Deal to Contact: "4" - Ticket to Contact: "15" - Ticket to Company: "339" Args: contact_id: Contact ID company_id: Company ID association_type: Association type ID """ def _op(): return self.associations.basic_api.create( from_object_type="contact", from_object_id=contact_id, to_object_type="company", to_object_id=company_id, association_spec=[{"associationCategory": "HUBSPOT_DEFINED", "associationTypeId": int(association_type)}], ) self._execute_with_retry(_op, f"associate_contact_to_company({contact_id}->{company_id})") logger.info("Associated contact %s to company %s", contact_id, company_id) # =================================================================== # Helpers # =================================================================== @staticmethod def _simple_object_to_dict(obj: Any) -> Dict[str, Any]: """Convert a SimplePublicObject to a usable dict. HubSpot SDK returns objects with properties as attributes. This flattens them into a single dict. """ if obj is None: return {} result: Dict[str, Any] = {} # Get ID if hasattr(obj, 'id'): result['id'] = obj.id # Get properties properties = getattr(obj, 'properties', None) or {} if isinstance(properties, dict): result.update(properties) # Get createdAt, updatedAt, archived for attr in ['created_at', 'updated_at', 'archived']: if hasattr(obj, attr): result[attr] = getattr(obj, attr) return result # Global client (lazy-loaded) _global_client: Optional[HubSpotClient] = None def get_hubspot_client() -> HubSpotClient: """Get or create global HubSpot client.""" global _global_client if _global_client is None: config = HubSpotConfig.from_env() _global_client = HubSpotClient(config) return _global_client ``` ### Pattern 2: Common CRM Operations & Search Queries ```python """Common HubSpot CRM operations and search query patterns. HubSpot CRM Search API: - Filter groups: OR between groups, AND within a group - Operators: EQ, NEQ, LT, GT, LTE, GTE, BETWEEN, IN, NOT_IN - String operators: CONTAINS_TOKEN, NOT_CONTAINS_TOKEN, HAS_PROPERTY - Sorting: Property name, prefix with "-" for descending - Pagination: Use 'after' cursor from paging.next.after Property names are lowercase with underscores: - Contacts: email, firstname, lastname, phone, company, createdate - Companies: name, domain, industry, phone, city, state - Deals: dealname, amount, dealstage, closedate, pipeline - Tickets: subject, content, status, priority, createdate """ from __future__ import annotations import logging from typing import Any, Optional, List, Dict from datetime import datetime, timedelta, timezone from dataclasses import dataclass logger = logging.getLogger(__name__) @dataclass class ContactFilters: """Pre-built contact filter templates.""" @staticmethod def by_email(email: str) -> Dict[str, Any]: """Filter by exact email match.""" return { "filters": [ {"propertyName": "email", "operator": "EQ", "value": email.lower()} ] } @staticmethod def by_email_domain(domain: str) -> Dict[str, Any]: """Filter by email domain (contains token).""" return { "filters": [ {"propertyName": "email", "operator": "CONTAINS_TOKEN", "value": f"@{domain}"} ] } @staticmethod def created_in_last_days(days: int) -> Dict[str, Any]: """Filter by creation date within last N days.""" start_ms = int((datetime.now(timezone.utc) - timedelta(days=days)).timestamp() * 1000) return { "filters": [ {"propertyName": "createdate", "operator": "GTE", "value": start_ms} ] } @staticmethod def modified_in_last_hours(hours: int) -> Dict[str, Any]: """Filter by last modified within N hours.""" start_ms = int((datetime.now(timezone.utc) - timedelta(hours=hours)).timestamp() * 1000) return { "filters": [ {"propertyName": "lastmodifieddate", "operator": "GTE", "value": start_ms} ] } @staticmethod def has_phone() -> Dict[str, Any]: """Filter contacts that have a phone number.""" return { "filters": [ {"propertyName": "phone", "operator": "HAS_PROPERTY"} ] } @staticmethod def by_lead_status(status: str) -> Dict[str, Any]: """Filter by hs_lead_status.""" return { "filters": [ {"propertyName": "hs_lead_status", "operator": "EQ", "value": status} ] } @staticmethod def by_lifecycle_stage(stage: str) -> Dict[str, Any]: """Filter by lifecyclestage. Common values: subscriber, lead, marketingqualifiedlead, salesqualifiedlead, opportunity, customer, evangelist, other """ return { "filters": [ {"propertyName": "lifecyclestage", "operator": "EQ", "value": stage} ] } class HubSpotCRMMixin: """Mixin with common CRM operations for HubSpotClient.""" # =================================================================== # Contact Operations # =================================================================== def find_or_create_contact( self, email: str, create_properties: Optional[Dict[str, Any]] = None, ) -> tuple[Dict[str, Any], bool]: """Find a contact by email, or create if not found. Args: email: Email address create_properties: Properties for creation if not found Returns: Tuple of (contact_dict, was_created) """ existing = self.get_contact_by_email(email) if existing: return existing, False # Create new props = dict(create_properties or {}) props["email"] = email # Auto-set firstname/lastname from email if not provided if "firstname" not in props and "@" in email: local = email.split("@")[0] if "." in local: first, last = local.split(".", 1) props["firstname"] = first.capitalize() props["lastname"] = last.capitalize() else: props["firstname"] = local.capitalize() created = self.create_contact(props) return created, True def update_contact_if_changed( self, contact_id: str, new_properties: Dict[str, Any], current_contact: Optional[Dict[str, Any]] = None, ) -> Optional[Dict[str, Any]]: """Update contact only if properties have changed. Reduces unnecessary API calls. Args: contact_id: Contact ID new_properties: P …(truncated)