Python SDK

pip install blitz-api-py
# or: uv add blitz-api-py

from blitz_api import BlitzAPI
from blitz_api.types import Industry, JobLevel

# api_key defaults to the BLITZ_API_KEY environment variable.
with BlitzAPI() as client:
    # Health-check the key before a batch job.
    info = client.account.key_info()
    print(info.valid, info.remaining_credits, info.max_requests_per_seconds)

    # LinkedIn profile URL -> verified work email.
    email = client.enrichment.email(
        person_linkedin_url="https://www.linkedin.com/in/example-person",
    )
    if email.found:
        print(email.email)

    # Search people with typed, autocompleted filters.
    people = client.search.people(
        company={"industry": {"include": [Industry.SOFTWARE_DEVELOPMENT]}},
        people={"job_level": [JobLevel.VP]},
        max_results=10,
    )
    for person in people.results:
        print(person.full_name, person.headline)

client = BlitzAPI(api_key="sk_...")   # explicit
client = BlitzAPI()                    # reads BLITZ_API_KEY

import asyncio
from blitz_api import AsyncBlitzAPI

async def main() -> None:
    async with AsyncBlitzAPI() as client:
        result = await client.enrichment.company(
            company_linkedin_url="https://www.linkedin.com/company/openai",
        )
        print(result.company.name if result.company else None)

asyncio.run(main())

info = client.account.key_info()
print(info.valid, info.remaining_credits, info.max_requests_per_seconds)
print(info.allowed_apis)

# People across many companies (cursor-paginated — see Pagination below).
people = client.search.people(
    company={
        "industry": {"include": ["IT Services and IT Consulting"]},
        "employee_range": ["51-200", "201-500"],
        "hq": {"sales_region": ["EMEA"]},
    },
    people={
        "job_level": ["VP", "Director"],
        "job_function": ["Sales & Business Development"],
        "min_connections": 200,
    },
    max_results=25,
)
for person in people.results:
    print(person.full_name, person.headline, person.linkedin_url)

# Companies by ICP (cursor-paginated).
companies = client.search.companies(
    company={
        "keywords": {"include": ["SaaS"]},
        "industry": {"include": ["Software Development"]},
        "hq": {"country_code": ["FR", "DE"]},
        "employee_range": ["51-200", "201-500"],
    },
    max_results=25,
)
for company in companies.results:
    print(company.name, company.industry, company.employees_on_linkedin)

# All employees at one company (page-paginated).
employees = client.search.employee_finder(
    company_linkedin_url="https://www.linkedin.com/company/openai",
    job_level=["C-Team", "VP", "Director"],
    job_function=["Sales & Business Development"],
    sales_region=["NORAM"],
    max_results=50,
)
print(f"Page {employees.page} of {employees.total_pages}")
for person in employees.results:
    print(person.full_name, person.headline)

# Single best decision-maker via a priority cascade.
match = client.search.waterfall_icp(
    company_linkedin_url="https://www.linkedin.com/company/openai",
    cascade=[
        {"include_title": ["CTO", "VP Engineering"], "location": ["WORLD"], "include_headline_search": False},
        {"include_title": ["Engineering Director", "Engineering Manager"], "location": ["WORLD"], "include_headline_search": False},
    ],
    max_results=5,
)
for item in match.results:
    print(f"[Tier {item.icp} | Rank #{item.ranking}]", item.person.full_name)

# LinkedIn profile URL -> verified work email.
email = client.enrichment.email(person_linkedin_url="https://www.linkedin.com/in/example-person")
if email.found:
    print(email.email)

# LinkedIn profile URL -> direct phone (US only).
phone = client.enrichment.phone(person_linkedin_url="https://www.linkedin.com/in/example-person")
if phone.found:
    print(phone.phone)

# Work email -> full person profile.
by_email = client.enrichment.email_to_person(email="jane.doe@acme.com")
if by_email.found:
    print(by_email.person.full_name)

# Phone number -> full person profile (US only).
by_phone = client.enrichment.phone_to_person(phone="+14155551234")
if by_phone.found:
    print(by_phone.person.full_name)

# Company LinkedIn URL -> full company profile.
company = client.enrichment.company(company_linkedin_url="https://www.linkedin.com/company/openai")
print(company.company.name, company.company.industry, company.company.employees_on_linkedin)

# Website domain -> Company LinkedIn URL.
to_li = client.enrichment.domain_to_linkedin(domain="openai.com")
if to_li.found:
    print(to_li.company_linkedin_url)

# Company LinkedIn URL -> email domain.
to_domain = client.enrichment.linkedin_to_domain(company_linkedin_url="https://www.linkedin.com/company/openai")
if to_domain.found:
    print(to_domain.email_domain)

# Current server date/time.
now = client.utils.current_date()

# Company employees grouped by country.
distribution = client.utils.company_employment_distribution(
    company_linkedin_url="https://www.linkedin.com/company/openai",
)

# Stream every match across all pages — no cursor handling needed.
for person in client.search.people(people={"job_level": ["VP"]}):
    print(person.full_name)

client = BlitzAPI(
    api_key=None,                 # falls back to BLITZ_API_KEY
    base_url="https://api.blitz-api.ai",
    timeout=30.0,                 # seconds, or an httpx.Timeout
    max_retries=3,                # retries on 429 / 5xx / network errors
    rate_limit_rps=5.0,           # client-side throttle, per endpoint; None to disable
)

client.search.people(people={"job_level": ["VP"]}, timeout=10.0)

from blitz_api import (
    BlitzError, AuthenticationError, InsufficientCreditsError,
    NotFoundError, RateLimitError, APIStatusError, APIConnectionError,
    APITimeoutError, APIResponseValidationError,
)

try:
    client.enrichment.email(person_linkedin_url="...")
except InsufficientCreditsError:
    ...                            # 402 — out of credits
except AuthenticationError:
    ...                            # 401 — bad key
except APIStatusError as err:
    print(err.status_code, err.message, err.body)
except APIResponseValidationError:
    ...                            # 2xx whose body didn't match the model
except BlitzError:
    ...                            # base class for everything this SDK raises

from blitz_api.types import Industry, JobLevel, Continent

client.search.people(
    company={"industry": {"include": [Industry.SOFTWARE_DEVELOPMENT]}},
    people={"job_level": [JobLevel.VP]},
)