Databar Python SDK

Official Python SDK and CLI for Databar.ai — run data enrichments, waterfall lookups, flows, and manage tables via api.databar.ai/v1.

Installation

pip install databar

Requires Python 3.9+.

Authentication

Get your API key from databar.ai → Integrations.

Option 1 — CLI (recommended):

databar login

Opens the browser to sign in and saves your workspace API key to ~/.databar/config. For CI or agents, use databar login --api-key <key> instead.

Option 2 — Environment variable:

export DATABAR_API_KEY=your-key-here

Option 3 — In code:

from databar import DatabarClient
client = DatabarClient(api_key="your-key-here")

Python SDK

Quick start

from databar import DatabarClient

client = DatabarClient()  # reads DATABAR_API_KEY from env

# Check your balance
user = client.get_user()
print(f"Balance: {user.balance} credits")

# Find enrichments
enrichments = client.list_enrichments(q="linkedin")
for e in enrichments:
    print(f"  [{e.id}] {e.name} — {e.price} credits")

# Run a single enrichment (submit + poll in one call)
result = client.run_enrichment_sync(123, {"email": "alice@example.com"})
print(result)

# Run a waterfall
result = client.run_waterfall_sync("email_getter", {"linkedin_url": "https://linkedin.com/in/alice"})
print(result)

Enrichments

# List all enrichments
enrichments = client.list_enrichments()

# Search enrichments
enrichments = client.list_enrichments(q="phone")

# Paginated list (returns EnrichmentListResponse)
page = client.list_enrichments(page=1, limit=50, category="Company Data")
for e in page.items:
    print(f"  [{e.id}] {e.name} — {e.price} credits")
if page.has_next_page:
    page2 = client.list_enrichments(page=2, limit=50)

# Get full details (params, response fields)
enrichment = client.get_enrichment(123)
for param in enrichment.params:
    print(f"  {param.name} (required={param.is_required}): {param.description}")

# Run single enrichment (async — returns task)
task = client.run_enrichment(123, {"email": "alice@example.com"})
data = client.poll_task(task.task_id)

# Run single enrichment (sync convenience wrapper)
data = client.run_enrichment_sync(123, {"email": "alice@example.com"})

# Run with pagination (for list-style enrichments)
data = client.run_enrichment_sync(123, {"query": "CEO"}, pages=3)

# Bulk run — results are aligned to inputs: one element per input, in input
# order, with None for inputs that returned no data (len(data) == len(inputs)).
data = client.run_enrichment_bulk_sync(123, [
    {"email": "alice@example.com"},
    {"email": "bob@example.com"},
])

# Get choices for a select parameter
choices = client.get_param_choices(123, "country", q="united")
for choice in choices.items:
    print(f"  {choice.id}: {choice.name}")

Waterfalls

# List waterfalls
waterfalls = client.list_waterfalls()

# Run a waterfall (tries all providers in sequence)
result = client.run_waterfall_sync(
    "email_getter",
    {"linkedin_url": "https://linkedin.com/in/alice"},
)

# Run with specific providers only
result = client.run_waterfall_sync(
    "email_getter",
    {"linkedin_url": "https://linkedin.com/in/alice"},
    enrichments=[10, 11],  # provider IDs
)

# Bulk waterfall
results = client.run_waterfall_bulk_sync(
    "email_getter",
    [{"linkedin_url": url} for url in urls],
)

Flows

# List saved flows
flows = client.list_flows()

# Inspect a flow's declared inputs
flow = client.get_flow("flow-uuid")
for inp in flow.inputs:
    print(f"  {inp.id} (required={inp.required}): {inp.description}")

# Run a flow (submit + poll in one call).
# inputs maps each flow input id → value.
result = client.run_flow_sync("flow-uuid", {"email": "alice@example.com"})

# Or submit and poll manually
task = client.run_flow("flow-uuid", {"email": "alice@example.com"})
result = client.poll_task(task.task_id)

Tables

# List and manage tables
tables = client.list_tables()
table = client.create_table(name="My Leads", columns=["email", "name", "company"])
client.rename_table(table.identifier, "Leads 2026")
client.delete_table(table.identifier)

# Columns
columns = client.get_columns(table.identifier)
col = client.create_column(table.identifier, "Phone", type="text")
client.rename_column(table.identifier, col.identifier, "Mobile")
client.delete_column(table.identifier, col.identifier)

# Get rows with optional server-side filter
import json
data = client.get_rows(table.identifier, page=1, per_page=500)
filtered = client.get_rows(
    table.identifier,
    filter=json.dumps({"company": {"contains": "OpenAI"}}),
)

# Insert rows (auto-batched at 50)
from databar import InsertRow, InsertOptions, DedupeOptions

rows = [InsertRow(fields={"email": e, "name": n}) for e, n in leads]
response = client.create_rows(
    table.identifier,
    rows,
    options=InsertOptions(
        allow_new_columns=True,
        dedupe=DedupeOptions(enabled=True, keys=["email"]),
    ),
)
print(f"Created: {len([r for r in response.results if r.action == 'created'])}")

# Update rows by UUID
from databar import BatchUpdateRow
rows = [BatchUpdateRow(id=row_id, fields={"name": "Updated Name"})]
client.patch_rows(table.identifier, rows)

# Upsert rows by key column
from databar import UpsertRow
rows = [UpsertRow(key={"email": "alice@example.com"}, fields={"name": "Alice"})]
client.upsert_rows(table.identifier, rows)

# Delete specific rows
client.delete_rows(table.identifier, ["row-uuid-1", "row-uuid-2"])

Enrichments on tables

# Add an enrichment (with column mapping)
result = client.add_enrichment(
    table.identifier,
    enrichment_id=123,
    mapping={
        "email": {"type": "mapping", "value": "email_col"},   # from column
        "country": {"type": "simple", "value": "US"},          # static value
    },
    launch_strategy="run_on_click",  # or "run_on_update"
)
print(f"Added enrichment #{result.id}: {result.enrichment_name}")

# Run it (run_strategy: run_all | run_empty | run_errors)
status = client.run_table_enrichment(
    table.identifier,
    enrichment_id=str(result.id),
    run_strategy="run_empty",
)
print(f"Processing {status.processing_rows} rows")

Waterfalls on tables

# Add a waterfall
wf = client.add_waterfall(
    table.identifier,
    waterfall_identifier="email_getter",
    enrichments=[833, 966],
    mapping={"first_name": "first_name", "company": "company"},
    email_verifier=10,
)
print(f"Added waterfall #{wf.id}: {wf.waterfall_name}")

# List installed waterfalls
installed = client.get_table_waterfalls(table.identifier)

# Run the waterfall
client.run_table_enrichment(table.identifier, enrichment_id=str(wf.id))

Exporters

# List available exporters
exporters = client.list_exporters()              # plain list
page = client.list_exporters(page=1, limit=50)  # paginated envelope

# Get exporter details (params, authorization info)
detail = client.get_exporter(exporters[0].id)
print(detail.params, detail.authorization.required)

# Add an exporter to a table
result = client.add_exporter(
    table.identifier,
    exporter_id=detail.id,
    mapping={"email": {"type": "mapping", "value": "email_col"}},
)
client.run_table_enrichment(table.identifier, enrichment_id=str(result.id))

Connectors

# Create a custom HTTP API connector
connector = client.create_connector(
    name="My Scoring API",
    method="post",
    url="https://api.example.com/v1/score",
    headers=[{"name": "Authorization", "value": "Bearer sk-xxx"}],
    body=[{"name": "domain", "value": ""}],
    rate_limit=60,
)

# CRUD
connectors = client.list_connectors()
c = client.get_connector(connector.id)
client.update_connector(connector.id, name="Updated API", method="post", url="https://...")
client.delete_connector(connector.id)

Folders

# Organize tables in folders
folder = client.create_folder("My Leads")
folders = client.list_folders()
client.rename_folder(folder.id, "Leads 2026")
client.move_table_to_folder(table.identifier, folder_id=folder.id)
client.move_table_to_folder(table.identifier)          # remove from folder
client.delete_folder(folder.id)                        # tables move to root

Tasks

task = client.run_enrichment_bulk(123, [{"email": "a@b.com"}, {"email": "c@d.com"}])

# While a bulk run is going, progress tells you whether it is advancing or stuck
status = client.get_task(task.task_id)
status.progress          # {"total": 2, "completed": 1, "failed": 0, "processing": 1}

# Rows that have already finished, without waiting for the rest
client.get_task(task.task_id, include_partial=True).data

# Stop it — unfinished requests are refunded, finished rows keep their results
client.cancel_task(task.task_id)

Error handling

from databar import (
    DatabarClient,
    DatabarAuthError,
    DatabarInsufficientCreditsError,
    DatabarNotFoundError,
    DatabarTaskCancelledError,
    DatabarTaskFailedError,
    DatabarTimeoutError,
)

try:
    result = client.run_enrichment_sync(123, {"email": "alice@example.com"})
except DatabarAuthError:
    print("Invalid API key")
except DatabarInsufficientCreditsError:
    print("Not enough credits")
except DatabarNotFoundError:
    print("Enrichment not found")
except DatabarTaskFailedError as e:
    print(f"Task failed: {e.message}")
except DatabarTaskCancelledError as e:
    # Only the rows that finished, and not aligned to the inputs.
    print(f"Cancelled, got {len(e.partial_data or [])} rows")
except DatabarTimeoutError as e:
    print(f"Timed out after polling {e.max_attempts} times")

Context manager

with DatabarClient() as client:
    result = client.run_enrichment_sync(123, {"email": "alice@example.com"})
# connection pool closed automatically

CLI

After installing, the databar command is available in your terminal.

Authentication

databar login              # browser OAuth → save API key
databar login --api-key …  # paste a key (CI / agents)
databar logout             # remove saved key
databar whoami             # show name, email, balance, plan
databar whoami --format json

Enrichments

# List enrichments
databar enrich list
databar enrich list --query "linkedin"
databar enrich list --format json

# Get enrichment details
databar enrich get 123

# Run a single enrichment
databar enrich run 123 --params '{"email": "alice@example.com"}'
databar enrich run 123 --params '{"email": "alice@example.com"}' --format json

# Bulk run from CSV
databar enrich bulk 123 --input emails.csv --format csv --out results.csv

# Get choices for a select parameter
databar enrich choices 123 country
databar enrich choices 123 country --query "united"

Waterfalls

# List waterfalls
databar waterfall list
databar waterfall list --query "email"

# Get waterfall details
databar waterfall get email_getter

# Run a waterfall
databar waterfall run email_getter --params '{"linkedin_url": "https://linkedin.com/in/alice"}'

# Bulk run from CSV
databar waterfall bulk email_getter --input leads.csv --out results.csv

Flows

# List saved flows
databar flow list
databar flow list --query "buyer"

# Get flow details (declared inputs)
databar flow get <flow-id>

# Run a flow (--inputs maps each flow input id → value)
databar flow run <flow-id> --inputs '{"email": "alice@example.com"}'
databar flow run <flow-id> --inputs '{"email": "alice@example.com"}' --format json

Tables

# List tables
databar table list

# Create a table
databar table create --name "My Leads"
databar table create --name "My Leads" --columns "email,name,company"

# Inspect a table
databar table columns <uuid>
databar table rows <uuid>
databar table rows <uuid> --page 2 --per-page 500
databar table rows <uuid> --format csv --out rows.csv

# Insert rows
databar table insert <uuid> --data '[{"email":"alice@example.com","name":"Alice"}]'
databar table insert <uuid> --input data.csv --allow-new-columns
databar table insert <uuid> --input data.csv --dedupe-keys email

# Update rows by UUID
databar table patch <uuid> --data '[{"id":"<row-uuid>","email":"new@example.com"}]'

# Upsert rows by key column
databar table upsert <uuid> --key-col email --input data.csv

# Enrichments on a table
databar table enrichments <uuid>
databar table add-enrichment <uuid> --enrichment-id 123 --mapping '{"email": "email_col"}'
databar table run-enrichment <uuid> --enrichment-id <table-enrichment-id>

Tasks

# Check a task status — a bulk task also reports how many inputs are done
databar task get <task-id>

# Poll until complete
databar task get <task-id> --poll

# Collect the rows a running bulk task has already finished
databar task get <task-id> --partial

# Stop a running task; finished rows keep their results
databar task cancel <task-id>

Output formats

All commands support --format table|json|csv (default: table).

--format json prints one envelope on stdout: {"ok": true, "data": ...} or {"ok": false, "error": {"code", "message", ...}}.

# Pipe JSON output (payload is under .data)
databar table rows <uuid> --format json | jq '.data[].email'

# Save to CSV
databar enrich bulk 123 --input input.csv --format csv --out output.csv

Configuration

Development

git clone https://github.com/databar-ai/databar-python
cd databar-python
pip install -e ".[dev]"
pytest

License

MIT — see LICENSE.