"""HubSpot CRM connector.
Syncs contacts, deals, and companies from HubSpot REST API v3.
Paginates through the CRM API, merges (upserts) records into local DB,
and returns actual record counts.
"""
from __future__ import annotations
import logging
import time
from datetime import datetime, timezone
from typing import Any, Dict, List, Optional
import requests
from . import OAuthConnector, register_connector, _REGISTRY
logger = logging.getLogger(__name__)
# -- Registration metadata -------------------------------------------------
register_connector(
"hubspot",
{
"service": "hubspot",
"name": "HubSpot",
"category": "crm",
"description": "CRM data — contacts, deals, pipeline stages.",
"auth_type": "oauth2",
"auth_fields": ["access_token", "refresh_token"],
"capabilities": ["contacts", "deals", "companies", "tickets"],
"rate_limit": "10 req/s (free), 30 req/s (paid)",
"docs_url": "https://developers.hubspot.com/docs/api/overview",
},
)
class HubSpotConnector(OAuthConnector):
"""HubSpot CRM integration — pulls contacts, deals, companies."""
_SERVICE = "hubspot"
_BASE_URL = "https://api.hubapi.com"
PAGE_LIMIT = 100
# -- OAuth 2.0 ------------------------------------------------------------
OAUTH_AUTHORIZE_URL="https://app.hubspot.com/oauth/authorize"
OAUTH_TOKEN_URL="https://api.hubapi.com/oauth/v3/token"
OAUTH_SCOPES=[
"crm.objects.contacts.read",
"crm.objects.deals.read",
"crm.objects.companies.read",
]
@property
def _oauth_client_id(self) -> str:
from flask import current_app
return current_app.config.get("OAUTH_HUBSPOT_CLIENT_ID", "")
@property
def _oauth_client_secret(self) -> str:
from flask import current_app
return current_app.config.get("OAUTH_HUBSPOT_CLIENT_SECRET", "")
def oauth_authorize_url(self, state: str) -> str:
"""Build the HubSpot OAuth authorization URL."""
from urllib.parse import urlencode
params = {
"client_id": self._oauth_client_id,
"redirect_uri": self._get_redirect_uri(),
"scope": " ".join(self.OAUTH_SCOPES),
"state": state,
}
return f"{self.OAUTH_AUTHORIZE_URL}?{urlencode(params)}"
def exchange_code_for_tokens(self, code: str, **kwargs) -> Dict[str, Any]:
"""Exchange authorization code for access/refresh tokens via HubSpot."""
client_id = self._oauth_client_id
client_secret = self._oauth_client_secret
if not client_id or not client_secret:
raise ValueError(
"HubSpot OAuth credentials not configured. "
"Set OAUTH_HUBSPOT_CLIENT_ID and OAUTH_HUBSPOT_CLIENT_SECRET."
)
response = requests.post(
self.OAUTH_TOKEN_URL,
data={
"grant_type": "authorization_code",
"code": code,
"redirect_uri": self._get_redirect_uri(),
"client_id": client_id,
"client_secret": client_secret,
},
headers={"Content-Type": "application/x-www-form-urlencoded"},
timeout=15,
)
response.raise_for_status()
data = response.json()
return {
"access_token": data.get("access_token", ""),
"refresh_token": data.get("refresh_token", ""),
"expires_in": data.get("expires_in", ""),
"token_type": data.get("token_type", "Bearer"),
"scope": data.get("scope", ""),
}
def refresh_access_token(self, refresh_token: str) -> Dict[str, str]:
"""Refresh expired access token via HubSpot token endpoint.
Note: HubSpot personal access tokens don't expire, but OAuth app tokens do.
"""
client_id = self._oauth_client_id
client_secret = self._oauth_client_secret
if not client_id or not client_secret:
raise ValueError("HubSpot OAuth credentials not configured.")
response = requests.post(
self.OAUTH_TOKEN_URL,
data={
"grant_type": "refresh_token",
"refresh_token": refresh_token,
"client_id": client_id,
"client_secret": client_secret,
},
headers={"Content-Type": "application/x-www-form-urlencoded"},
timeout=15,
)
response.raise_for_status()
data = response.json()
return {
"access_token": data.get("access_token", ""),
"refresh_token": data.get("refresh_token", ""),
"expires_in": data.get("expires_in", ""),
"token_type": data.get("token_type", "Bearer"),
}
# -- connect / disconnect -----------------------------------------------
def connect(self) -> Dict[str, Any]:
"""Validate the HubSpot access token by fetching account info."""
self._log(event_type="connect_attempt", status="pending")
start = time.monotonic()
try:
access_token = self.config.get("access_token", "")
if not access_token:
raise ValueError("HubSpot access_token is required")
response = _api_get(
f"{self._BASE_URL}/crm/v3/owners/me",
{"access_token": access_token},
)
self._connected = True
duration_ms = int((time.monotonic() - start) * 1000)
self._log(
event_type="connect_success",
status="success",
duration_ms=duration_ms,
details={"account_id": response.get("accountId")},
)
return {
"status": "connected",
"service": "hubspot",
"account_id": response.get("accountId"),
"duration_ms": duration_ms,
}
except Exception as exc:
self._log(event_type="connect_error", status="error", error_message=str(exc))
return {"status": "error", "error": str(exc)}
def disconnect(self) -> Dict[str, Any]:
self._connected = False
self._log(event_type="disconnect", status="success")
return {"status": "disconnected", "service": "hubspot"}
# -- sync ----------------------------------------------------------------
def sync(self) -> Dict[str, Any]:
"""Sync contacts, deals, and companies from HubSpot."""
self._log(event_type="sync_start", status="pending")
start = time.monotonic()
total_records = 0
try:
results: Dict[str, Any] = {}
# Sync contacts first (deals reference them)
contacts = self._sync_contacts()
total_records += contacts
results["contacts"] = contacts
# Sync companies
companies = self._sync_companies()
total_records += companies
results["companies"] = companies
# Sync deals (they reference contacts and companies)
deals = self._sync_deals()
total_records += deals
results["deals"] = deals
duration_ms = int((time.monotonic() - start) * 1000)
self._log(
event_type="sync_complete",
status="success",
record_count=total_records,
duration_ms=duration_ms,
details=results,
)
return {
"status": "success",
"record_count": total_records,
"duration_ms": duration_ms,
"details": results,
"last_sync_at": datetime.now(timezone.utc).isoformat(),
}
except Exception as exc:
duration_ms = int((time.monotonic() - start) * 1000)
self._log(
event_type="sync_error",
status="error",
record_count=total_records,
duration_ms=duration_ms,
error_message=str(exc),
)
return {
"status": "error",
"error": str(exc),
"record_count": total_records,
}
# -- Paginated fetch helper ---------------------------------------------
def _fetch_all(self, object_type: str, properties: Optional[List[str]] = None) -> List[Dict[str, Any]]:
"""Paginate through HubSpot CRM API and return all objects.
Uses the after= cursor for pagination with limit={PAGE_LIMIT} per page.
"""
access_token = self.config.get("access_token", "")
if not access_token:
raise ValueError("No access_token configured for HubSpot")
all_items: List[Dict[str, Any]] = []
after: Optional[str] = None
params = {"access_token": access_token, "limit": str(self.PAGE_LIMIT)}
if properties:
params["properties"] = ",".join(properties)
while True:
url = f"{self._BASE_URL}/crm/v3/objects/{object_type}"
self._check_rate_limit()
data = _api_get(url, params.copy())
results = data.get("results", [])
if not results:
break
all_items.extend(results)
logger.info("HubSpot %s: fetched %d (total %d)", object_type, len(results), len(all_items))
paging = data.get("paging", {})
next_page = paging.get("next", {})
after = next_page.get("after")
if not after:
break
params["after"] = after
# Small pause between pages to stay within rate limits
time.sleep(0.1)
logger.info("HubSpot %s: total %d records", object_type, len(all_items))
return all_items
# -- Merge (upsert) helper -----------------------------------------------
@staticmethod
def _merge(model, company_id: str, external_id: str, defaults: Dict[str, Any]) -> int:
"""Find or create a record by (company_id, external_id) and update fields.
Returns 1 if a record was inserted or updated.
"""
from app.models import db
record = model.query.filter_by(
company_id=company_id, external_id=external_id
).first()
if record:
for key, value in defaults.items():
setattr(record, key, value)
else:
record = model(
company_id=company_id,
external_id=external_id,
**defaults
)
db.session.add(record)
return 1
# -- Contact sync --------------------------------------------------------
def _sync_contacts(self) -> int:
"""Pull contacts from HubSpot, merge into local CrmContact table."""
from app.models import db, CrmContact
items = self._fetch_all(
"contacts",
properties=[
"email", "firstname", "lastname", "phone", "company",
"lifecyclestage", "hs_object_id", "hubspot_owner_id",
],
)
count = 0
for raw in items:
props = {}
for prop in raw.get("properties", {}).values():
if isinstance(prop, dict):
props[prop.get("name", "")] = prop.get("value", "")
hs_id = props.get("hs_object_id", raw.get("id", ""))
if not hs_id:
continue
count += self._merge(
CrmContact,
self.company_id,
hs_id,
{
"email": props.get("email", ""),
"first_name": props.get("firstname", ""),
"last_name": props.get("lastname", ""),
"phone": props.get("phone", ""),
"company_name": props.get("company", ""),
"lifecycle_stage": props.get("lifecyclestage", ""),
"hubspot_owner_id": props.get("hubspot_owner_id", ""),
"properties_json": raw.get("properties", {}),
},
)
db.session.commit()
logger.info("HubSpot contacts synced: %d records merged", count)
return count
# -- Deal sync -----------------------------------------------------------
def _sync_deals(self) -> int:
"""Pull deals from HubSpot, merge into local CrmDeal table."""
from app.models import db, CrmDeal, CrmContact
items = self._fetch_all(
"deals",
properties=[
"dealname", "amount", "dealstage", "pipeline", "probability",
"closedate", "hs_object_id", "hubspot_owner_id",
"associatedvid", "associatedcompanyid",
],
)
count = 0
for raw in items:
props = {}
for prop in raw.get("properties", {}).values():
if isinstance(prop, dict):
props[prop.get("name", "")] = prop.get("value", "")
hs_id = props.get("hs_object_id", raw.get("id", ""))
if not hs_id:
continue
# Parse expected_close_date
expected_close_str = props.get("closedate", "")
expected_close_date: Optional[datetime] = None
if expected_close_str:
try:
expected_close_date = datetime.fromisoformat(expected_close_str.replace("Z", "+00:00"))
except (ValueError, TypeError):
pass
# Parse amount
amount_str = props.get("amount", "")
amount: Optional[float] = None
if amount_str:
try:
amount = float(amount_str)
except (ValueError, TypeError):
pass
# Parse probability
prob_str = props.get("probability", "")
probability: Optional[float] = None
if prob_str:
try:
probability = float(prob_str) / 100.0 # HubSpot uses 0-100
except (ValueError, TypeError):
pass
# Try to link to contact
associated_vid = props.get("associatedvid", "")
contact_id: Optional[str] = None
if associated_vid:
contact = CrmContact.query.filter_by(
company_id=self.company_id,
external_id=associated_vid
).first()
if contact:
contact_id = contact.id
count += self._merge(
CrmDeal,
self.company_id,
hs_id,
{
"name": props.get("dealname", ""),
"amount": amount,
"stage": props.get("dealstage", ""),
"pipeline": props.get("pipeline", ""),
"probability": probability,
"expected_close_date": expected_close_date,
"contact_id": contact_id,
"deal_owner_id": props.get("hubspot_owner_id", ""),
"properties_json": raw.get("properties", {}),
},
)
db.session.commit()
logger.info("HubSpot deals synced: %d records merged", count)
return count
# -- Company sync --------------------------------------------------------
def _sync_companies(self) -> int:
"""Pull companies from HubSpot, merge into local CrmCompany table."""
from app.models import db, CrmCompany
items = self._fetch_all(
"companies",
properties=[
"name", "domain", "industry", "numberofemployees",
"hs_object_id",
],
)
count = 0
for raw in items:
props = {}
for prop in raw.get("properties", {}).values():
if isinstance(prop, dict):
props[prop.get("name", "")] = prop.get("value", "")
hs_id = props.get("hs_object_id", raw.get("id", ""))
if not hs_id:
continue
# Parse num_employees
emp_str = props.get("numberofemployees", "")
num_employees: Optional[int] = None
if emp_str:
try:
num_employees = int(emp_str)
except (ValueError, TypeError):
pass
count += self._merge(
CrmCompany,
self.company_id,
hs_id,
{
"name": props.get("name", ""),
"domain": props.get("domain", ""),
"industry": props.get("industry", ""),
"num_employees": num_employees,
"properties_json": raw.get("properties", {}),
},
)
db.session.commit()
logger.info("HubSpot companies synced: %d records merged", count)
return count
# -- status --------------------------------------------------------------
def status(self) -> Dict[str, Any]:
if not self._connected:
return {
"service": "hubspot",
"connected": False,
"config_present": bool(self.config.get("access_token")),
}
try:
response = _api_get(
f"{self._BASE_URL}/crm/v3/owners/me",
{"access_token": self.config["access_token"]},
)
return {
"service": "hubspot",
"connected": True,
"account_id": response.get("accountId"),
"last_sync_at": self.config.get("last_sync_at"),
}
except Exception as exc:
return {"service": "hubspot", "connected": False, "error": str(exc)}
# -- Register in the framework registry -------------------------------------
_REGISTRY["hubspot"] = HubSpotConnector
# -- Internal API helpers ---------------------------------------------------
def _api_get(url: str, params: Dict[str, str]) -> Dict[str, Any]:
"""Make a GET request to HubSpot API with proper error handling."""
import requests
resp = requests.get(url, params=params, timeout=30)
# Capture rate-limit headers
if resp.status_code == 429:
retry_after = resp.headers.get("Retry-After")
reset = resp.headers.get("X-RateLimit-Reset")
logger.warning(
"HubSpot rate limited: status=429, Retry-After=%s, X-RateLimit-Reset=%s",
retry_after, reset,
)
if retry_after:
try:
time.sleep(int(retry_after))
except ValueError:
time.sleep(2)
resp.raise_for_status()
return resp.json()