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 all available providers by default
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)

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"

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", or "gone"

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

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

Tables

Headless

User

Installation

Authentication

Quickstart

Enrichments

Waterfalls

Tables

Tasks

Error handling

Configuration

Source code

databar-ai/databar-python

​Installation

​Authentication

​Quickstart

​Enrichments

​Waterfalls

​Tables

​Tasks

​Error handling

​Configuration

​Source code

databar-ai/databar-python

Installation

Authentication

Quickstart

Enrichments

Waterfalls

Tables

Tasks

Error handling

Configuration

Source code