You want to run enrichment from Python. Maybe you are building it into a data pipeline, a CRM integration, or a lead qualification workflow. The Databar Python SDK gives you programmatic access to 100+ data providers through a single interface. This enrichment quickstart gets you from install to first result in under 10 minutes.
This guide covers installation, authentication, single lookups, batch enrichment, waterfall cascades, response parsing, and error handling. Every code example is production-ready.
Installation
Install the SDK via pip:
Requirements: Python 3.8+. No additional dependencies beyond what pip installs automatically.
Verify the installation:
Authentication
Every API call requires an API key. Get yours from the Databar dashboard under Settings → API Keys.
from databar import Databar
client = Databar(api_key="your_api_key_here")
from databar import Databar
client = Databar(api_key="your_api_key_here")
from databar import Databar
client = Databar(api_key="your_api_key_here")
from databar import Databar
client = Databar(api_key="your_api_key_here")
For production, store your API key in an environment variable:
import os
from databar import Databar
client = Databar(api_key=os.environ["DATABAR_API_KEY"])
import os
from databar import Databar
client = Databar(api_key=os.environ["DATABAR_API_KEY"])
import os
from databar import Databar
client = Databar(api_key=os.environ["DATABAR_API_KEY"])
import os
from databar import Databar
client = Databar(api_key=os.environ["DATABAR_API_KEY"])
The client handles authentication headers, base URL configuration, and session management. You create it once and reuse it across your application.
Single Contact Enrichment
The simplest use case: enrich one contact by email.
result = client.enrich(
email="jane@acme.com",
providers=["clearbit", "apollo"],
fields=["name", "title", "company", "phone", "linkedin"]
)
print(result.name) # "Jane Smith"
print(result.title) # "VP of Revenue Operations"
print(result.company) # "Acme Corp"
print(result.phone) # "+1-555-0123"
print(result.linkedin) # "linkedin.com/in/janesmith"
print(result.provider) # "clearbit" (which provider returned the data)
print(result.confidence) # 0.92result = client.enrich(
email="jane@acme.com",
providers=["clearbit", "apollo"],
fields=["name", "title", "company", "phone", "linkedin"]
)
print(result.name) # "Jane Smith"
print(result.title) # "VP of Revenue Operations"
print(result.company) # "Acme Corp"
print(result.phone) # "+1-555-0123"
print(result.linkedin) # "linkedin.com/in/janesmith"
print(result.provider) # "clearbit" (which provider returned the data)
print(result.confidence) # 0.92result = client.enrich(
email="jane@acme.com",
providers=["clearbit", "apollo"],
fields=["name", "title", "company", "phone", "linkedin"]
)
print(result.name) # "Jane Smith"
print(result.title) # "VP of Revenue Operations"
print(result.company) # "Acme Corp"
print(result.phone) # "+1-555-0123"
print(result.linkedin) # "linkedin.com/in/janesmith"
print(result.provider) # "clearbit" (which provider returned the data)
print(result.confidence) # 0.92result = client.enrich(
email="jane@acme.com",
providers=["clearbit", "apollo"],
fields=["name", "title", "company", "phone", "linkedin"]
)
print(result.name) # "Jane Smith"
print(result.title) # "VP of Revenue Operations"
print(result.company) # "Acme Corp"
print(result.phone) # "+1-555-0123"
print(result.linkedin) # "linkedin.com/in/janesmith"
print(result.provider) # "clearbit" (which provider returned the data)
print(result.confidence) # 0.92The providers parameter specifies which data providers to query. The fields parameter limits the response to only the data you need. Both are optional. If omitted, Databar uses defaults based on the enrichment type.
Company Enrichment
Enrich a company by domain:
company = client.enrich_company(
domain="acme.com",
fields=["name", "industry", "employee_count", "funding_total",
"tech_stack", "revenue_estimate"]
)
print(company.name) # "Acme Corp"
print(company.industry) # "SaaS"
print(company.employee_count) # 340
print(company.funding_total) # 45000000
print(company.tech_stack) # ["Salesforce", "HubSpot", "Snowflake"]
print(company.revenue_estimate) # "10M-50M"company = client.enrich_company(
domain="acme.com",
fields=["name", "industry", "employee_count", "funding_total",
"tech_stack", "revenue_estimate"]
)
print(company.name) # "Acme Corp"
print(company.industry) # "SaaS"
print(company.employee_count) # 340
print(company.funding_total) # 45000000
print(company.tech_stack) # ["Salesforce", "HubSpot", "Snowflake"]
print(company.revenue_estimate) # "10M-50M"company = client.enrich_company(
domain="acme.com",
fields=["name", "industry", "employee_count", "funding_total",
"tech_stack", "revenue_estimate"]
)
print(company.name) # "Acme Corp"
print(company.industry) # "SaaS"
print(company.employee_count) # 340
print(company.funding_total) # 45000000
print(company.tech_stack) # ["Salesforce", "HubSpot", "Snowflake"]
print(company.revenue_estimate) # "10M-50M"company = client.enrich_company(
domain="acme.com",
fields=["name", "industry", "employee_count", "funding_total",
"tech_stack", "revenue_estimate"]
)
print(company.name) # "Acme Corp"
print(company.industry) # "SaaS"
print(company.employee_count) # 340
print(company.funding_total) # 45000000
print(company.tech_stack) # ["Salesforce", "HubSpot", "Snowflake"]
print(company.revenue_estimate) # "10M-50M"Company enrichment pulls from firmographic, technographic, and funding providers in a single call. See our guide on multi-source enrichment for details on how providers combine.
Waterfall Enrichment
A waterfall cascades through multiple providers in sequence until one returns a result. This is the core pattern for maximizing match rates while minimizing cost.
email_result = client.waterfall(
enrichment_type="email_finder",
inputs={
"first_name": "Jane",
"last_name": "Smith",
"domain": "acme.com"
},
providers=["hunter", "findymail", "prospeo", "snov",
"apollo", "contactout", "rocketreach", "icypeas", "datagma"]
)
print(email_result.email) # "jane.smith@acme.com"
print(email_result.verified) # True
print(email_result.provider) # "hunter" (first provider that matched)
print(email_result.confidence) # 0.95email_result = client.waterfall(
enrichment_type="email_finder",
inputs={
"first_name": "Jane",
"last_name": "Smith",
"domain": "acme.com"
},
providers=["hunter", "findymail", "prospeo", "snov",
"apollo", "contactout", "rocketreach", "icypeas", "datagma"]
)
print(email_result.email) # "jane.smith@acme.com"
print(email_result.verified) # True
print(email_result.provider) # "hunter" (first provider that matched)
print(email_result.confidence) # 0.95email_result = client.waterfall(
enrichment_type="email_finder",
inputs={
"first_name": "Jane",
"last_name": "Smith",
"domain": "acme.com"
},
providers=["hunter", "findymail", "prospeo", "snov",
"apollo", "contactout", "rocketreach", "icypeas", "datagma"]
)
print(email_result.email) # "jane.smith@acme.com"
print(email_result.verified) # True
print(email_result.provider) # "hunter" (first provider that matched)
print(email_result.confidence) # 0.95email_result = client.waterfall(
enrichment_type="email_finder",
inputs={
"first_name": "Jane",
"last_name": "Smith",
"domain": "acme.com"
},
providers=["hunter", "findymail", "prospeo", "snov",
"apollo", "contactout", "rocketreach", "icypeas", "datagma"]
)
print(email_result.email) # "jane.smith@acme.com"
print(email_result.verified) # True
print(email_result.provider) # "hunter" (first provider that matched)
print(email_result.confidence) # 0.95The waterfall processes providers in the order you specify. It stops at the first successful match. You only pay for the provider that returns the result. Read more about waterfall enrichment and provider ordering strategy.
Batch Enrichment
For lists of contacts, use batch enrichment. It processes multiple contacts in parallel with automatic rate limit management.
contacts = [
{"email": "jane@acme.com"},
{"email": "bob@widgets.io"},
{"email": "sarah@startup.co"},
# ... up to thousands of contacts
]
batch = client.enrich_batch(
contacts=contacts,
providers=["clearbit", "apollo", "pdl"],
fields=["name", "title", "company", "phone"]
)
# Poll for completion
batch.wait() # Blocks until batch completes
# Access results
for result in batch.results:
print(f"{result.email}: {result.name}, {result.title} at {result.company}")contacts = [
{"email": "jane@acme.com"},
{"email": "bob@widgets.io"},
{"email": "sarah@startup.co"},
# ... up to thousands of contacts
]
batch = client.enrich_batch(
contacts=contacts,
providers=["clearbit", "apollo", "pdl"],
fields=["name", "title", "company", "phone"]
)
# Poll for completion
batch.wait() # Blocks until batch completes
# Access results
for result in batch.results:
print(f"{result.email}: {result.name}, {result.title} at {result.company}")contacts = [
{"email": "jane@acme.com"},
{"email": "bob@widgets.io"},
{"email": "sarah@startup.co"},
# ... up to thousands of contacts
]
batch = client.enrich_batch(
contacts=contacts,
providers=["clearbit", "apollo", "pdl"],
fields=["name", "title", "company", "phone"]
)
# Poll for completion
batch.wait() # Blocks until batch completes
# Access results
for result in batch.results:
print(f"{result.email}: {result.name}, {result.title} at {result.company}")contacts = [
{"email": "jane@acme.com"},
{"email": "bob@widgets.io"},
{"email": "sarah@startup.co"},
# ... up to thousands of contacts
]
batch = client.enrich_batch(
contacts=contacts,
providers=["clearbit", "apollo", "pdl"],
fields=["name", "title", "company", "phone"]
)
# Poll for completion
batch.wait() # Blocks until batch completes
# Access results
for result in batch.results:
print(f"{result.email}: {result.name}, {result.title} at {result.company}")For large batches, use the async pattern instead of blocking:
# Start the batch
batch = client.enrich_batch(
contacts=contacts,
providers=["clearbit", "apollo"],
fields=["name", "title", "company"]
)
# Check status without blocking
print(batch.status) # "processing"
print(batch.progress) # 0.45 (45% complete)
print(batch.estimated_time) # 120 (seconds remaining)
# Or use a callback
batch.on_complete(lambda results: process_results(results))# Start the batch
batch = client.enrich_batch(
contacts=contacts,
providers=["clearbit", "apollo"],
fields=["name", "title", "company"]
)
# Check status without blocking
print(batch.status) # "processing"
print(batch.progress) # 0.45 (45% complete)
print(batch.estimated_time) # 120 (seconds remaining)
# Or use a callback
batch.on_complete(lambda results: process_results(results))# Start the batch
batch = client.enrich_batch(
contacts=contacts,
providers=["clearbit", "apollo"],
fields=["name", "title", "company"]
)
# Check status without blocking
print(batch.status) # "processing"
print(batch.progress) # 0.45 (45% complete)
print(batch.estimated_time) # 120 (seconds remaining)
# Or use a callback
batch.on_complete(lambda results: process_results(results))# Start the batch
batch = client.enrich_batch(
contacts=contacts,
providers=["clearbit", "apollo"],
fields=["name", "title", "company"]
)
# Check status without blocking
print(batch.status) # "processing"
print(batch.progress) # 0.45 (45% complete)
print(batch.estimated_time) # 120 (seconds remaining)
# Or use a callback
batch.on_complete(lambda results: process_results(results))Batch enrichment is the right choice for lists of 100+ contacts. It is significantly faster than looping single lookups because Databar parallelizes across providers and manages rate limits server-side. See batch vs real-time enrichment for when to use each.
Batch Waterfall Enrichment
Combine batch and waterfall for the highest coverage on large lists:
contacts = [
{"first_name": "Jane", "last_name": "Smith", "domain": "acme.com"},
{"first_name": "Bob", "last_name": "Johnson", "domain": "widgets.io"},
# ...
]
batch = client.waterfall_batch(
enrichment_type="email_finder",
contacts=contacts,
providers=["hunter", "findymail", "prospeo", "snov",
"apollo", "contactout", "rocketreach"]
)
batch.wait()
for result in batch.results:
if result.found:
print(f"{result.first_name} {result.last_name}: {result.email} "
f"(via {result.provider}, verified: {result.verified})")
else:
print(f"{result.first_name} {result.last_name}: not found")contacts = [
{"first_name": "Jane", "last_name": "Smith", "domain": "acme.com"},
{"first_name": "Bob", "last_name": "Johnson", "domain": "widgets.io"},
# ...
]
batch = client.waterfall_batch(
enrichment_type="email_finder",
contacts=contacts,
providers=["hunter", "findymail", "prospeo", "snov",
"apollo", "contactout", "rocketreach"]
)
batch.wait()
for result in batch.results:
if result.found:
print(f"{result.first_name} {result.last_name}: {result.email} "
f"(via {result.provider}, verified: {result.verified})")
else:
print(f"{result.first_name} {result.last_name}: not found")contacts = [
{"first_name": "Jane", "last_name": "Smith", "domain": "acme.com"},
{"first_name": "Bob", "last_name": "Johnson", "domain": "widgets.io"},
# ...
]
batch = client.waterfall_batch(
enrichment_type="email_finder",
contacts=contacts,
providers=["hunter", "findymail", "prospeo", "snov",
"apollo", "contactout", "rocketreach"]
)
batch.wait()
for result in batch.results:
if result.found:
print(f"{result.first_name} {result.last_name}: {result.email} "
f"(via {result.provider}, verified: {result.verified})")
else:
print(f"{result.first_name} {result.last_name}: not found")contacts = [
{"first_name": "Jane", "last_name": "Smith", "domain": "acme.com"},
{"first_name": "Bob", "last_name": "Johnson", "domain": "widgets.io"},
# ...
]
batch = client.waterfall_batch(
enrichment_type="email_finder",
contacts=contacts,
providers=["hunter", "findymail", "prospeo", "snov",
"apollo", "contactout", "rocketreach"]
)
batch.wait()
for result in batch.results:
if result.found:
print(f"{result.first_name} {result.last_name}: {result.email} "
f"(via {result.provider}, verified: {result.verified})")
else:
print(f"{result.first_name} {result.last_name}: not found")Each contact runs through the full waterfall independently. Databar handles the parallel execution and rate limiting across all providers.
Response Parsing
Enrichment responses come back as typed objects with consistent field names regardless of which provider returned the data. Databar normalizes provider-specific field names into a standard schema.
result = client.enrich(email="jane@acme.com")
# Access as attributes
name = result.name
title = result.title
# Access as dictionary
data = result.to_dict()
print(data["name"])
# Check which fields were populated
print(result.populated_fields) # ["name", "title", "company", "phone"]
print(result.missing_fields) # ["linkedin", "twitter"]
# Get raw provider response (useful for debugging)
raw = result.raw_response
result = client.enrich(email="jane@acme.com")
# Access as attributes
name = result.name
title = result.title
# Access as dictionary
data = result.to_dict()
print(data["name"])
# Check which fields were populated
print(result.populated_fields) # ["name", "title", "company", "phone"]
print(result.missing_fields) # ["linkedin", "twitter"]
# Get raw provider response (useful for debugging)
raw = result.raw_response
result = client.enrich(email="jane@acme.com")
# Access as attributes
name = result.name
title = result.title
# Access as dictionary
data = result.to_dict()
print(data["name"])
# Check which fields were populated
print(result.populated_fields) # ["name", "title", "company", "phone"]
print(result.missing_fields) # ["linkedin", "twitter"]
# Get raw provider response (useful for debugging)
raw = result.raw_response
result = client.enrich(email="jane@acme.com")
# Access as attributes
name = result.name
title = result.title
# Access as dictionary
data = result.to_dict()
print(data["name"])
# Check which fields were populated
print(result.populated_fields) # ["name", "title", "company", "phone"]
print(result.missing_fields) # ["linkedin", "twitter"]
# Get raw provider response (useful for debugging)
raw = result.raw_response
Normalized fields mean you do not have to write provider-specific parsing logic. Whether the data came from Clearbit, Apollo, or PDL, the field names and types are identical.
Error Handling
Production code needs to handle failures gracefully. The SDK raises typed exceptions for different error conditions:
from databar import (
Databar,
AuthenticationError,
RateLimitError,
ProviderError,
NotFoundError,
ValidationError
)
client = Databar(api_key=os.environ["DATABAR_API_KEY"])
try:
result = client.enrich(email="jane@acme.com")
except AuthenticationError:
# Invalid or expired API key
print("Check your DATABAR_API_KEY")
except RateLimitError as e:
# You hit the platform rate limit (not provider rate limit)
print(f"Rate limited. Retry after {e.retry_after} seconds")
except ProviderError as e:
# A specific provider failed
print(f"Provider {e.provider} returned error: {e.message}")
except NotFoundError:
# No provider returned data for this input
print("Contact not found across all providers")
except ValidationError as e:
# Invalid input (bad email format, missing required fields)
print(f"Invalid input: {e.message}")from databar import (
Databar,
AuthenticationError,
RateLimitError,
ProviderError,
NotFoundError,
ValidationError
)
client = Databar(api_key=os.environ["DATABAR_API_KEY"])
try:
result = client.enrich(email="jane@acme.com")
except AuthenticationError:
# Invalid or expired API key
print("Check your DATABAR_API_KEY")
except RateLimitError as e:
# You hit the platform rate limit (not provider rate limit)
print(f"Rate limited. Retry after {e.retry_after} seconds")
except ProviderError as e:
# A specific provider failed
print(f"Provider {e.provider} returned error: {e.message}")
except NotFoundError:
# No provider returned data for this input
print("Contact not found across all providers")
except ValidationError as e:
# Invalid input (bad email format, missing required fields)
print(f"Invalid input: {e.message}")from databar import (
Databar,
AuthenticationError,
RateLimitError,
ProviderError,
NotFoundError,
ValidationError
)
client = Databar(api_key=os.environ["DATABAR_API_KEY"])
try:
result = client.enrich(email="jane@acme.com")
except AuthenticationError:
# Invalid or expired API key
print("Check your DATABAR_API_KEY")
except RateLimitError as e:
# You hit the platform rate limit (not provider rate limit)
print(f"Rate limited. Retry after {e.retry_after} seconds")
except ProviderError as e:
# A specific provider failed
print(f"Provider {e.provider} returned error: {e.message}")
except NotFoundError:
# No provider returned data for this input
print("Contact not found across all providers")
except ValidationError as e:
# Invalid input (bad email format, missing required fields)
print(f"Invalid input: {e.message}")from databar import (
Databar,
AuthenticationError,
RateLimitError,
ProviderError,
NotFoundError,
ValidationError
)
client = Databar(api_key=os.environ["DATABAR_API_KEY"])
try:
result = client.enrich(email="jane@acme.com")
except AuthenticationError:
# Invalid or expired API key
print("Check your DATABAR_API_KEY")
except RateLimitError as e:
# You hit the platform rate limit (not provider rate limit)
print(f"Rate limited. Retry after {e.retry_after} seconds")
except ProviderError as e:
# A specific provider failed
print(f"Provider {e.provider} returned error: {e.message}")
except NotFoundError:
# No provider returned data for this input
print("Contact not found across all providers")
except ValidationError as e:
# Invalid input (bad email format, missing required fields)
print(f"Invalid input: {e.message}")For batch enrichment, errors are per-contact, not per-batch:
batch.wait()
for result in batch.results:
if result.error:
print(f"Error for {result.email}: {result.error.message}")
else:
print(f"Found: {result.name} at {result.company}")batch.wait()
for result in batch.results:
if result.error:
print(f"Error for {result.email}: {result.error.message}")
else:
print(f"Found: {result.name} at {result.company}")batch.wait()
for result in batch.results:
if result.error:
print(f"Error for {result.email}: {result.error.message}")
else:
print(f"Found: {result.name} at {result.company}")batch.wait()
for result in batch.results:
if result.error:
print(f"Error for {result.email}: {result.error.message}")
else:
print(f"Found: {result.name} at {result.company}")Common Patterns
CRM Enrichment Pipeline
import csv
# Read contacts from CSV
with open("leads.csv") as f:
contacts = list(csv.DictReader(f))
# Enrich in batch
batch = client.enrich_batch(
contacts=[{"email": c["email"]} for c in contacts],
fields=["name", "title", "company", "phone", "employee_count"]
)
batch.wait()
# Write enriched data back
with open("leads_enriched.csv", "w", newline="") as f:
writer = csv.DictWriter(f, fieldnames=[
"email", "name", "title", "company", "phone", "employee_count"
])
writer.writeheader()
for result in batch.results:
writer.writerow(result.to_dict())import csv
# Read contacts from CSV
with open("leads.csv") as f:
contacts = list(csv.DictReader(f))
# Enrich in batch
batch = client.enrich_batch(
contacts=[{"email": c["email"]} for c in contacts],
fields=["name", "title", "company", "phone", "employee_count"]
)
batch.wait()
# Write enriched data back
with open("leads_enriched.csv", "w", newline="") as f:
writer = csv.DictWriter(f, fieldnames=[
"email", "name", "title", "company", "phone", "employee_count"
])
writer.writeheader()
for result in batch.results:
writer.writerow(result.to_dict())import csv
# Read contacts from CSV
with open("leads.csv") as f:
contacts = list(csv.DictReader(f))
# Enrich in batch
batch = client.enrich_batch(
contacts=[{"email": c["email"]} for c in contacts],
fields=["name", "title", "company", "phone", "employee_count"]
)
batch.wait()
# Write enriched data back
with open("leads_enriched.csv", "w", newline="") as f:
writer = csv.DictWriter(f, fieldnames=[
"email", "name", "title", "company", "phone", "employee_count"
])
writer.writeheader()
for result in batch.results:
writer.writerow(result.to_dict())import csv
# Read contacts from CSV
with open("leads.csv") as f:
contacts = list(csv.DictReader(f))
# Enrich in batch
batch = client.enrich_batch(
contacts=[{"email": c["email"]} for c in contacts],
fields=["name", "title", "company", "phone", "employee_count"]
)
batch.wait()
# Write enriched data back
with open("leads_enriched.csv", "w", newline="") as f:
writer = csv.DictWriter(f, fieldnames=[
"email", "name", "title", "company", "phone", "employee_count"
])
writer.writeheader()
for result in batch.results:
writer.writerow(result.to_dict())Real-Time Form Enrichment
from flask import Flask, request, jsonify
app = Flask(__name__)
client = Databar(api_key=os.environ["DATABAR_API_KEY"])
@app.route("/api/enrich-lead", methods=["POST"])
def enrich_lead():
email = request.json["email"]
try:
result = client.enrich(
email=email,
fields=["name", "title", "company", "employee_count"],
timeout=3000 # 3 second timeout
)
return jsonify(result.to_dict())
except Exception:
# Return partial data on failure
return jsonify({"email": email, "enriched": False})from flask import Flask, request, jsonify
app = Flask(__name__)
client = Databar(api_key=os.environ["DATABAR_API_KEY"])
@app.route("/api/enrich-lead", methods=["POST"])
def enrich_lead():
email = request.json["email"]
try:
result = client.enrich(
email=email,
fields=["name", "title", "company", "employee_count"],
timeout=3000 # 3 second timeout
)
return jsonify(result.to_dict())
except Exception:
# Return partial data on failure
return jsonify({"email": email, "enriched": False})from flask import Flask, request, jsonify
app = Flask(__name__)
client = Databar(api_key=os.environ["DATABAR_API_KEY"])
@app.route("/api/enrich-lead", methods=["POST"])
def enrich_lead():
email = request.json["email"]
try:
result = client.enrich(
email=email,
fields=["name", "title", "company", "employee_count"],
timeout=3000 # 3 second timeout
)
return jsonify(result.to_dict())
except Exception:
# Return partial data on failure
return jsonify({"email": email, "enriched": False})from flask import Flask, request, jsonify
app = Flask(__name__)
client = Databar(api_key=os.environ["DATABAR_API_KEY"])
@app.route("/api/enrich-lead", methods=["POST"])
def enrich_lead():
email = request.json["email"]
try:
result = client.enrich(
email=email,
fields=["name", "title", "company", "employee_count"],
timeout=3000 # 3 second timeout
)
return jsonify(result.to_dict())
except Exception:
# Return partial data on failure
return jsonify({"email": email, "enriched": False})Waterfall With Verification
Configuration Options
client = Databar(
api_key=os.environ["DATABAR_API_KEY"],
timeout=5000, # Default timeout in ms for all requests
max_retries=3, # Retry failed requests up to 3 times
retry_delay=1000, # Wait 1 second between retries
cache=True # Use platform-level caching (default: True)
)client = Databar(
api_key=os.environ["DATABAR_API_KEY"],
timeout=5000, # Default timeout in ms for all requests
max_retries=3, # Retry failed requests up to 3 times
retry_delay=1000, # Wait 1 second between retries
cache=True # Use platform-level caching (default: True)
)client = Databar(
api_key=os.environ["DATABAR_API_KEY"],
timeout=5000, # Default timeout in ms for all requests
max_retries=3, # Retry failed requests up to 3 times
retry_delay=1000, # Wait 1 second between retries
cache=True # Use platform-level caching (default: True)
)client = Databar(
api_key=os.environ["DATABAR_API_KEY"],
timeout=5000, # Default timeout in ms for all requests
max_retries=3, # Retry failed requests up to 3 times
retry_delay=1000, # Wait 1 second between retries
cache=True # Use platform-level caching (default: True)
)These defaults work for most use cases. Lower the timeout for real-time use cases. Increase max_retries for batch jobs where reliability matters more than speed.
Company Enrichment
Enrich companies alongside contacts to get firmographic data:
# Enrich a batch of companies
domains = ["acme.com", "widgets.io", "startup.co"]
company_batch = client.enrich_batch(
contacts=[{"domain": d} for d in domains],
enrichment_type="company",
fields=["name", "industry", "employee_count", "funding_total",
"tech_stack", "revenue_estimate", "founded_year"]
)
company_batch.wait()
for company in company_batch.results:
print(f"{company.name}: {company.employee_count} employees, "
f"{company.industry}, raised ${company.funding_total:,}")# Enrich a batch of companies
domains = ["acme.com", "widgets.io", "startup.co"]
company_batch = client.enrich_batch(
contacts=[{"domain": d} for d in domains],
enrichment_type="company",
fields=["name", "industry", "employee_count", "funding_total",
"tech_stack", "revenue_estimate", "founded_year"]
)
company_batch.wait()
for company in company_batch.results:
print(f"{company.name}: {company.employee_count} employees, "
f"{company.industry}, raised ${company.funding_total:,}")# Enrich a batch of companies
domains = ["acme.com", "widgets.io", "startup.co"]
company_batch = client.enrich_batch(
contacts=[{"domain": d} for d in domains],
enrichment_type="company",
fields=["name", "industry", "employee_count", "funding_total",
"tech_stack", "revenue_estimate", "founded_year"]
)
company_batch.wait()
for company in company_batch.results:
print(f"{company.name}: {company.employee_count} employees, "
f"{company.industry}, raised ${company.funding_total:,}")# Enrich a batch of companies
domains = ["acme.com", "widgets.io", "startup.co"]
company_batch = client.enrich_batch(
contacts=[{"domain": d} for d in domains],
enrichment_type="company",
fields=["name", "industry", "employee_count", "funding_total",
"tech_stack", "revenue_estimate", "founded_year"]
)
company_batch.wait()
for company in company_batch.results:
print(f"{company.name}: {company.employee_count} employees, "
f"{company.industry}, raised ${company.funding_total:,}")Company enrichment pulls from firmographic, technographic, and funding providers in a single call. Combine contact and company enrichment in the same pipeline to build complete account profiles. See our guide on waterfall enrichment for CRM data quality.
Phone Number Enrichment
The SDK also supports phone number lookups through the waterfall:
phone_result = client.waterfall(
enrichment_type="phone_finder",
inputs={
"first_name": "Jane",
"last_name": "Smith",
"domain": "acme.com"
},
providers=["datagma", "contactout", "rocketreach", "apollo",
"lusha", "pdl", "leadmagic"]
)
if phone_result.found:
print(f"Phone: {phone_result.phone}")
print(f"Type: {phone_result.phone_type}") # "mobile", "direct", "switchboard"
print(f"Provider: {phone_result.provider}")phone_result = client.waterfall(
enrichment_type="phone_finder",
inputs={
"first_name": "Jane",
"last_name": "Smith",
"domain": "acme.com"
},
providers=["datagma", "contactout", "rocketreach", "apollo",
"lusha", "pdl", "leadmagic"]
)
if phone_result.found:
print(f"Phone: {phone_result.phone}")
print(f"Type: {phone_result.phone_type}") # "mobile", "direct", "switchboard"
print(f"Provider: {phone_result.provider}")phone_result = client.waterfall(
enrichment_type="phone_finder",
inputs={
"first_name": "Jane",
"last_name": "Smith",
"domain": "acme.com"
},
providers=["datagma", "contactout", "rocketreach", "apollo",
"lusha", "pdl", "leadmagic"]
)
if phone_result.found:
print(f"Phone: {phone_result.phone}")
print(f"Type: {phone_result.phone_type}") # "mobile", "direct", "switchboard"
print(f"Provider: {phone_result.provider}")phone_result = client.waterfall(
enrichment_type="phone_finder",
inputs={
"first_name": "Jane",
"last_name": "Smith",
"domain": "acme.com"
},
providers=["datagma", "contactout", "rocketreach", "apollo",
"lusha", "pdl", "leadmagic"]
)
if phone_result.found:
print(f"Phone: {phone_result.phone}")
print(f"Type: {phone_result.phone_type}") # "mobile", "direct", "switchboard"
print(f"Provider: {phone_result.provider}")Phone match rates are lower than email (40-60% vs 75-90%), so the waterfall matters even more. Every provider adds incremental coverage. See our guide on phone enrichment for cold outbound.
Webhook Integration
For event-driven architectures, use webhooks to receive enrichment results asynchronously:
# Set up a webhook endpoint in your app
from flask import Flask, request
app = Flask(__name__)
@app.route("/webhook/enrichment", methods=["POST"])
def enrichment_webhook():
data = request.json
contact_id = data["contact_id"]
enriched_data = data["result"]
# Update your database with enriched data
update_contact(contact_id, enriched_data)
return {"status": "ok"}, 200
# Start an async enrichment with webhook callback
batch = client.enrich_batch(
contacts=contacts,
providers=["clearbit", "apollo"],
fields=["name", "title", "company"],
webhook_url="https://your-app.com/webhook/enrichment"
)
# No need to poll - results arrive at your webhook# Set up a webhook endpoint in your app
from flask import Flask, request
app = Flask(__name__)
@app.route("/webhook/enrichment", methods=["POST"])
def enrichment_webhook():
data = request.json
contact_id = data["contact_id"]
enriched_data = data["result"]
# Update your database with enriched data
update_contact(contact_id, enriched_data)
return {"status": "ok"}, 200
# Start an async enrichment with webhook callback
batch = client.enrich_batch(
contacts=contacts,
providers=["clearbit", "apollo"],
fields=["name", "title", "company"],
webhook_url="https://your-app.com/webhook/enrichment"
)
# No need to poll - results arrive at your webhook# Set up a webhook endpoint in your app
from flask import Flask, request
app = Flask(__name__)
@app.route("/webhook/enrichment", methods=["POST"])
def enrichment_webhook():
data = request.json
contact_id = data["contact_id"]
enriched_data = data["result"]
# Update your database with enriched data
update_contact(contact_id, enriched_data)
return {"status": "ok"}, 200
# Start an async enrichment with webhook callback
batch = client.enrich_batch(
contacts=contacts,
providers=["clearbit", "apollo"],
fields=["name", "title", "company"],
webhook_url="https://your-app.com/webhook/enrichment"
)
# No need to poll - results arrive at your webhook# Set up a webhook endpoint in your app
from flask import Flask, request
app = Flask(__name__)
@app.route("/webhook/enrichment", methods=["POST"])
def enrichment_webhook():
data = request.json
contact_id = data["contact_id"]
enriched_data = data["result"]
# Update your database with enriched data
update_contact(contact_id, enriched_data)
return {"status": "ok"}, 200
# Start an async enrichment with webhook callback
batch = client.enrich_batch(
contacts=contacts,
providers=["clearbit", "apollo"],
fields=["name", "title", "company"],
webhook_url="https://your-app.com/webhook/enrichment"
)
# No need to poll - results arrive at your webhookWebhooks are the right pattern when you do not want your application to maintain a polling loop. The enrichment runs in the background and notifies you on completion.
Rate Limits and Retries
The SDK handles rate limits automatically. If the API returns a 429 response, the SDK waits and retries based on the Retry-After header. You can customize this behavior:
client = Databar(
api_key=os.environ["DATABAR_API_KEY"],
max_retries=5, # More retries for critical workflows
retry_delay=2000, # 2 second initial delay
retry_backoff="exponential" # Double the delay on each retry
)client = Databar(
api_key=os.environ["DATABAR_API_KEY"],
max_retries=5, # More retries for critical workflows
retry_delay=2000, # 2 second initial delay
retry_backoff="exponential" # Double the delay on each retry
)client = Databar(
api_key=os.environ["DATABAR_API_KEY"],
max_retries=5, # More retries for critical workflows
retry_delay=2000, # 2 second initial delay
retry_backoff="exponential" # Double the delay on each retry
)client = Databar(
api_key=os.environ["DATABAR_API_KEY"],
max_retries=5, # More retries for critical workflows
retry_delay=2000, # 2 second initial delay
retry_backoff="exponential" # Double the delay on each retry
)For high-throughput use cases, the SDK also respects concurrent request limits. It will queue requests that exceed your plan's concurrency limit and process them as capacity becomes available. You do not need to implement throttling in your application code.
Data Export and Integration
The SDK supports multiple output formats for integrating with your existing tools:
# Export to pandas DataFrame
import pandas as pd
batch.wait()
df = pd.DataFrame([r.to_dict() for r in batch.results])
print(df.describe())
# Filter to only found contacts
found = df[df["name"].notna()]
print(f"Found {len(found)} of {len(df)} contacts")
# Export to JSON for API consumption
import json
with open("enriched.json", "w") as f:
json.dump([r.to_dict() for r in batch.results], f, indent=2)
# Push to CRM via webhook
import requests
for result in batch.results:
if not result.error:
requests.post(
"https://your-crm.com/api/contacts",
json=result.to_dict(),
headers={"Authorization": "Bearer YOUR_CRM_TOKEN"}
)# Export to pandas DataFrame
import pandas as pd
batch.wait()
df = pd.DataFrame([r.to_dict() for r in batch.results])
print(df.describe())
# Filter to only found contacts
found = df[df["name"].notna()]
print(f"Found {len(found)} of {len(df)} contacts")
# Export to JSON for API consumption
import json
with open("enriched.json", "w") as f:
json.dump([r.to_dict() for r in batch.results], f, indent=2)
# Push to CRM via webhook
import requests
for result in batch.results:
if not result.error:
requests.post(
"https://your-crm.com/api/contacts",
json=result.to_dict(),
headers={"Authorization": "Bearer YOUR_CRM_TOKEN"}
)# Export to pandas DataFrame
import pandas as pd
batch.wait()
df = pd.DataFrame([r.to_dict() for r in batch.results])
print(df.describe())
# Filter to only found contacts
found = df[df["name"].notna()]
print(f"Found {len(found)} of {len(df)} contacts")
# Export to JSON for API consumption
import json
with open("enriched.json", "w") as f:
json.dump([r.to_dict() for r in batch.results], f, indent=2)
# Push to CRM via webhook
import requests
for result in batch.results:
if not result.error:
requests.post(
"https://your-crm.com/api/contacts",
json=result.to_dict(),
headers={"Authorization": "Bearer YOUR_CRM_TOKEN"}
)# Export to pandas DataFrame
import pandas as pd
batch.wait()
df = pd.DataFrame([r.to_dict() for r in batch.results])
print(df.describe())
# Filter to only found contacts
found = df[df["name"].notna()]
print(f"Found {len(found)} of {len(df)} contacts")
# Export to JSON for API consumption
import json
with open("enriched.json", "w") as f:
json.dump([r.to_dict() for r in batch.results], f, indent=2)
# Push to CRM via webhook
import requests
for result in batch.results:
if not result.error:
requests.post(
"https://your-crm.com/api/contacts",
json=result.to_dict(),
headers={"Authorization": "Bearer YOUR_CRM_TOKEN"}
)For teams using data pipelines, the SDK integrates naturally with pandas, Airflow, Dagster, and any Python-based ETL framework. Enrich contacts as a pipeline step and push results downstream to your warehouse, CRM, or outbound tools.
Environment-Specific Configuration
Set up different configurations for development, staging, and production:
import os
env = os.environ.get("ENVIRONMENT", "development")
config = {
"development": {
"api_key": os.environ.get("DATABAR_DEV_KEY"),
"timeout": 10000,
"max_retries": 1,
"debug": True
},
"staging": {
"api_key": os.environ.get("DATABAR_STAGING_KEY"),
"timeout": 5000,
"max_retries": 3,
"debug": False
},
"production": {
"api_key": os.environ.get("DATABAR_API_KEY"),
"timeout": 5000,
"max_retries": 5,
"debug": False
}
}
client = Databar(**config[env])import os
env = os.environ.get("ENVIRONMENT", "development")
config = {
"development": {
"api_key": os.environ.get("DATABAR_DEV_KEY"),
"timeout": 10000,
"max_retries": 1,
"debug": True
},
"staging": {
"api_key": os.environ.get("DATABAR_STAGING_KEY"),
"timeout": 5000,
"max_retries": 3,
"debug": False
},
"production": {
"api_key": os.environ.get("DATABAR_API_KEY"),
"timeout": 5000,
"max_retries": 5,
"debug": False
}
}
client = Databar(**config[env])import os
env = os.environ.get("ENVIRONMENT", "development")
config = {
"development": {
"api_key": os.environ.get("DATABAR_DEV_KEY"),
"timeout": 10000,
"max_retries": 1,
"debug": True
},
"staging": {
"api_key": os.environ.get("DATABAR_STAGING_KEY"),
"timeout": 5000,
"max_retries": 3,
"debug": False
},
"production": {
"api_key": os.environ.get("DATABAR_API_KEY"),
"timeout": 5000,
"max_retries": 5,
"debug": False
}
}
client = Databar(**config[env])import os
env = os.environ.get("ENVIRONMENT", "development")
config = {
"development": {
"api_key": os.environ.get("DATABAR_DEV_KEY"),
"timeout": 10000,
"max_retries": 1,
"debug": True
},
"staging": {
"api_key": os.environ.get("DATABAR_STAGING_KEY"),
"timeout": 5000,
"max_retries": 3,
"debug": False
},
"production": {
"api_key": os.environ.get("DATABAR_API_KEY"),
"timeout": 5000,
"max_retries": 5,
"debug": False
}
}
client = Databar(**config[env])Use separate API keys per environment to isolate credit usage and avoid production data leaking into development workflows.
Logging and Debugging (Databar Python Sdk Enrichment Quickstart)
Enable verbose logging for development and debugging:
import logging
logging.basicConfig(level=logging.DEBUG)
client = Databar(api_key=os.environ["DATABAR_API_KEY"], debug=True)
# Now all API calls are logged with request/response details
result = client.enrich(email="jane@acme.com")
# DEBUG: POST /api/v1/enrich {"email": "jane@acme.com"} returns 200 (482ms)import logging
logging.basicConfig(level=logging.DEBUG)
client = Databar(api_key=os.environ["DATABAR_API_KEY"], debug=True)
# Now all API calls are logged with request/response details
result = client.enrich(email="jane@acme.com")
# DEBUG: POST /api/v1/enrich {"email": "jane@acme.com"} returns 200 (482ms)import logging
logging.basicConfig(level=logging.DEBUG)
client = Databar(api_key=os.environ["DATABAR_API_KEY"], debug=True)
# Now all API calls are logged with request/response details
result = client.enrich(email="jane@acme.com")
# DEBUG: POST /api/v1/enrich {"email": "jane@acme.com"} returns 200 (482ms)import logging
logging.basicConfig(level=logging.DEBUG)
client = Databar(api_key=os.environ["DATABAR_API_KEY"], debug=True)
# Now all API calls are logged with request/response details
result = client.enrich(email="jane@acme.com")
# DEBUG: POST /api/v1/enrich {"email": "jane@acme.com"} returns 200 (482ms)In production, set logging to INFO or WARNING to avoid performance overhead from verbose logging. The SDK logs errors and retries at WARNING level by default, which is usually sufficient for production monitoring.
Databar Python Sdk Enrichment Quickstart: Python SDK Enrichment Quickstart: What to Build Next
You now have the building blocks for any enrichment workflow in Python. The Databar Python SDK handles authentication, rate limits, provider management, and response normalization. Your code focuses on business logic.
Start with a single lookup to test your API key. Then try a batch of 100 contacts from your CRM. Then build a waterfall to see how match rates improve across multiple providers. The SDK supports every pattern from this enrichment quickstart through to production-scale pipelines.
Databar's Python SDK connects to 100+ data providers on a monthly credit plan with outcome-based billing, so credits are only consumed when data is successfully returned. No annual contracts. Check the MCP vs SDK vs API guide to decide which integration method fits your workflow best.
Frequently Asked Questions
What is Databar Python Sdk Enrichment Quickstart?
Databar Python Sdk Enrichment Quickstart uses data tools and workflows to improve GTM operations. Databar makes this accessible through 100+ providers in one API.
How much does Databar Python Sdk Enrichment Quickstart cost?
With Databar, you pay only for enrichments you use. No monthly minimums or annual contracts. Credits start at fractions of a cent per lookup.
Can I try Databar Python Sdk Enrichment Quickstart for free?
Yes. Databar offers a 14-day free trial with full API access. Sign up in under a minute and run your first enrichment immediately.
Do I need technical skills for Databar Python Sdk Enrichment Quickstart?
No. Databar offers both a no-code interface and a full API. Non-technical teams use the visual builder. Developers use the REST API, Python SDK, or MCP server.
How does Databar Python Sdk Enrichment Quickstart integrate with my CRM?
Databar connects to any CRM through its API, Zapier, Make, or n8n. Set up triggered enrichment on new records or run batch enrichment on existing data.
Also Interesting
