The official Python client for the Data Legion API.
pip install datalegionfrom datalegion import DataLegion
client = DataLegion(api_key="legion_...")Or set the DATALEGION_API_KEY environment variable and omit the api_key argument:
client = DataLegion()Look up a person by email, phone, social URL, name, or other identifiers. Returns a PersonResponse with full type hints and dot access.
person = client.person.enrich(email="john@example.com")
print(person.full_name)
print(person.job_title)
print(person.company_name)
# Multiple results
results = client.person.enrich(
email="john@example.com",
multiple_results=True,
limit=5,
)
for match in results.matches:
print(match.person.full_name, match.match_metadata.match_confidence)
# Enrich by name + company
person = client.person.enrich(
first_name="John",
last_name="Doe",
company="Google",
)
# Enrich by LinkedIn URL
person = client.person.enrich(social_url="https://linkedin.com/in/johndoe")Search for people using SQL queries.
results = client.person.search(
query="SELECT * FROM people WHERE company_name ILIKE '%google%' AND city = 'San Francisco'",
limit=10,
)
for match in results.matches:
print(match.person.full_name)Search for people using natural language.
results = client.person.discover(
query="engineers in San Francisco who worked at Google",
limit=10,
)
for match in results.matches:
print(match.person.full_name)Look up a company by domain, name, LinkedIn ID, or ticker symbol. Returns a CompanyResponse with dot access.
# By domain
company = client.company.enrich(domain="google.com")
print(company.name.cleaned)
print(company.industry)
print(company.legion_employee_count)
# By name
company = client.company.enrich(name="Google")
# Multiple results
results = client.company.enrich(name="Apple", multiple_results=True, limit=3)
for match in results.matches:
print(match.company.name.cleaned)Search for companies using SQL queries.
results = client.company.search(
query="SELECT * FROM companies WHERE industry = 'software development'",
limit=10,
)
for match in results.matches:
print(match.company.name.cleaned)Search for companies using natural language.
results = client.company.discover(
query="AI companies with more than 100 employees",
limit=10,
)
for match in results.matches:
print(match.company.name.cleaned)cleaned = client.utility.clean(
fields={
"email": "John.Doe+tag@gmail.com",
"phone": "(555) 123-4567",
"domain": "https://www.Google.com/about",
}
)
for field, result in cleaned.results.items():
print(f"{field}: {result.original} -> {result.cleaned}")hashed = client.utility.hash_email(email="john@example.com")
print(hashed.hashes["sha256"])
print(hashed.hashes["md5"])validated = client.utility.validate(
email="john@example.com",
phone="+15551234567",
first_name="John",
)
print(validated.valid)
for error in validated.errors:
print(f"{error.field}: {error.error}")health = client.health()
print(health.status)import asyncio
from datalegion import AsyncDataLegion
async def main():
client = AsyncDataLegion(api_key="legion_...")
person = await client.person.enrich(email="john@example.com")
print(person.full_name)
company = await client.company.enrich(domain="google.com")
print(company.name.cleaned)
await client.close()
asyncio.run(main())Or use as a context manager:
async with AsyncDataLegion(api_key="legion_...") as client:
person = await client.person.enrich(email="john@example.com")All responses are Pydantic models with full type hints and autocomplete support. Import them directly:
from datalegion import PersonResponse, CompanyResponse, PersonMatchesResponseAfter each request, these attributes are updated on the client:
person = client.person.enrich(email="john@example.com")
print(client.request_id) # unique request ID
print(client.credits_used) # credits consumed
print(client.credits_remaining) # credits remaining
print(client.contract_id) # contract ID
print(client.rate_limit_limit) # requests quota in current window
print(client.rate_limit_remaining) # remaining requests in current window
print(client.rate_limit_reset) # unix timestamp when window resets
print(client.rate_limit_policy) # rate limit policy (e.g. "100/min")
print(client.retry_after) # seconds to wait (on 429 responses)
print(client.generated_query) # SQL from discover endpoints
print(client.correlation_id) # echoed Correlation-IDfrom datalegion import (
DataLegion,
AuthenticationError,
InsufficientCreditsError,
RateLimitError,
ValidationError,
APIError,
)
client = DataLegion(api_key="legion_...")
try:
person = client.person.enrich(email="john@example.com")
except AuthenticationError as e:
print(f"Invalid API key: {e.message}")
except InsufficientCreditsError as e:
print(f"Out of credits: {e.message}")
except RateLimitError as e:
print(f"Rate limited: {e.message}")
except ValidationError as e:
print(f"Invalid request: {e.message}, details: {e.details}")
except APIError as e:
print(f"Server error ({e.status_code}): {e.message}")All exceptions inherit from DataLegionError and include these attributes:
message- Human-readable error messagestatus_code- HTTP status codeerror- Error type/code from the APIdetails- Additional error details (if any)
| Parameter | Default | Description |
|---|---|---|
api_key |
DATALEGION_API_KEY env var |
Your API key |
base_url |
https://api.datalegion.ai |
API base URL |
timeout |
60.0 |
Request timeout in seconds |
httpx_client |
None | Custom httpx.Client / httpx.AsyncClient |
For full API documentation, visit https://www.datalegion.ai/docs.