pip install databar

databar login        # saves your API key
databar agent-guide  # setup notes for Claude Code, Cursor and other agents

import os
from databar import DatabarClient

client = DatabarClient(api_key=os.environ["DATABAR_API_KEY"])

user = client.get_user()
print(user.email, user.plan, user.balance)

from databar import DatabarClient

# api_key is optional: the client reads DATABAR_API_KEY from the environment
with DatabarClient() as client:
    print(client.get_user().balance)

client = DatabarClient(
    api_key=os.environ["DATABAR_API_KEY"],
    base_url="https://api.databar.ai/v1",  # default
    timeout=30.0,            # seconds per HTTP request
    max_poll_attempts=150,   # polling gives up after 150 attempts
    poll_interval_s=2.0,     # 2s between polls, so ~5 minutes total
)

# Search the catalog: 'q' matches name and description
for e in client.list_enrichments(q="email", limit=10):
    print(e.id, e.name, "|", e.data_source, "|", e.price, "credits")

# Then read the full spec: parameters in, fields out
enrichment = client.get_enrichment(1591)
print(enrichment.name, enrichment.price)

for param in enrichment.params or []:
    print("param:", param.name, "required" if param.is_required else "optional")

for field in enrichment.response_fields or []:
    print("returns:", field.name)

# Use the exact param names from get_enrichment(...).params
data = client.run_enrichment_sync(
    enrichment_id=1591,
    params={"linkedin_url": "https://www.linkedin.com/in/janesmith/"},
)

print(data)  # dict of the enrichment's response fields

run = client.run_enrichment(
    enrichment_id=1591,
    params={"linkedin_url": "https://www.linkedin.com/in/janesmith/"},
)
print(run.task_id, run.status)   # "processing"

# Check without blocking
task = client.get_task(run.task_id)
print(task.status)               # processing | completed | no_data | failed | ...

# Or hand it to the poller, which returns the data payload
data = client.poll_task(run.task_id)

inputs = [
    {"linkedin_url": "https://www.linkedin.com/in/janesmith/"},
    {"linkedin_url": "https://www.linkedin.com/in/bobjohnson/"},
    {"linkedin_url": "https://www.linkedin.com/in/sarahlee/"},
]

results = client.run_enrichment_bulk_sync(enrichment_id=1591, params=inputs)

# One element per input, in input order. None means that input returned nothing.
for src, result in zip(inputs, results):
    if result is None:
        print(src["linkedin_url"], "-> no data")
    else:
        print(src["linkedin_url"], "->", result)

for wf in client.list_waterfalls():
    print(wf.identifier, "|", wf.name)
    print("   inputs:", [p.get("name") for p in wf.input_params])
    print("   providers:", [(e.id, e.name, e.price) for e in wf.available_enrichments])
    print("   verifiers:", wf.email_verifiers if wf.is_email_verifying else "none")

wf = client.get_waterfall("email_getter")   # use an identifier from list_waterfalls()

data = client.run_waterfall_sync(
    identifier=wf.identifier,
    params={"first_name": "Jane", "last_name": "Smith", "domain": "acme.com"},
    enrichments=[e.id for e in wf.available_enrichments],  # provider order = list order
    # email_verifier=<id from wf.email_verifiers>

people = [
    {"first_name": "Jane", "last_name": "Smith", "domain": "acme.com"},
    {"first_name": "Bob", "last_name": "Johnson", "domain": "widgets.io"},
]

results = client.run_waterfall_bulk_sync(identifier="email_getter", params=people)

for person, result in zip(people, results):
    status = "no match" if result is None else result
    print(person["first_name"], person["last_name"], "->", status)

from databar import InsertRow, InsertOptions

table = client.create_table(name="Q3 inbound leads", columns=["email", "company"])

client.create_rows(
    table_uuid=table.identifier,
    rows=[
        InsertRow(fields={"email": "jane@acme.com", "company": "Acme"}),
        InsertRow(fields={"email": "bob@widgets.io", "company": "Widgets"}),
    ],
    options=InsertOptions(allow_new_columns=True),
)

# Attach an enrichment and map its params to columns
added = client.add_enrichment(
    table_uuid=table.identifier,
    enrichment_id=1591,
    mapping={"linkedin_url": "LinkedIn URL"},
    launch_strategy="run_on_click",      # or "run_on_update"
)

client.run_table_enrichment(
    table_uuid=table.identifier,
    enrichment_id=str(added.id),
    run_strategy="run_empty",            # run_all | run_empty | run_errors
)

rows = client.get_rows(table.identifier, per_page=100)
for row in rows.data:
    print(row)
if rows.has_next_page:
    more = client.get_rows(table.identifier, page=2)

from databar import (
    DatabarAuthError,
    DatabarValidationError,
    DatabarInsufficientCreditsError,
    DatabarRateLimitError,
    DatabarTaskFailedError,
    DatabarTimeoutError,
    DatabarGoneError,
    DatabarError,
)

try:
    data = client.run_enrichment_sync(1591, {"linkedin_url": url})
except DatabarAuthError:
    raise SystemExit("Check DATABAR_API_KEY")
except DatabarValidationError as e:
    print("Bad input:", e, e.response_body)
except DatabarInsufficientCreditsError:
    print("Out of credits, or the plan does not allow this enrichment")
except DatabarRateLimitError as e:
    print("Rate limited:", e.response_body)   # the client already retried with backoff
except DatabarTaskFailedError as e:
    print("The provider run failed:", e, "task", e.task_id)
except DatabarTimeoutError:
    print("Still running after the poll budget. Keep the task_id and poll later")
except DatabarGoneError:
    print("Results expired (24h). Re-run the enrichment")
except DatabarError as e:
    print("Other API error:", e.status_code, e)

import csv, os
from databar import DatabarClient, DatabarError

client = DatabarClient(api_key=os.environ["DATABAR_API_KEY"])

with open("leads.csv") as f:
    leads = list(csv.DictReader(f))

inputs = [{"first_name": l["first_name"], "last_name": l["last_name"],
           "domain": l["domain"]} for l in leads]

try:
    results = client.run_waterfall_bulk_sync("email_getter", inputs)
except DatabarError as e:
    raise SystemExit(f"Enrichment failed: {e}")

with open("leads_enriched.csv", "w", newline="") as f:
    writer = csv.DictWriter(f, fieldnames=["first_name", "last_name", "domain", "result"])
    writer.writeheader()
    for lead, result in zip(leads, results):
        writer.writerow({**{k: lead[k] for k in ("first_name", "last_name", "domain")},
                         "result": result})

from flask import Flask, request, jsonify
from databar import DatabarClient, DatabarError

app = Flask(__name__)
client = DatabarClient(timeout=5.0, max_poll_attempts=5, poll_interval_s=1.0)

@app.route("/api/enrich-lead", methods=["POST"])
def enrich_lead():
    email = request.json["email"]
    try:
        data = client.run_enrichment_sync(1305, {"email": email})
        return jsonify({"email": email, "enriched": True, "data": data})
    except DatabarError:
        # Never block the form on an enrichment: fall back and enrich later
        return jsonify({"email": email, "enriched": False})

user = client.get_user()
if user.balance < 500:
    notify_ops(f"Databar balance low: {user.balance} credits")