Lists and Bulk enrichment

From version 0.2.0, client.lists manages your Lists and client.bulk runs Bulk enrichment jobs.

bulk.enrich_persons starts a job of up to 25,000 people. Use bulk.wait to wait until the job is completed, cancelled, failed or paused. Once the job is completed or cancelled, bulk.results returns the results CSV as text.

The SDK sends an Idempotency-Key with each new job, generating one if you omit the key. The SDK reuses the key on its own retries, so a retried request never creates a second job. Items keep the API's camelCase field names.

import os

from reversecontact import ReverseContact

with ReverseContact(api_key=os.environ["RC_API_KEY"]) as client:
    created = client.bulk.enrich_persons(
        name="CRM export, November",
        items=[
            {"linkedinUrl": "https://social.com/in/janedoe", "externalId": "crm-001"},
            {"firstName": "John", "lastName": "Smith", "companyDomain": "acme.com", "externalId": "crm-002"},
        ],
        families=["profile", "contact_email"],
    )
    if created.data is None or created.data.job is None or created.data.job.id is None:
        raise RuntimeError("The response is missing the bulk job id")
    job_id = created.data.job.id

    job = client.bulk.wait(job_id)
    if job.data and job.data.links and job.data.links.results:
        with open("results.csv", "w", encoding="utf-8") as file:
            file.write(client.bulk.results(job_id))
    if job.data:
        print(job.data.status, "credits:", job.data.credits.used if job.data.credits else None)

From version 0.3.0, you can use the results before the job settles. bulk.iter_rows yields them one at a time, in the order sent, as soon as each one is ready. While the next item runs, it waits out the Retry-After of the API. It ends with the job, or when the job is paused: read on later with bulk.rows and the last cursor. A cancelled job first finishes the items it had started, within the hour. bulk.rows reads one page (cursor, limit up to 100), with the job status in extras["status"]. Each item is a JobResultLine, exported by reversecontact.types. bulk.results(job_id, format="ndjson") returns the results file as NDJSON, one JobResultLine per line. The client.bulk reference documents each method.

import json

for item in client.bulk.iter_rows(job_id):
    print(item.row, item.input.get("externalId"), item.status, item.credits_consumed)

# Or the whole file once the job settles, one JSON object per line:
ndjson = client.bulk.results(job_id, format="ndjson")
items = [json.loads(line) for line in ndjson.splitlines() if line]

A List keeps its members. Start an import with lists.import_signals, then call lists.operations.wait. The helper returns when the operation's client_action is done, resume or reconcile, so your code can decide what comes next.

If you omit the identifiers, lists.create sends an Idempotency-Key and lists.import_signals sends an operation_id. To resume a paused import after a restart, pass your own operation_id.

import os

from reversecontact import ReverseContact

with ReverseContact(api_key=os.environ["RC_API_KEY"]) as client:
    created = client.lists.create(name="Prospects", entity="persons")
    if created.data is None or created.data.id is None:
        raise RuntimeError("The response is missing the List id")
    list_id = created.data.id

    imported = client.lists.import_signals(
        list_id,
        signals=[{"email": "jane@acme.com", "externalId": "crm-001"}],
        families=["contact_email"],
        max_credits=100,
    )
    if imported.data is None or imported.data.operation_id is None:
        raise RuntimeError("The response is missing the operation id")
    operation = client.lists.operations.wait(list_id, imported.data.operation_id)
    if operation.data:
        print(operation.data.client_action, operation.data.next_step)

    for member in client.lists.members.list_all(list_id):
        print(member)
    csv_text = client.lists.export(list_id)

lists.list_all, lists.members.list_all and bulk.list_all read every page for you. They are free, so they need no max_results.

Choose the right wait method

client.bulk.wait(job_id) returns when the job leaves queued or running, including when the job is failed or paused. Inspect data.status to decide what to do next.

client.lists.operations.wait(list_id, operation_id) returns when data.client_action is no longer poll. Inspect client_action and next_step. Neither helper raises JobFailedError for those returned states.

Both helpers accept timeout (1800 seconds by default), poll_wait (20 by default) and on_poll. If the deadline expires, the helper raises JobTimeoutError with the job or operation id. For a List operation, resume with client.lists.operations.wait(list_id, operation_id).

For bulk.list_all(), pass per_page=100 explicitly. SDK 0.3.0 otherwise starts from 200 items per page before applying max_results, but the API accepts at most 100.

The free list_all() iterators accept an optional max_results if you want to stop early. Without a cap, the iterators read all pages. The client.lists reference and client.bulk reference describe each method, cost and return type.

Return to the Python SDK overview.