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

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

board   = crawlsnap.sport_snap.livescores()                       # live-score board
fixtures = crawlsnap.sport_snap.matches("TR")                     # fixtures for a region
match   = crawlsnap.sport_snap.match(5542814)                     # full match view
league  = crawlsnap.sport_snap.competition("england", "premier-league")
team    = crawlsnap.sport_snap.club_team("spain", "barcelona")
results = crawlsnap.sport_snap.search_all("messi")                # url fields feed other calls
player  = crawlsnap.sport_snap.player("messi", 123)

serp    = crawlsnap.serp_api.search("kubernetes operator")   # ranked Google results
for hit in serp.results:
    print(hit.rank, hit.domain, hit.url)

Every method takes its lookup value(s) as positional arguments and accepts raw_response=True (see below).

sport_snap is the football (soccer) data resource. It exposes the full source surface: live scores with in-match events (livescores), fixture lists (matches, matches_extended), single-match views (match, match_extended, match_stats, match_commentaries, match_channels, match_extra_broadcasts), the competition catalog and detail (competitions, competition, competition_tables, competition_tv_rights, competition_twitter), teams (popular_teams, all_teams, national_team, club_team), TV channel directories (all_channels, channels, channel_info, channel_repeats), a news feed (news, news_by_tag, news_article, competition_news, national_team_news, club_team_news), full-text search (search_all, search_teams, search_competitions, search_matches, search_players, popular_searches), and player profiles (player). Path segments (competition country/slug, team country/team, player slug/id) come from the url fields returned by list, search, and detail payloads. iso_code is an optional two-letter region code that resolves region-specific broadcast channels; pass "" (the default) for the server default.

serp_api.search runs one Google search and returns a single result page: ranked organic results plus the related searches Google suggests. Each result's url is the real target URL — never a search-engine redirector — so you can follow or store it directly. Refine with count, page, language, country, safe, time_range, site and filetype; anything you leave out keeps the API's own default. One call is one page of about ten results, so ask for page=2 rather than a larger count — count caps what is parsed out of the page you requested, it does not fetch more of them.

page = crawlsnap.serp_api.search(
    "actions", site="github.com", time_range="month", count=20
)
print(page.query, page.elapsed_ms)
for hit in page.results:
    print(f"{hit.rank}. {hit.title} — {hit.url}")
print("related:", page.suggestions)

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)

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)

sport_snap.channel_repeats returns one ~20-fixture page around the current date; walk forward (or backward) with its fixtures_next / fixtures_prev cursors — None means no further page:

page = crawlsnap.sport_snap.channel_repeats("dazn-spain", "ES")
while page.fixtures_next:
    page = crawlsnap.sport_snap.channel_repeats("dazn-spain", "ES", cursor=page.fixtures_next)

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.

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

Changelog

Release notes for every published version live in CHANGELOG.md and are mirrored on the GitHub Releases page. Each entry records the crawlsnap-contracts version it was generated from.