Python SDK

Installation

pip install databar

Requires Python 3.9+. Dependencies (httpx, pydantic, typer, rich) are installed automatically.

Authentication

Get your API key from your Databar workspace → Integrations.

Environment variable (recommended)
Explicit in code

export DATABAR_API_KEY=your-key-here

from databar import DatabarClient

client = DatabarClient()  # reads DATABAR_API_KEY automatically

from databar import DatabarClient

client = DatabarClient(api_key="your-key-here")

Quickstart

from databar import DatabarClient

client = DatabarClient()

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

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

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

Enrichments

List and search enrichments

# List all enrichments
enrichments = client.list_enrichments()

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

for e in enrichments:
    print(f"[{e.id}] {e.name} — {e.price} credits/call")

Get enrichment details

enrichment = client.get_enrichment(123)

print(enrichment.name)
print(enrichment.description)

for param in enrichment.params:
    required = "required" if param.is_required else "optional"
    print(f"  {param.name} ({param.type_field}, {required}): {param.description}")

for field in enrichment.response_fields:
    print(f"  → {field.name} ({field.type_field})")

Run a single enrichment

# Async — returns immediately with a task ID
task = client.run_enrichment(123, {"email": "alice@example.com"})
print(task.task_id)  # poll this later

# Sync — submits and waits for completion in one call
result = client.run_enrichment_sync(123, {"email": "alice@example.com"})
print(result)

Bulk run

inputs = [
    {"email": "alice@example.com"},
    {"email": "bob@example.com"},
    {"email": "carol@example.com"},
]

# Sync — submits and waits for all results
results = client.run_enrichment_bulk_sync(123, inputs)

# Results are aligned to inputs: one element per input, in input order,
# with None for inputs that returned no data (len(results) == len(inputs)).
for lead, result in zip(inputs, results):
    if result is None:
        print(f"{lead['email']}: no data")
    else:
        print(f"{lead['email']}: {result}")

Get choices for a select parameter

# For enrichments with select/multiselect parameters
choices = client.get_param_choices(123, "country", q="united")

for choice in choices.items:
    print(f"  {choice.id}: {choice.name}")

if choices.has_next_page:
    next_page = client.get_param_choices(123, "country", page=2)

Waterfalls

Waterfalls chain multiple enrichment providers together with automatic fallback — if provider A doesn’t return a result, provider B is tried automatically.

List waterfalls

waterfalls = client.list_waterfalls()

for w in waterfalls:
    providers = len(w.available_enrichments)
    print(f"{w.identifier}: {w.name} ({providers} providers)")

Run a waterfall

# Uses the providers enabled by default (enabled_by_default=True)
result = client.run_waterfall_sync(
    "email_getter",
    {"linkedin_url": "https://linkedin.com/in/alice"}
)
print(result)

# Or specify providers explicitly
result = client.run_waterfall_sync(
    "email_getter",
    {"linkedin_url": "https://linkedin.com/in/alice"},
    enrichments=[10, 11, 12]  # provider IDs from get_waterfall()
)

Bulk run a waterfall

inputs = [
    {"linkedin_url": "https://linkedin.com/in/alice"},
    {"linkedin_url": "https://linkedin.com/in/bob"},
]

results = client.run_waterfall_bulk_sync("email_getter", inputs)
print(results)

Flows

flows = client.list_flows()
flow = client.get_flow("flow-uuid")

# inputs maps each flow input id to a value
result = client.run_flow_sync("flow-uuid", {"email": "alice@example.com"})
results = client.run_flow_bulk_sync(
    "flow-uuid",
    [{"email": "a@x.com"}, {"email": "b@x.com"}],
)

Tables

Create and list tables

# Create a table with predefined columns
table = client.create_table(
    name="My Leads",
    columns=["email", "name", "company", "linkedin_url"]
)
print(f"Created: {table.identifier}")

# List all tables
tables = client.list_tables()
for t in tables:
    print(f"{t.identifier}: {t.name}")

Read rows

# Get rows with pagination
data = client.get_rows(table.identifier, page=1, per_page=500)

Insert rows

from databar import InsertRow, InsertOptions, DedupeOptions

rows = [
    InsertRow(fields={"email": "alice@example.com", "name": "Alice"}),
    InsertRow(fields={"email": "bob@example.com", "name": "Bob"}),
]

# With deduplication on email column
response = client.create_rows(
    table.identifier,
    rows,
    options=InsertOptions(
        allow_new_columns=True,
        dedupe=DedupeOptions(enabled=True, keys=["email"])
    )
)

created = [r for r in response.results if r.action == "created"]
skipped = [r for r in response.results if r.action == "skipped_duplicate"]
print(f"Inserted {len(created)}, skipped {len(skipped)} duplicates")

Large inserts are auto-batched in chunks of 50 — no manual chunking needed.

Update rows

from databar import BatchUpdateRow

rows = [
    BatchUpdateRow(id="row-uuid-1", fields={"name": "Alice Smith"}),
    BatchUpdateRow(id="row-uuid-2", fields={"name": "Bob Jones"}),
]

response = client.patch_rows(table.identifier, rows)

Upsert rows

from databar import UpsertRow

rows = [
    UpsertRow(key={"email": "alice@example.com"}, fields={"name": "Alice", "company": "Acme"}),
    UpsertRow(key={"email": "new@example.com"}, fields={"name": "New User"}),
]

response = client.upsert_rows(table.identifier, rows)

for r in response.results:
    print(f"{r.id}: {r.action}")  # "created" or "updated"

Duplicate, clear, export, dedupe

copy = client.duplicate_table(table.identifier)

client.clear_table(table.identifier)

export = client.export_table(table.identifier, format="csv")
status = client.get_table_export(table.identifier)
print(status.file)

# Omit columns to compare every main column. Oldest row in each group is kept.
result = client.dedupe_table_rows(table.identifier)
print(result.count_rows_delete)

Reorder a column

client.reorder_column(table.identifier, column_id, index=0)

Workbooks

A workbook is the document that owns one or more tables (sheets). Table.workbook is the parent workbook id.

workbooks = client.list_workbooks()
wb = client.get_workbook(workbooks[0].identifier)

client.rename_workbook(wb.identifier, "Q4 leads")
client.add_workbook_table(wb.identifier, name="Sheet 2")

# Set the sheet (tab) order: table_ids lists every live sheet exactly once
sheets = client.get_workbook(wb.identifier).tables
client.reorder_workbook_tables(wb.identifier, [s.identifier for s in reversed(sheets)])

copy = client.duplicate_workbook(wb.identifier)

# Fold other workbooks' sheets into this one
client.merge_workbooks(wb.identifier, [copy.identifier])

# Move one sheet into another workbook, or split it into a new document
client.move_table(table.identifier, workbook=wb.identifier)
client.move_table(table.identifier, new_workbook=True, name="Split out")

# Trash for 30 days
client.delete_workbook(wb.identifier)

Trash

for item in client.list_trash():
    print(item.kind, item.identifier, item.name)

client.restore_trash(workbooks=["wb_id"], tables=["table-uuid"])
client.purge_trash(workbooks=["wb_id"])

Tasks

For async operations, you can check task status manually or poll until complete:

# Submit without waiting
task = client.run_enrichment(123, {"email": "alice@example.com"})
print(f"Task submitted: {task.task_id}")

# Check status once
status = client.get_task(task.task_id)
print(status.status)    # "processing", "completed", "failed", "cancelled", or "gone"
print(status.progress)  # bulk runs: {"total", "completed", "failed", "processing"}

# Poll until complete (blocks until done or times out)
result = client.poll_task(task.task_id)
print(result)

# Rows a running bulk task has already finished, without waiting for the rest
client.get_task(task.task_id, include_partial=True).data

# Stop a task that is going nowhere; unfinished requests are refunded
client.cancel_task(task.task_id)

Error handling

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

try:
    result = client.run_enrichment_sync(123, {"email": "alice@example.com"})
except DatabarAuthError:
    print("Invalid API key — check your DATABAR_API_KEY")
except DatabarInsufficientCreditsError:
    print("Out of credits — top up at databar.ai")
except DatabarNotFoundError:
    print("Enrichment not found")
except DatabarTaskFailedError as e:
    print(f"Enrichment failed: {e.message}")
except DatabarTimeoutError as e:
    print(f"Timed out after {e.max_attempts} polls — task may still be running")

Exception	HTTP Status	When raised
`DatabarAuthError`	401 / 403	Invalid or missing API key
`DatabarNotFoundError`	404	Enrichment, waterfall, or table not found
`DatabarInsufficientCreditsError`	406	Not enough credits
`DatabarGoneError`	410	Task results expired (stored 24 hours)
`DatabarValidationError`	422	Invalid request parameters
`DatabarRateLimitError`	429	Rate limit exceeded
`DatabarTaskFailedError`	—	Task status returned `failed`
`DatabarTimeoutError`	—	Polling exceeded max attempts

Configuration

client = DatabarClient(
    api_key="...",              # default: DATABAR_API_KEY env var
    base_url="https://api.databar.ai/v1",  # default
    timeout=30,                 # seconds per request (default: 30)
    max_poll_attempts=150,      # polling attempts before timeout (default: 150)
    poll_interval_s=2.0,        # seconds between polls (default: 2.0)
)

The client also works as a context manager — connection pool is closed automatically:

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

Source code

The SDK is open source under the MIT License.

databar-ai/databar-python

View source, report issues, and contribute on GitHub.

Getting Started

Overview

SDKs

Workspace

Workbooks

Tables

Headless

User

Installation

Authentication

Quickstart

Enrichments

Waterfalls

Flows

Tables

Workbooks

Trash

Tasks

Error handling

Configuration

Source code

databar-ai/databar-python

​Installation

​Authentication

​Quickstart

​Enrichments

​Waterfalls

​Flows

​Tables

​Workbooks

​Trash

​Tasks

​Error handling

​Configuration

​Source code

databar-ai/databar-python

Installation

Authentication

Quickstart

Enrichments

Waterfalls

Flows

Tables

Workbooks

Trash

Tasks

Error handling

Configuration

Source code