> ## Documentation Index
> Fetch the complete documentation index at: https://docs.databar.ai/llms.txt
> Use this file to discover all available pages before exploring further.

# Python SDK

> Install the official Databar Python SDK and start enriching data in minutes.

## Installation

```bash theme={null}
pip install databar
```

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

***

## Authentication

Get your API key from your [Databar workspace](https://databar.ai) → **Integrations**.

<Tabs>
  <Tab title="Environment variable (recommended)">
    ```bash theme={null}
    export DATABAR_API_KEY=your-key-here
    ```

    ```python theme={null}
    from databar import DatabarClient

    client = DatabarClient()  # reads DATABAR_API_KEY automatically
    ```
  </Tab>

  <Tab title="Explicit in code">
    ```python theme={null}
    from databar import DatabarClient

    client = DatabarClient(api_key="your-key-here")
    ```
  </Tab>
</Tabs>

***

## Quickstart

```python theme={null}
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

<AccordionGroup>
  <Accordion title="List and search enrichments">
    ```python theme={null}
    # 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")
    ```
  </Accordion>

  <Accordion title="Get enrichment details">
    ```python theme={null}
    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})")
    ```
  </Accordion>

  <Accordion title="Run a single enrichment">
    ```python theme={null}
    # 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)
    ```
  </Accordion>

  <Accordion title="Bulk run">
    ```python theme={null}
    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)
    print(results)
    ```
  </Accordion>

  <Accordion title="Get choices for a select parameter">
    ```python theme={null}
    # 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)
    ```
  </Accordion>
</AccordionGroup>

***

## Waterfalls

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

<AccordionGroup>
  <Accordion title="List waterfalls">
    ```python theme={null}
    waterfalls = client.list_waterfalls()

    for w in waterfalls:
        providers = len(w.available_enrichments)
        print(f"{w.identifier}: {w.name} ({providers} providers)")
    ```
  </Accordion>

  <Accordion title="Run a waterfall">
    ```python theme={null}
    # 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()
    )
    ```
  </Accordion>

  <Accordion title="Bulk run a waterfall">
    ```python theme={null}
    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)
    ```
  </Accordion>
</AccordionGroup>

***

## Tables

<AccordionGroup>
  <Accordion title="Create and list tables">
    ```python theme={null}
    # 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}")
    ```
  </Accordion>

  <Accordion title="Read rows">
    ```python theme={null}
    # Get rows with pagination
    data = client.get_rows(table.identifier, page=1, per_page=500)
    ```
  </Accordion>

  <Accordion title="Insert rows">
    ```python theme={null}
    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.
  </Accordion>

  <Accordion title="Update rows">
    ```python theme={null}
    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)
    ```
  </Accordion>

  <Accordion title="Upsert rows">
    ```python theme={null}
    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"
    ```
  </Accordion>
</AccordionGroup>

***

## Tasks

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

```python theme={null}
# 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

```python theme={null}
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

```python theme={null}
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:

```python theme={null}
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.

<Card title="databar-ai/databar-python" icon="github" href="https://github.com/databar-ai/databar-python">
  View source, report issues, and contribute on GitHub.
</Card>