crawlsnap 0.5.0

Official Python SDK for CrawlSnap — the data intelligence API platform

crawlsnap

Official Python SDK for CrawlSnap — a data intelligence platform that delivers structured, on-demand data through fast, typed APIs. Authenticate once and call any CrawlSnap data product, with first-class types, automatic retries, and pagination built in.

• Idiomatic, fully typed client (httpx + pydantic v2)
• Sync and async: CrawlSnap or AsyncCrawlSnap — identical surface
• crawlsnap.init(...) singleton or an explicit client
• Resource namespacing: crawlsnap.vector_snap.ip(...)
• Per-API version pinning with a stable default — your calls never silently jump to a new API version
• Returns typed data; raises typed exceptions — no envelope bookkeeping
• Built-in retries with exponential backoff, configurable timeout, auto-pagination

---

Installation

pip install crawlsnap

Requires Python 3.8+.

Authentication

Get an API key (sk-cs-...) from your CrawlSnap dashboard. Provide it either via the environment or explicitly:

export CRAWLSNAP_API_KEY=sk-cs-...

crawlsnap.init()                       # reads CRAWLSNAP_API_KEY
crawlsnap.init(api_key="sk-cs-...")    # or pass it explicitly

The key is sent as Authorization: Bearer sk-cs-.... Treat it like a password.

Quick start

import crawlsnap

crawlsnap.init(api_key="sk-cs-...")

ip = crawlsnap.vector_snap.ip("8.8.8.8")
print(ip.reputation, ip.as_owner, ip.country)

Each call returns the typed enrichment payload directly, and raises a typed exception on failure — you never inspect an is_success envelope yourself.

Async

AsyncCrawlSnap is the awaitable twin of CrawlSnap — same constructor, same resources, same retries and typed errors. Ideal for enriching many indicators concurrently:

import asyncio
from crawlsnap import AsyncCrawlSnap

async def main():
    async with AsyncCrawlSnap(api_key="sk-cs-...") as client:
        ip, dom = await asyncio.gather(
            client.vector_snap.ip("8.8.8.8"),
            client.vector_snap.domain("google.com"),
        )
        print(ip.as_owner, dom.reputation)

        async for subdomain in client.subdo_snap.scan_iter("example.com"):
            print(subdomain)

asyncio.run(main())

Every method is a coroutine; scan_iter is an async generator (use async for).

Resources

Resource	Methods	Returns
vector_snap	url · hash · ip · domain	reputation, detections, categories, relationships
pulse_snap	url · hash · ip · domain	threat-intelligence pulse (and sandbox) summary
subdo_snap	scan · scan_iter	enumerated subdomains (paginated)
sport_snap	channel · channel_schedule · match · country_channels · daily_schedule	live football TV listings: channels, schedules, match details

url    = crawlsnap.vector_snap.url("https://example.com")
file   = crawlsnap.vector_snap.hash("44d88612fea8a8f36de82e1278abb02f")
domain = crawlsnap.vector_snap.domain("google.com")

pulse  = crawlsnap.pulse_snap.ip("8.8.8.8")

channel  = crawlsnap.sport_snap.channel("bein-connect-turkey")
schedule = crawlsnap.sport_snap.channel_schedule("bein-connect-turkey")
match    = crawlsnap.sport_snap.match(5542814)
channels = crawlsnap.sport_snap.country_channels("turkey")
day      = crawlsnap.sport_snap.daily_schedule("2026-07-05")  # or datetime.date

Every method takes its lookup value as the first positional argument and accepts raw_response=True (see below).

sport_snap covers live football (soccer) TV listings: TV channel metadata and broadcast rights, channel broadcast schedules, match details with per-country broadcast coverage (score, events, statistics, and lineups for finished matches), country channel directories, and daily schedules grouped by competition. match.status is scheduled, live, or finished and discriminates how much of the payload is populated. Match ids are discovered via daily_schedule and channel_schedule entries.

API versioning

Each CrawlSnap data product is versioned independently, and the version is just a value the SDK puts in the request path — not something baked into your call site. A direct resource call targets that product's stable default version for this SDK release:

crawlsnap.vector_snap.ip("8.8.8.8")        # default VectorSnap version (stable)

The default is pinned per SDK release and never moves on its own: upgrading the SDK does not silently retarget your calls at a newer API version. When a product ships a new version, you opt in explicitly — per product, without touching the others:

crawlsnap.vector_snap.v1.ip("8.8.8.8")     # explicitly VectorSnap v1
crawlsnap.pulse_snap.url("https://x.com")  # unaffected — PulseSnap default

The same applies on an explicit client (client.vector_snap.v1.ip(...)) and on AsyncCrawlSnap. When a product's default version is bumped in a future SDK release, it ships as a deliberate, documented change — so you upgrade at your own pace.

Error handling

Failures raise a typed exception instead of returning an error envelope:

from crawlsnap import (
    NotFoundError, RateLimitError, QuotaExceededError,
    AuthenticationError, CrawlSnapError,
)

try:
    res = crawlsnap.vector_snap.domain("example.com")
except NotFoundError:
    ...                              # 404 — no data for this indicator
except QuotaExceededError as e:
    print(e.message)                 # 402 — out of credits / monthly quota
except RateLimitError as e:
    print(e.retry_after)             # 429 — daily limit; seconds to wait
except AuthenticationError:
    ...                              # 401 — missing / invalid key
except CrawlSnapError as e:          # base class for every SDK error
    print(e)

HTTP	Exception	Notes
400	BadRequestError	invalid indicator
401	AuthenticationError	missing / invalid key
402	QuotaExceededError	out of credits or monthly quota
403	SubscriptionInactiveError	subscription not active
404	NotFoundError	no data for the indicator
429	RateLimitError	daily limit; .retry_after (seconds)
5xx	ServerError	server / upstream failure
—	APIConnectionError / APITimeoutError	network failure / client timeout

Every status error carries .status_code, .message, and .request_id (share the request id with support to speed up debugging).

Pagination

subdo_snap is paginated. Stream every subdomain across all pages — the cursor is handled for you:

for subdomain in crawlsnap.subdo_snap.scan_iter("example.com"):
    print(subdomain)

Or page manually:

page = crawlsnap.subdo_snap.scan("example.com")
while page.cursor:
    page = crawlsnap.subdo_snap.scan("example.com", cursor=page.cursor)

Configuration

The singleton is a thin layer over the CrawlSnap client. For multiple keys, multiple environments, or thread isolation, instantiate it directly:

from crawlsnap import CrawlSnap

client = CrawlSnap(
    api_key="sk-cs-...",
    timeout=30.0,
    max_retries=3,
    base_url="https://api.crawlsnap.com",
)
ip = client.vector_snap.ip("1.1.1.1")
client.close()

AsyncCrawlSnap accepts the exact same options (api_key, base_url, timeout, max_retries); await client.close() or use async with.

Option	Default	Description
api_key	$CRAWLSNAP_API_KEY	Your sk-cs- key
base_url	$CRAWLSNAP_BASE_URL or https://api.crawlsnap.com	API host override
timeout	30.0	Per-request timeout (seconds)
max_retries	2	Retries for 429 / 5xx / connection errors

Retries use exponential backoff and honor the Retry-After header on 429.

Raw response

Pass raw_response=True to get the full envelope (status, headers, request id) instead of just the data:

raw = crawlsnap.vector_snap.ip("8.8.8.8", raw_response=True)
print(raw.status_code, raw.request_id, raw.is_success, raw.data)

Development

The typed models under crawlsnap/models/ are generated from the public OpenAPI contract; the client facade is hand-written. To refresh the models after the contract changes:

./scripts/regenerate.sh        # re-bundles the contract and regenerates models/

Run the tests:

pip install -e ".[dev]"
pytest