datalegion 0.3.3

Python client for the Data Legion API

Data Legion Python SDK

     

The official Python client for the Data Legion API.

Installation

pip install datalegion

Quick Start

from datalegion import DataLegion

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

Or set the DATALEGION_API_KEY environment variable and omit the api_key argument:

client = DataLegion()

Person Enrichment

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="[e-mail]")
print(person.full_name)
print(person.job_title)
print(person.company_name)

# Multiple results
results = client.person.enrich(
    email="[e-mail]",
    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")

Person Search

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)

Person Discover

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)

Company Enrichment

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)

Company Search

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)

Company Discover

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)

Utilities

Clean & Normalize Fields

cleaned = client.utility.clean(
    fields={
        "email": "[e-mail]",
        "phone": "[phone]",
        "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="[e-mail]")
print(hashed.hashes["sha256"])
print(hashed.hashes["md5"])

Validate Data

validated = client.utility.validate(
    email="[e-mail]",
    phone="[phone]",
    first_name="John",
)
print(validated.valid)
for error in validated.errors:
    print(f"{error.field}: {error.error}")

Report a Data Issue

Report an inaccuracy in a person or company record against its legion_id. Free; see the docs 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)

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

Health Check

health = client.health()
print(health.status)

Async Usage

import asyncio
from datalegion import AsyncDataLegion

async def main():
    client = AsyncDataLegion(api_key="legion_...")

    person = await client.person.enrich(email="[e-mail]")
    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="[e-mail]")

Response Types

All responses are Pydantic models with full type hints and autocomplete support. Import them directly:

from datalegion import PersonResponse, CompanyResponse, PersonMatchesResponse

Response Headers

After each request, these attributes are updated on the client:

person = client.person.enrich(email="[e-mail]")
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-ID

Error Handling

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

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

try:
    person = client.person.enrich(email="[e-mail]")
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 message
• status_code - HTTP status code
• error - Error type/code from the API
• details - Additional error details (if any)

Configuration

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

Documentation

For full API documentation, visit https://www.datalegion.ai/docs.