ipscope - Official IPScope Server-Side Python SDK

Official, strictly typed server-side Python SDK for the IPScope IP Intelligence Platform, powered by httpx and supporting both Synchronous (IpScopeClient) and Asynchronous (AsyncIpScopeClient) execution.

Scope Notice: This SDK is intentionally scoped strictly to IP intelligence, network prefix, dataset metadata, feedback, and health operations. It contains zero account, billing, wallet, or key-management operations, making it safe to deploy in production backend microservices, workers, and API gateways.

Features

• Both Sync & Async Clients Powered by httpx:
  • IpScopeClient: Thread-safe synchronous client backed by persistent httpx.Client connection pooling (client = IpScopeClient(api_key="...")).
  • AsyncIpScopeClient: Coroutine-safe asynchronous client backed by httpx.AsyncClient (client = AsyncIpScopeClient(api_key="...")) with singleflight request coalescing (asyncio.Task deduplication) and bounded concurrency (asyncio.Semaphore).
• 100% Type-Safe (PEP 561 py.typed): Uses @dataclass(frozen=True, slots=True) models for all responses (IpLookupResponse, GeoResponseBlock, NetworkResponseBlock, SecurityResponseBlock, ConnectionResponseBlock, SecurityExtensionsResponseBlock, BatchLookupResponse, NetworkPrefixLookupResponse, DatasetsMetadataResponse, FeedbackSubmissionResponse, ServiceStatusResponse).
• Low-RAM & Zero Memory Leaks:
  • Slotted frozen dataclasses (slots=True) eliminate per-instance __dict__ overhead (~55% lower RAM per record).
  • Bounded OrderedDict LRU + TTL cache (max_entries=512 default, < 400 KB worst-case footprint).
  • Hard 2 MiB streaming response ceiling (MAX_RESPONSE_BYTES) via httpx streaming prevents OOM on malformed upstream payloads.
  • Bounded httpx.Limits(max_connections=20, max_keepalive_connections=10, keepalive_expiry=30.0).
  • lookup_batch_stream(...) generator / async generator yields batch chunks incrementally without holding large arrays in memory.

Installation

uv add ipscope
# or
pip install ipscope

Development (uv + ruff)

cd sdks/python
uv sync
uv run ruff check --fix .
uv run ruff format .
uv run pytest -v

Quick Start (Synchronous Client)

API keys have the form <public>:<private>. Both clients are context managers that close their connection pool on exit (or call close() yourself).

from ipscope import IpScopeClient

large_ip_list = [f"10.0.{i // 256}.{i % 256}" for i in range(1_000)]

with IpScopeClient(api_key="<public>:<private>") as client:
    # 1. Single IPv4 or IPv6 lookup (cached automatically in LRU cache)
    ip_info = client.lookup_ip("8.8.8.8")
    print(ip_info.ip, ip_info.geo.country_code, ip_info.network.asn)
    print("Coordinates:", ip_info.geo.latitude, ip_info.geo.longitude)
    print("VPN:", ip_info.security.is_vpn, "Relay:", ip_info.security.is_relay)
    if ip_info.security_extensions is not None:
        # 0-100 (needs the security_extensions field family on your plan)
        print("Abuse velocity:", ip_info.security_extensions.abuse_velocity_score)

    # 2. Batch lookup (up to 100 IPs; deduplicates & reuses cached entries automatically)
    batch = client.lookup_batch(["8.8.8.8", "1.1.1.1", "2606:4700:4700::1111"])
    for record in batch.results:
        print(record.ip, record.network.organization)

    # 3. Memory-bounded streaming batch lookup for 10,000+ IPs
    for chunk in client.lookup_batch_stream(large_ip_list):
        for record in chunk.results:
            print(record.ip, record.geo.country_code)

    # 4. Network CIDR prefix lookup
    prefix = client.lookup_network("8.8.8.0/24")
    print(prefix.queried_cidr, prefix.matched_network, prefix.bgp_prefix_length)
    if prefix.record is not None:
        print(prefix.record.asn, prefix.record.country_code, prefix.record.latitude)
    if prefix.lookup is not None:
        print(prefix.lookup.geo.city)

Quick Start (Asynchronous Client)

import asyncio

from ipscope import AsyncIpScopeClient


async def ips():
    for i in range(1_000):
        yield f"10.0.{i // 256}.{i % 256}"


async def main() -> None:
    async with AsyncIpScopeClient(api_key="<public>:<private>") as client:
        # 1. Concurrent singleflight-coalesced lookups
        ip_info = await client.lookup_ip("1.1.1.1")
        print(ip_info.ip, ip_info.network.organization)

        # 2. Chunked async batch lookup with bounded concurrency (default max 4 in-flight)
        batch = await client.lookup_batch_chunked(
            ["8.8.8.8", "1.1.1.1", "9.9.9.9"],
            concurrency=4,
        )
        print("Resolved:", batch.count)

        # 3. Async streaming over any Iterable or AsyncIterable, O(chunk) memory
        async for chunk in client.lookup_batch_stream(ips()):
            for record in chunk.results:
                print(record.ip)


asyncio.run(main())

Rate Limits

Authenticated responses carry X-RateLimit-Limit, X-RateLimit-Remaining, X-RateLimit-Reset (seconds until the window resets) and X-RateLimit-Scope.

from ipscope import IpScopeClient

with IpScopeClient(api_key="<public>:<private>") as client:
    result = client.lookup_ip_with_meta("8.8.8.8")
    rl = result.rate_limit
    print(rl.limit_per_minute, rl.remaining, rl.reset_after_seconds, rl.scope)

On 429 the client waits for Retry-After (or, failing that, X-RateLimit-Reset), capped by RetryOptions.max_delay_seconds, then retries. reset_epoch_seconds is a deprecated alias of reset_after_seconds (same value, in seconds); burst_capacity and rate_limit_rps are deprecated and always None.

Feedback

from ipscope import FeedbackSubmissionInput, IpScopeClient

with IpScopeClient(api_key="<public>:<private>") as client:
    fb = client.submit_feedback(
        FeedbackSubmissionInput(
            prefix_or_ip="203.0.113.0/24",
            role="network_owner",  # "network_owner" | "user" | "researcher"
            issue_type="wrong_city",
            requested_correction={"city": "Amsterdam", "country_code": "NL"},
            evidence_url="https://example.net/geofeed.csv",
            contact_consent=True,
            contact_email="noc@example.net",
        )
    )
    print(fb.id, fb.status, fb.created_at)

issue_type is one of wrong_country, wrong_city, false_positive_vpn, false_positive_hosting, false_positive_tor, virtual_location_mismatch, other.

Error Handling

from ipscope import (
    IpScopeClient,
    IpScopeError,
    IpScopeValidationError,
    IpScopeAuthenticationError,
    IpScopeRateLimitError,
    IpScopeTimeoutError,
    IpScopeApiError,
)

client = IpScopeClient(api_key="<public>:<private>")

try:
    res = client.lookup_ip("8.8.8.8")
except IpScopeValidationError as err:
    # Invalid IP / CIDR / batch size before any network request is made
    print("Validation failed:", err.field, err)
except IpScopeAuthenticationError as err:
    # 401 / 403 invalid or revoked API key
    print("Auth error:", err.status, err.code)
except IpScopeRateLimitError as err:
    # 429 after retries exhausted
    print("Rate limited, retry after s:", err.retry_after_seconds, err.rate_limit)
except IpScopeTimeoutError as err:
    print("Timed out after s:", err.timeout_seconds)
except IpScopeApiError as err:
    print("API error:", err.status, err.code)
except IpScopeError as err:
    print("SDK error:", err)