Python SDK

# 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")

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})")

# 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)

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)

# 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 = client.list_waterfalls()

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

# 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()
)

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)

# 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}")

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

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")

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)

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"