Automating HubSpot with Python: deduplication and data cleaning

A CRM does not decay all at once. It decays one imported list, one mistyped email and one duplicated contact at a time, until reporting drifts, sales chase the same lead twice, and an automation that should have fired once fires three times. Most teams respond with a periodic clean-up: a painful afternoon every quarter spent merging records by hand. That approach never keeps up, because the CRM keeps filling with new mess the moment the clean-up ends.

The better answer is to treat hygiene as an engineering problem. HubSpot exposes everything you need through its v3 API, and a few hundred lines of Python can turn CRM cleaning from a recurring panic into a scheduled job that runs quietly in the background. This guide walks the full loop: reading contacts efficiently, detecting duplicates, merging them safely, normalising the fields your reporting depends on, scoring overall data quality, and wrapping it all into a repeatable maintenance run.

Why a CRM decays, and why it matters

Before writing code it helps to name the sources of decay, because each one needs a different defence. Contacts arrive through imported lists that were never de-duplicated at source. They arrive through form fills where a visitor types their name three different ways across three visits. They arrive through integrations that create a new record instead of updating an existing one. And they arrive through manual entry, where the same person is logged by two sales reps who never saw each other’s work.

The cost is rarely a single dramatic failure. It is a slow erosion of trust in the numbers. A pipeline report double counts an opportunity because the contact exists twice. A nurture sequence emails the same person under two addresses, which looks careless to the recipient and inflates your send volume. A lifecycle-stage dashboard shows a country called both “UK” and “United Kingdom”, so the regional split is quietly wrong. None of these are catastrophic on their own, but together they are why leadership stops believing the CRM. Automated hygiene is how you earn that trust back and keep it.

Start with a properly scoped client

Everything begins with a private app token and a single, rate-aware HTTP session. HubSpot enforces both per-second and daily limits, so the wrapper below centralises the token, retries on a 429 with exponential backoff, and keeps every call in one place. Centralising this logic is the difference between a script that dies at record four thousand and one that finishes a full sweep unattended.

import time
import requests

class HubSpot:
    def __init__(self, token: str):
        self.base = "https://api.hubapi.com"
        self.s = requests.Session()
        self.s.headers.update({"Authorization": f"Bearer {token}"})

    def _req(self, method, path, **kw):
        for attempt in range(5):
            r = self.s.request(method, self.base + path, **kw)
            if r.status_code == 429:            # rate limited
                time.sleep(2 ** attempt)        # 1, 2, 4, 8, 16 seconds
                continue
            r.raise_for_status()
            return r.json() if r.content else {}
        raise RuntimeError("rate limit not clearing")

Scope the private app to the minimum needed: read and write on contacts, and the merge permission. A token that can only touch what the job needs is a token that cannot cause a surprise elsewhere in the account.

Read the whole book efficiently

To audit a CRM you need every contact, not a page. The list endpoint paginates with an after cursor, and you should request only the properties you actually intend to inspect. Pulling forty properties when you need six is the fastest way to burn through your rate budget for no benefit.

def all_contacts(hs, props):
    after, out = None, []
    while True:
        params = {"limit": 100, "properties": ",".join(props)}
        if after:
            params["after"] = after
        page = hs._req("GET", "/crm/v3/objects/contacts", params=params)
        out.extend(page["results"])
        after = page.get("paging", {}).get("next", {}).get("after")
        if not after:
            return out

For very large books, pull incrementally by filtering on lastmodifieddate through the search endpoint so each run only inspects records that changed since the previous sweep. A full scan is fine weekly; a daily job should look only at the delta.

Detect duplicates on a normalised key

Duplicates rarely match exactly. The same person appears as Jane.Doe@Firm.com and jane.doe@firm.com, or twice with a trailing space, or once with a plus-tagged address. The reliable approach is to build a normalised key and group on it, rather than trusting raw equality.

• Lowercase and trim the email.
• Strip the plus-tag on addresses if your policy treats name+news@ and name@ as one person.
• Fall back to a normalised firstname + lastname + company only when there is genuinely no email.

import re
from collections import defaultdict

def norm_email(e):
    e = (e or "").strip().lower()
    m = re.match(r"([^@]+)@(.+)", e)
    if not m:
        return e
    local, domain = m.groups()
    local = local.split("+")[0]           # drop plus-tag
    return f"{local}@{domain}"

def find_duplicates(contacts):
    groups = defaultdict(list)
    for c in contacts:
        key = norm_email(c["properties"].get("email"))
        if key:
            groups[key].append(c)
    return {k: v for k, v in groups.items() if len(v) > 1}

The normalisation rules are a policy decision, not a technical one. Write them down, agree them with whoever owns the CRM, and keep them in one function so the definition of “the same person” lives in a single, reviewable place.

Merge safely, keeping the richer record

HubSpot’s merge endpoint keeps one contact as the primary and folds the other into it, preserving associations, activity and timeline. The judgement is which record survives. A defensible rule is to keep the record with the most non-empty properties, breaking ties by earliest creation date so the historical timeline stays intact.

def richness(contact):
    return sum(1 for v in contact["properties"].values() if v)

def merge_group(hs, group):
    primary = max(group, key=richness)
    for other in group:
        if other["id"] == primary["id"]:
            continue
        hs._req("POST", "/crm/v3/objects/contacts/merge",
                json={"primaryObjectId": primary["id"],
                      "objectIdToMerge": other["id"]})

Two safeguards matter here. First, run the entire detection pass in a dry run that only logs the merges it would perform, and have a human read that log once before you let it write anything. Merges are effectively irreversible, so the first live run should never be a surprise. Second, never merge across a boundary that carries meaning in your account: two different legal entities that share a support inbox, or a personal and a corporate contact you deliberately track apart, must be excluded by rule rather than folded together by accident.

Normalise the fields reporting depends on

Deduplication removes rows; cleaning fixes fields. The properties that break dashboards are almost always the low-cardinality ones: country, lifecycle stage, industry, phone format. A small mapping table plus the batch update endpoint fixes them in a single pass.

COUNTRY_MAP = {"usa": "United States", "u.s.": "United States",
               "uk": "United Kingdom", "uae": "United Arab Emirates"}

def clean_country(contacts):
    updates = []
    for c in contacts:
        raw = (c["properties"].get("country") or "").strip().lower()
        fixed = COUNTRY_MAP.get(raw)
        if fixed and fixed != c["properties"].get("country"):
            updates.append({"id": c["id"], "properties": {"country": fixed}})
    return updates

def batch_update(hs, updates):
    for i in range(0, len(updates), 100):        # 100 records per batch
        hs._req("POST", "/crm/v3/objects/contacts/batch/update",
                json={"inputs": updates[i:i + 100]})

Batching in chunks of one hundred respects HubSpot’s limits and turns thousands of individual writes into a handful of calls. The same pattern extends to phone formatting, job-title tidying and any other field where a finite set of valid values exists.

Score the data quality so you can see progress

A number that trends over time is what turns hygiene from a chore into a managed metric. A simple completeness-and-validity score, run every sweep and logged, tells you whether the CRM is getting healthier or quietly rotting.

def quality_score(contacts, required=("email", "firstname", "country")):
    if not contacts:
        return 0.0
    filled = 0
    for c in contacts:
        p = c["properties"]
        filled += sum(1 for k in required if p.get(k))
    return round(100 * filled / (len(contacts) * len(required)), 1)

Publish that percentage to a dashboard or a weekly message. When it climbs after each run and then holds steady, you have proof the automation works. When it dips, you have an early warning that a new integration or import is introducing mess, long before anyone notices a broken report.

Turn it into a maintenance job

The final step is orchestration. A single entry point pulls the book, cleans properties, scores quality, detects duplicates, and either reports or acts depending on a flag. Scheduled weekly through a cron job or a serverless function, it keeps the CRM in a steady state rather than letting it drift between manual clean-ups.

def run(hs, apply=False):
    props = ["email", "firstname", "lastname", "company", "country",
             "lifecyclestage", "createdate"]
    contacts = all_contacts(hs, props)

    updates = clean_country(contacts)
    dups = find_duplicates(contacts)
    print(f"score={quality_score(contacts)}  contacts={len(contacts)}  "
          f"normalise={len(updates)}  dup_groups={len(dups)}")

    if apply:
        batch_update(hs, updates)
        for group in dups.values():
            merge_group(hs, group)

Keep the apply flag off for the first few runs and read the log. Once the numbers look right and the dry-run merges are all defensible, flip it on and schedule it. From then on, hygiene is a background process rather than a calendar reminder nobody wants to honour.

Where to run it, and how to watch it

A maintenance job is only as reliable as its schedule and its visibility. For most teams the cleanest home is a small serverless function or a container triggered by a scheduler once a week during quiet hours, when a burst of API calls will not compete with live traffic from forms and integrations. Store the token in a secrets manager rather than the code, and give the function a tight timeout so a stuck run fails loudly instead of hanging.

Observability is what keeps the job trustworthy over months. Every run should emit three things: the data-quality score, the number of records normalised, and the number of duplicate groups merged. Push those to wherever your team already looks, whether that is a logging service, a spreadsheet, or a single scheduled message in a shared channel. The first time the duplicate count spikes, you will know that a new import or integration has started creating mess, and you can fix the source rather than forever cleaning up after it. A hygiene job that reports on itself turns silent decay into an early warning.

A note on consent and compliance

Cleaning data is also a moment where you touch personal information at scale, so the job should respect the same consent boundaries as the rest of your stack. Do not merge a contact who has opted out of communication into one who has not, because the merged record inherits the more permissive state and you may end up emailing someone who asked you to stop. Where your account tracks a legal basis or a subscription status, treat those fields as merge blockers, and log every merge with the surviving record id so the change is auditable. Good hygiene and good governance are the same discipline seen from two angles.

Key takeaways

1. Centralise the token and rate-limit handling in one client so long sweeps finish unattended.
2. Detect duplicates on a normalised key, not raw equality, and merge into the richer record.
3. Always dry-run merges and have a human approve the log before the first live run.
4. Cleaning is mostly about the low-cardinality fields your reporting groups by; fix them with a mapping table and batch updates.
5. Track a data-quality score every run so hygiene becomes a visible, managed metric.
6. Respect consent boundaries as merge blockers, and log every change for audit.

Clean CRM data is not a project you finish; it is a job you automate. Once the maintenance run exists, every downstream report, workflow and forecast inherits its reliability for free, and the quarterly clean-up afternoon never has to happen again.