# Keep HubSpot company records current A weekly Python job that enriches every HubSpot company with a domain: employee count, size band, industry, headquarters and its LinkedIn page. ## What it does Once a week, the script reads every company in your HubSpot account, looks up the ones with a domain, and writes back what it finds. Companies it enriched in the last seven days are skipped, so a weekly run only pays for what's due. | HubSpot property | From HappyFetch | |---|---| | Number of Employees (`numberofemployees`) | `employeeCount` | | City, Country/Region (`city`, `country`) | `headquarter.city`, `headquarter.country` | | LinkedIn Company Page (`linkedin_company_page`) | `linkedInUrl` | | Year Founded (`founded_year`) | `foundedOn.year` | | Industry (HappyFetch) (`happyfetch_industry`) | `industry` | | Size band (HappyFetch) (`happyfetch_size_band`) | `employeeCountRange` | | Enriched by HappyFetch (`happyfetch_enriched_at`) | when the script last updated the company | HubSpot's own Industry property only accepts values from its fixed list, so the industry goes into a text property of its own. The script creates the three HappyFetch properties on its first run. It never blanks out a value: a field we don't know leaves HubSpot's value as it is. ## What you need - A HubSpot private app token in `HUBSPOT_TOKEN`, with the scopes `crm.objects.companies.read`, `crm.objects.companies.write` and `crm.schemas.companies.write`. Create one under **Settings > Integrations > Private Apps**. - An API key in `HAPPYFETCH_API_KEY`. New accounts get 200 credits. - Python 3.10 or newer, and `pip install requests`. ## The script ```python """Keep HubSpot company records current with HappyFetch: size, industry, headquarters, LinkedIn page. Usage: python hubspot_enrich.py Needs HUBSPOT_TOKEN, a private app token with the scopes crm.objects.companies.read, crm.objects.companies.write and crm.schemas.companies.write, and HAPPYFETCH_API_KEY. Run it weekly: companies it enriched in the last FRESH_DAYS days are skipped. """ import datetime import os import time import uuid from concurrent.futures import ThreadPoolExecutor import requests HAPPYFETCH = "https://api.happyfetch.net" HUBSPOT = "https://api.hubapi.com" HF = {"Authorization": f"Bearer {os.environ['HAPPYFETCH_API_KEY']}"} HS = {"Authorization": f"Bearer {os.environ['HUBSPOT_TOKEN']}"} FRESH_DAYS = 7 # HubSpot's own industry property only takes values from its fixed list, so the industry and the size # band go into text properties of their own, created on the first run. OWN_PROPERTIES = [ {"name": "happyfetch_industry", "label": "Industry (HappyFetch)", "type": "string", "fieldType": "text"}, {"name": "happyfetch_size_band", "label": "Size band (HappyFetch)", "type": "string", "fieldType": "text"}, {"name": "happyfetch_enriched_at", "label": "Enriched by HappyFetch", "type": "datetime", "fieldType": "date"}, ] def ensure_properties() -> None: for prop in OWN_PROPERTIES: res = requests.post(f"{HUBSPOT}/crm/v3/properties/companies", headers=HS, json={**prop, "groupName": "companyinformation"}, timeout=30) if res.status_code not in (201, 409): # 409: it exists from an earlier run res.raise_for_status() def companies(): """Every company in HubSpot, 100 at a time, with its domain and when we last enriched it.""" params = {"limit": 100, "properties": "domain,happyfetch_enriched_at"} while True: res = requests.get(f"{HUBSPOT}/crm/v3/objects/companies", headers=HS, params=params, timeout=30) res.raise_for_status() page = res.json() yield from page["results"] after = page.get("paging", {}).get("next", {}).get("after") if not after: return params["after"] = after def due(company: dict, now: datetime.datetime) -> bool: props = company["properties"] if not props.get("domain"): return False last = props.get("happyfetch_enriched_at") return not last or now - datetime.datetime.fromisoformat(last.replace("Z", "+00:00")) > datetime.timedelta(days=FRESH_DAYS) def lookup(domain: str): key = str(uuid.uuid4()) # one key per domain, reused on every retry: a retry never pays twice for attempt in range(6): try: res = requests.post(f"{HAPPYFETCH}/v1/domain", headers={**HF, "Idempotency-Key": key}, json={"domain": domain}, timeout=90) answer = res.json() except (requests.RequestException, ValueError): time.sleep(2**attempt) # a timeout or a dropped connection continue if res.ok: return answer["data"], None error = answer["error"] if not error["retryable"]: return None, error["code"] time.sleep(error.get("retry_after_seconds") or 2**attempt) return None, "gave_up" def properties(company: dict, stamp: str) -> dict: hq = company["headquarter"] or {} size = company["employeeCountRange"] or {} founded = (company["foundedOn"] or {}).get("year") values = { "numberofemployees": company["employeeCount"], "city": hq.get("city"), "country": hq.get("country"), "linkedin_company_page": company["linkedInUrl"], "founded_year": str(founded) if founded else None, "happyfetch_industry": company["industry"], "happyfetch_size_band": f"{size['start']}-{size['end'] or ''}" if size else None, "happyfetch_enriched_at": stamp, } return {k: v for k, v in values.items() if v is not None} # never blank out what HubSpot already has def main() -> None: ensure_properties() now = datetime.datetime.now(datetime.timezone.utc) stamp = now.isoformat(timespec="seconds").replace("+00:00", "Z") todo = [c for c in companies() if due(c, now)] print(f"{len(todo):,} companies to enrich") updates, found, missing, invalid, failed = [], 0, 0, 0, 0 with ThreadPoolExecutor(max_workers=8) as pool: for company, (data, error) in zip(todo, pool.map(lambda c: lookup(c["properties"]["domain"]), todo)): if data: updates.append({"id": company["id"], "properties": properties(data, stamp)}) found += 1 elif error == "not_found": # no company at that domain: don't ask again this week updates.append({"id": company["id"], "properties": {"happyfetch_enriched_at": stamp}}) missing += 1 elif error == "invalid_domain": # not a website: never charged, worth fixing in HubSpot invalid += 1 else: failed += 1 # left as it was, so the next run tries again for i in range(0, len(updates), 100): # HubSpot takes 100 updates per call res = requests.post(f"{HUBSPOT}/crm/v3/objects/companies/batch/update", headers=HS, json={"inputs": updates[i:i + 100]}, timeout=60) res.raise_for_status() print(f"Updated {found:,} companies. {missing:,} domains have no company, {invalid:,} aren't websites, " f"and {failed:,} failed and wait for the next run.") if __name__ == "__main__": main() ``` Run it with `python hubspot_enrich.py`, then schedule it weekly: a cron job, a GitHub Actions schedule or any scheduler you already use. ## How it handles failures - **Retries never pay twice.** Each lookup has its own idempotency key, reused on every retry. - **A domain with no company** is an answer: the script stamps the company so it isn't asked again that week. It costs 4 credits, and nothing when the answer comes from our cache. - **A domain that isn't a website** is never charged. The script counts it so you can fix it in HubSpot. - **A call we couldn't complete** is refunded in the same request and left alone, so the next run tries again. ## What it costs A weekly run looks up each company once a week. Most companies change slowly, so set `FRESH_DAYS` to 30 for a monthly refresh at a quarter of the cost. | Lookups a month | Smallest pack (Starter) | Biggest pack (Business) | |---|---|---| | 1,000 lookups by website domain | $2.00 | $1.00 | | 10,000 lookups by website domain | $20.00 | $10.00 | | 100,000 lookups by website domain | $200.00 | $100.00 | Every lookup fetched fresh, at 4 credits each. Lookups our cache answers cost half. A confirmed not-found costs the same, and nothing from our cache. Starter: $10 for 20,000 credits. Business: $500 for 2,000,000 credits. Credits never expire.