Python SDK

The official Python SDK for the Data Legion API. Includes sync and async clients, Pydantic v2 response models, and full type hints.

Installation

pip install datalegion

uv add datalegion

poetry add datalegion

Requires Python 3.10+.

Authentication

Get an API key by signing up for free; it’s available in your dashboard as soon as you sign up. The same key works across the API, SDKs, MCP server, and CLI.

import os
os.environ["DATALEGION_API_KEY"] = "legion_..."

from datalegion import DataLegion
client = DataLegion()

from datalegion import DataLegion
client = DataLegion(api_key="legion_...")

Person

Enrichment

Look up a person by email, phone, name, LinkedIn, or other identifiers.

from datalegion import DataLegion

client = DataLegion(api_key="legion_...")

person = client.person.enrich(email="jane.doe@example.com")

print(person.full_name)          # "Jane Doe"
print(person.job_title)          # "VP of Engineering"
print(person.company_name)       # "Acme Corp"
print(person.city, person.state) # "San Francisco" "California"

Enterprise accounts can return several matches with confidence levels by setting multiple_results=True. The feature is switched on per account: ask your account manager or support@datalegion.ai. Without it, the request returns 403 feature_not_available. Each match returned is billed.

matches = client.person.enrich(
    first_name="Jane",
    last_name="Doe",
    company="Acme",
    multiple_results=True,
    limit=5,
    min_confidence="high",
)

print(f"Found {matches.total} matches")
for match in matches.matches:
    meta = match.match_metadata
    print(f"{match.person.full_name} - {meta.match_confidence}")

Search

Query people using a SQL SELECT against the people table. The query must start with SELECT and must not include a LIMIT clause; pass limit as a parameter.

results = client.person.search(
    query="SELECT * FROM people WHERE job_title ILIKE '%engineer%' AND city = 'san francisco'",
    limit=25,
)

# `total` reflects the full row count matching the WHERE clause across the
# database (useful for TAM / market-sizing), while `matches` is capped at `limit`.
print(f"{len(results.matches)} of {results.total} total matches")
for match in results.matches:
    print(f"{match.person.full_name} - {match.person.job_title}")

Discovery

Find people using natural language.

results = client.person.discover(
    query="product managers at AI startups in New York",
    limit=10,
)

# The generated SQL is included in the response
print(f"Generated query: {results.generated_query}")

for match in results.matches:
    print(f"{match.person.full_name} at {match.person.company_name}")

Company

Enrichment

Look up a company by domain, name, ticker, or LinkedIn.

company = client.company.enrich(domain="google.com")

print(company.name)                    # CleanedRaw object
print(company.industry)                # "Internet"
print(company.legion_employee_count)   # 182000
print(company.legion_employee_growth_rate)  # {"1m": 0.02, "3m": 0.05, ...}

Search

Query companies using SQL.

results = client.company.search(
    query="SELECT * FROM companies WHERE industry = 'software development' AND legion_employee_count > 50",
    limit=20,
)

for match in results.matches:
    print(f"{match.company.name} - {match.company.industry}")

Discovery

Find companies using natural language.

results = client.company.discover(
    query="fintech companies in London with over 200 employees",
    limit=10,
)

for match in results.matches:
    print(f"{match.company.name} - {match.company.legion_employee_count} employees")

Utility Endpoints

Utility endpoints are free and don’t consume credits.

Clean Fields

cleaned = client.utility.clean(
    fields={
        "email": "  JANE.DOE+work@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}")

Hash Email

hashed = client.utility.hash_email(email="Jane.Doe+tag@gmail.com")

print(hashed.normalized_email)    # "janedoe@gmail.com"
print(hashed.hashes["sha256"])    # SHA-256 hash
print(hashed.hashes["md5"])       # MD5 hash

Validate Data

result = client.utility.validate(
    email="not-an-email",
    phone="+15551234567",
    company="Google",
)

if not result.valid:
    for error in result.errors:
        print(f"{error.field}: {error.error}")

Report a Data Issue

Found something wrong in a record? Report it against its legion_id. See Report a Data Issue for the issue_type/issue_level rules.

report = client.utility.report_person(
    legion_id="abc123",
    issue_type="incorrect",
    issue_level="field",
    field="job_title",
    observed_value="CEO",
    correct_value="CTO",
)

print(report.report_id, report.status)  # e.g. 42 "new"

Company records use client.utility.report_company() with the same parameters.

Async Client

The AsyncDataLegion client has the same interface but all methods are async.

import asyncio
from datalegion import AsyncDataLegion

async def main():
    async with AsyncDataLegion(api_key="legion_...") as client:
        person = await client.person.enrich(email="jane@example.com")
        print(person.full_name)

        # Concurrent requests
        results = await asyncio.gather(
            client.person.enrich(email="jane@example.com"),
            client.company.enrich(domain="google.com"),
        )

asyncio.run(main())

Response Metadata

After each request, response metadata is available on the client.

person = client.person.enrich(email="jane@example.com")

print(client.request_id)            # Unique request ID
print(client.credits_used)          # Credits consumed
print(client.credits_remaining)     # Remaining balance
print(client.rate_limit_remaining)  # Requests left in window
print(client.generated_query)       # SQL from discover endpoints (in response body)
print(client.total_count_status)    # search/discover: "exact" | "estimate" | "page"

On search/discover, total_count_status tells you how the response total was derived: exact (true count), estimate (the query planner’s row estimate, returned on broad queries where an exact count is too expensive, so treat it as a magnitude), or page (count unavailable; total is just the returned page size).

Field Filtering

Control which fields are returned in the response.

# Only return specific fields
person = client.person.enrich(
    email="jane@example.com",
    include_fields="full_name,job_title,company_name,linkedin_url",
)

# Exclude fields
person = client.person.enrich(
    email="jane@example.com",
    exclude_fields="phones,locations,education",
)

# Require fields (skip records missing these)
person = client.person.enrich(
    email="jane@example.com",
    required_fields="work_email,mobile_phone",
)

Error Handling

The SDK raises a typed exception for each error status. Every one inherits from DataLegionError, and the classes for 400, 403, 404, 408 and 5xx also inherit from APIError.

from datalegion import (
    DataLegion,
    NotFoundError,
    AuthenticationError,
    InsufficientCreditsError,
    BadRequestError,
    ValidationError,
    RateLimitError,
    APIError,
)

client = DataLegion(api_key="legion_...")

try:
    person = client.person.enrich(email="jane@example.com")
except NotFoundError:
    # 404 - No match (uses no credits)
    print("No match")
except AuthenticationError:
    # 401 - Invalid or missing API key
    print("Check your API key")
except InsufficientCreditsError as e:
    # 402 - Out of credits
    print(f"Out of credits: {e.message}")
except (BadRequestError, ValidationError) as e:
    # 400 / 422 - Invalid request parameters
    print(f"Invalid input: {e.message} {e.details}")
except RateLimitError as e:
    # 429 - Rate limit exceeded
    print(f"Rate limited, retry after {client.retry_after}s")
except APIError as e:
    # Any other error status
    print(f"API error ({e.status_code}): {e.message}")

Status	Exception
400	`BadRequestError`
401	`AuthenticationError`
402	`InsufficientCreditsError`
403	`PermissionDeniedError`
404	`NotFoundError`
408	`RequestTimeoutError`
422	`ValidationError`
429	`RateLimitError`
5xx	`ServerError`
Other	`APIError`

The client doesn’t retry. On a 429, client.retry_after holds the seconds to wait.

Sandbox

Pass sandbox=True to build and test without spending credits. Enrich, bulk enrich, search and discover then call the sandbox and return sample records with every field a real record has; the utilities, which are free, call the real API.

client = DataLegion(api_key="legion_...", sandbox=True)
person = client.person.enrich(email="anyone@example.com")
print(person.legion_id)  # "sandbox_person_001"

Configuration

client = DataLegion(
    api_key="legion_...",                    # Or set DATALEGION_API_KEY
    base_url="https://api.datalegion.ai",         # Default
    timeout=60.0,                                 # Seconds, default 60
)

You can also pass a custom httpx client for advanced configuration:

import httpx

custom_client = httpx.Client(
    timeout=30.0,
    headers={"X-Correlation-ID": "my-request-123"},
)

client = DataLegion(httpx_client=custom_client)

Getting Started

Person Data

Company Data

Data Quality

Data Standardization

SDKs

Integrations

Installation

Authentication

Person

Enrichment

Search

Discovery

Company

Enrichment

Search

Discovery

Utility Endpoints

Clean Fields

Hash Email

Validate Data

Report a Data Issue

Async Client

Response Metadata

Field Filtering

Error Handling

Sandbox

Configuration

Resources

​Installation

​Authentication

​Person

​Enrichment

​Search

​Discovery

​Company

​Enrichment

​Search

​Discovery

​Utility Endpoints

​Clean Fields

​Hash Email

​Validate Data

​Report a Data Issue

​Async Client

​Response Metadata

​Field Filtering

​Error Handling

​Sandbox

​Configuration

​Resources

Installation

Authentication

Person

Enrichment

Search

Discovery

Company

Enrichment

Search

Discovery

Utility Endpoints

Clean Fields

Hash Email

Validate Data

Report a Data Issue

Async Client

Response Metadata

Field Filtering

Error Handling

Sandbox

Configuration

Resources