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.records_remaining, 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.records_remaining, 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)

# Live job postings across companies (cursor-paginated — see Pagination below).
jobs = client.jobs.search(
    job={
        "title": {"include": ["Head of Sales"]},
        "seniority": {"include": ["5-10"]},
        "employment_type": {"include": ["FULL_TIME"]},
        "work_arrangement": {"include": ["Hybrid", "Remote OK"]},
        "date_posted": {"last_days": 30},
    },
    company={
        "industry": {"include": ["Software Development"]},
        "size": {"include": ["51-200", "201-500"]},
    },
    max_results=25,
)
for job in jobs.results:
    print(job.company_name, job.title, job.location.city if job.location else None)

# Postings at one company (cursor-paginated).
company_jobs = client.jobs.company(
    company_linkedin_url="https://www.linkedin.com/company/openai",
    job={"field": {"include": ["Software Engineering"]}},
    max_results=25,
)
for job in company_jobs.results:
    print(job.title, job.url)

# Companies hiring for a role (deduplicated), each with its matching-posting count.
tam = client.company.tam_by_jobs(
    job={
        "title": {"include": ["Account Executive"]},
        "min_per_company": 3,
    },
    company={"industry": {"include": ["Software Development"]}},
    max_results=50,
)
for match in tam.results:
    print(match.company.name if match.company else None, match.matched_jobs)

# 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)

# Company employees grouped by country.
by_country = client.enrichment.company_distribution_by_country(
    company_linkedin_url="https://www.linkedin.com/company/openai",
)
for row in by_country.distribution:
    print(row.country, row.count, row.percentage_ratio)

# Company employees grouped by department.
by_department = client.enrichment.company_distribution_by_department(
    company_linkedin_url="https://www.linkedin.com/company/openai",
)
for row in by_department.distribution:
    print(row.department, row.count, row.percentage_ratio)

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

# Public API changelog, newest first. No API key required, and not paginated.
entries = client.changelog.list(days=30, limit=10)
for entry in entries:
    print(entry.date, entry.type, entry.title, entry.affected_endpoints)

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

result = client.enrichment.email(person_linkedin_url="https://www.linkedin.com/in/example-person")
if result.fair_usage is not None:
    print(result.fair_usage.records_used)        # records this call consumed
    print(result.fair_usage.records_remaining)   # a number, or "unlimited"
    print(result.fair_usage.next_reset_at)       # None on an unlimited plan
    print(result.fair_usage.request_id)          # quote this to support
    if result.fair_usage.rate_limit is not None: # omitted where the API doesn't send it
        print(result.fair_usage.rate_limit.remaining_this_second)

# Pages carry the block from the request that fetched them.
for page in client.search.people(people={"job_level": ["VP"]}).iter_pages(max_pages=5):
    print(page.fair_usage.records_used if page.fair_usage else None)

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, InsufficientRecordsError,
    NotFoundError, RateLimitError, APIStatusError, APIConnectionError,
    APITimeoutError, APIResponseValidationError,
)

try:
    client.enrichment.email(person_linkedin_url="...")
except InsufficientRecordsError as err:
    ...                            # 402: out of records; err.fair_usage.next_reset_at
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]},
)