Python SDK

status302 is the official Python client. It has one typed method per public operation, with TypedDict models generated from the same OpenAPI spec as the API reference; the package version is the spec version it was generated from.

pip install status302   # or: uv add status302

It needs Python 3.11 or newer and is built on httpx. It ships type information (py.typed) and is checked with mypy in strict mode.

Clients

Status302 is blocking, AsyncStatus302 is for asyncio; both have the same methods. Use them as context managers so connections are closed:

samples/python/first-request.py

# The next five upcoming events in the catalog.
import os

from status302 import Status302

with Status302(os.environ["STATUS302_API_KEY"], base_url=os.environ.get("STATUS302_API_URL", "https://api.status302.com")) as status302:
    page = status302.list_events(limit=5)
    for event in page["items"]:
        city = (event.get("venue") or {}).get("city") or ""
        print(event["id"], event["name"], event["startsAt"] or "date TBA", city)

Argument	Default
api_key		Your API key, sent as Authorization: Bearer.
base_url	https://api.status302.com	Another environment’s API origin.
max_retries	3	Retries after 429, 5xx (idempotent methods) and connection errors.
max_retry_after	60	The longest Retry-After in seconds the client waits out before raising RateLimitError.
timeout	30.0	Timeout per attempt, in seconds.
http_client		Your own httpx.Client / httpx.AsyncClient, for proxies or instrumentation.

Requests

Methods are named after the operation ids in snake case: list_events, get_event_metrics, list_sales_feed, create_saved_list, and so on. Path parameters are positional, query parameters and bodies are keyword arguments, and responses are typed dictionaries:

metrics = status302.get_event_metrics(4102)
for row in metrics["items"]:
    print(row["scope"], row["floor"], row["currency"])

status302.request(method, path, ...) returns an APIResponse with data, status, headers and the parsed rate_limit, for anything the typed methods don’t cover. The models live in status302.types.

Pagination

paginate(method, **params) yields every item and follows nextCursor; pages(...) yields whole pages.

samples/python/pagination.py

# Every upcoming event, 200 per page. `paginate` follows nextCursor for you;
# `pages` yields whole pages when you want to checkpoint the cursor.
import os

from status302 import Status302

with Status302(os.environ["STATUS302_API_KEY"], base_url=os.environ.get("STATUS302_API_URL", "https://api.status302.com")) as status302:
    total = 0
    for event in status302.paginate(status302.list_events, limit=200):
        total += 1
        if total <= 3:
            print(event["id"], event["name"])
    print(f"{total} events")

    for page in status302.pages(status302.list_events, limit=200):
        print(f"page of {len(page['items'])}, next cursor: {page['nextCursor'] or 'none (last page)'}")

Errors

Non-2xx responses raise a subclass of APIError carrying the problem as problem, with status and type:

Class	Status
BadRequestError	400
AuthenticationError	401
PlanRequiredError	402, with required_plan, current_plan and feature (problem["limit"] for count limits)
PermissionDeniedError	403, with required_scope for a missing scope
NotFoundError	404
ConflictError	409
RateLimitError	429, with retry_after and rate_limit
ServerError	5xx

Network failures raise APIConnectionError. All of them derive from Status302Error.

samples/python/errors.py

# Every non-2xx answer becomes a typed error carrying the problem+json body.
import os

from status302 import (
    AuthenticationError,
    NotFoundError,
    PermissionDeniedError,
    PlanRequiredError,
    RateLimitError,
    Status302,
)

with Status302(os.environ["STATUS302_API_KEY"], base_url=os.environ.get("STATUS302_API_URL", "https://api.status302.com")) as status302:
    try:
        status302.get_event(999_999_999)
    except NotFoundError as err:
        print(f"404 {err.type}: {err.problem.get('detail') or err.problem['title']}")
    except PlanRequiredError as err:
        print(f"needs the {err.required_plan} plan (you have {err.current_plan}) for {err.feature}")
    except PermissionDeniedError as err:
        print(f"forbidden: the key needs the {err.required_scope} scope" if err.required_scope else f"forbidden: {err}")
    except RateLimitError as err:
        print(f"quota exhausted, retry in {err.retry_after}s")
    except AuthenticationError:
        print("the API key is missing, revoked or expired")

Rate limits and retries

429 responses are retried for every method after Retry-After; 5xx responses and connection errors are retried with exponential backoff for GET, PUT and DELETE only. Each response’s RateLimit headers are parsed into rate_limit. See rate limits.

samples/python/rate-limits.py

# Reading the quota headers, and handling 429 without the SDK.
import os
import time
from typing import Any

import httpx

from status302 import Status302

api = os.environ.get("STATUS302_API_URL", "https://api.status302.com")
headers = {"Authorization": f"Bearer {os.environ['STATUS302_API_KEY']}"}


def get_json(url: str, attempts: int = 5) -> Any:
    """Plain httpx: retry a 429 after Retry-After seconds, a bounded number of times."""
    for attempt in range(1, attempts + 1):
        res = httpx.get(url, headers=headers)
        print("RateLimit-Policy:", res.headers.get("ratelimit-policy"))
        print("RateLimit:", res.headers.get("ratelimit"))
        if res.status_code == 429 and attempt < attempts:
            wait = int(res.headers.get("retry-after", "1"))
            print(f"rate limited, retrying in {wait}s")
            time.sleep(wait)
            continue
        res.raise_for_status()
        return res.json()


get_json(f"{api}/v1/events?limit=1")

# The SDK retries 429s itself and parses the headers on every response.
with Status302(os.environ["STATUS302_API_KEY"], base_url=api) as status302:
    res = status302.request("GET", "/v1/events", query={"limit": 1})
    if res.rate_limit:
        rl = res.rate_limit
        print(f"{rl.remaining} requests left in the {rl.policy} window, resets in {rl.reset}s")
        for p in rl.policies:
            print(f"quota {p.name}: {p.quota} per {p.window}s")

Stream

AsyncStatus302.stream() returns an async context manager that you iterate for messages. It reconnects with backoff, resubscribes every topic with since set to the last seq it delivered, drops duplicate seqs, and yields a ResyncMessage when a gap couldn’t be replayed; fetch_snapshot loads the REST snapshot for it.

samples/python/stream.py

# Your organization's alert firings, live. The client reconnects on its own,
# resumes after the last seq it delivered, and drops duplicates.
import asyncio
import os

from status302 import AsyncStatus302, DataMessage, ErrorMessage, ResyncMessage


async def main() -> None:
    api = os.environ.get("STATUS302_API_URL", "https://api.status302.com")
    async with AsyncStatus302(os.environ["STATUS302_API_KEY"], base_url=api) as status302:
        async with status302.stream() as stream:
            alerts = stream.subscribe("alerts")
            print(f"subscribed to alerts at seq {await alerts.ready}", flush=True)
            async for message in stream:
                if isinstance(message, DataMessage):
                    print(f"alert #{message.seq} at {message.ts}: {message.data['title']}", flush=True)
                elif isinstance(message, ResyncMessage):
                    # The gap could not be replayed: reload the REST snapshot, then continue live.
                    snapshot = await stream.fetch_snapshot(message)
                    print(f"resynced {message.topic} ({message.reason}) from {message.snapshot}", snapshot)
                elif isinstance(message, ErrorMessage):
                    print(f"subscription refused: {message.code}: {message.message}")


asyncio.run(main())

• stream.subscribe(topic, filter=..., since=...) returns a Subscription; await sub.ready gives the head seq.
• Iteration yields DataMessage (topic, seq, ts, data), ResyncMessage (topic, reason, snapshot) and ErrorMessage (code, message) when the server refuses a subscription.
• On the server the client sends your key on the upgrade. Pass auth="ticket" and get_ticket= to connect with tickets instead.
• Close codes 4401 and 4403 are final and end the iteration with StreamClosedError; other closes reconnect.

Webhooks

verify_webhook(body, signature_header, secret) checks a delivery’s signature in constant time and its timestamp against a 5-minute tolerance (tolerance=), and returns the parsed envelope. Pass a list of secrets while rotating. It raises WebhookVerificationError for an invalid delivery. sign_webhook(body, secret) produces a valid header for tests.

samples/python/webhook-receiver.py

# A webhook receiver on the standard library's http.server. Verify the
# signature over the raw body before parsing, answer 2xx fast, and process each
# event `id` once (retries and redeliveries repeat it).
import os
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer

from status302 import WebhookVerificationError, verify_webhook

SECRET = os.environ["STATUS302_WEBHOOK_SECRET"]
processed: set[str] = set()  # use your database in production


class Handler(BaseHTTPRequestHandler):
    def do_POST(self) -> None:
        if self.path != "/webhooks/status302":
            self.send_response(404)
            self.end_headers()
            return
        body = self.rfile.read(int(self.headers.get("Content-Length", "0")))
        try:
            event = verify_webhook(body, self.headers.get("Status302-Signature"), SECRET)
        except WebhookVerificationError as err:
            print(f"rejected: {err}", flush=True)
            self.send_response(400)
            self.end_headers()
            return
        if event["id"] in processed:
            print(f"duplicate {event['type']} {event['id']} (delivery {self.headers.get('Status302-Delivery-Id')})", flush=True)
        else:
            processed.add(event["id"])
            print(f"verified {event['type']} {event['id']}{' (test)' if event.get('test') else ''}", flush=True)
        self.send_response(204)
        self.end_headers()

    def log_message(self, format: str, *args: object) -> None:
        pass  # keep the output to the events above


port = int(os.environ.get("PORT", "3000"))
server = ThreadingHTTPServer(("", port), Handler)  # binds and listens here
print(f"listening on :{port}", flush=True)
server.serve_forever()