beaconhq 0.1.0

Beacon Python client SDK — auto-capture HTTP request telemetry and ship it to the Beacon ingest API. FastAPI/Starlette, Flask, and Django adapters.

beaconhq — Beacon Python SDK

Auto-capture HTTP request telemetry from your FastAPI / Starlette, Flask, or Django app and ship it to Beacon. This is the Python port of @beaconhq/sdk — same product, same ingest contract, idiomatic Python.

Full docs: https://beacon.skyware.dev/docs

• Distribution name: beaconhq (pip install beaconhq)
• Import package: beaconhq (import beaconhq)
• Zero required dependencies for the core — it ships telemetry using only the Python standard library (urllib, threading, json, atexit). Framework support is opt-in via extras and imported lazily, so import beaconhq works with no web framework installed.
• Python 3.9+.

Install

pip install beaconhq                 # core only
pip install "beaconhq[fastapi]"      # + Starlette/FastAPI adapter
pip install "beaconhq[flask]"        # + Flask adapter
pip install "beaconhq[django]"       # + Django adapter

Quickstart

You only need a project ingest key — the client defaults to Beacon's hosted ingest endpoint (https://ingest.beacon.skyware.dev/v1/ingest).

from beaconhq import BeaconClient

beacon = BeaconClient(ingest_key="your-per-project-ingest-key")

Or configure from the environment (BEACON_INGEST_KEY, optional BEACON_INGEST_URL, optional BEACON_ENABLED):

from beaconhq import BeaconClient
beacon = BeaconClient.from_env()

FastAPI / Starlette

from fastapi import FastAPI
from beaconhq import BeaconClient
from beaconhq.integrations.asgi import BeaconASGIMiddleware

beacon = BeaconClient(ingest_key=KEY)

app = FastAPI()
app.add_middleware(BeaconASGIMiddleware, client=beacon, consumer_header="x-api-key")

@app.get("/users/{user_id}")
def get_user(user_id: int):
    return {"id": user_id}
# Captured route template: "/users/{user_id}"  (not the concrete "/users/123")

Flask

from flask import Flask
from beaconhq import BeaconClient
from beaconhq.integrations.flask import init_flask

beacon = BeaconClient(ingest_key=KEY)

app = Flask(__name__)
init_flask(app, beacon, consumer_header="X-API-Key")

@app.route("/users/<int:user_id>")
def get_user(user_id):
    return {"id": user_id}
# Captured route template: "/users/<int:user_id>"

Django

# settings.py
from beaconhq import BeaconClient

BEACON_CLIENT = BeaconClient(ingest_key="your-per-project-ingest-key")
BEACON_CONSUMER_HEADER = "X-API-Key"   # optional

MIDDLEWARE = [
    "beaconhq.integrations.django.BeaconDjangoMiddleware",   # put it near the top
    # ... your other middleware ...
]
# Captured route template: "/users/<int:pk>"  (from request.resolver_match.route)

If BEACON_CLIENT is missing the middleware disables itself cleanly (Django MiddlewareNotUsed) rather than erroring.

Self-hosting / overriding the endpoint

Point at your own Beacon ingest by passing ingest_url (a full endpoint, or a base URL to which /v1/ingest is appended):

beacon = BeaconClient(
    ingest_key=KEY,
    ingest_url="https://beacon.internal.example.com",  # /v1/ingest appended if omitted
)

How it works

capture() only appends to an in-memory buffer — it is non-blocking and adds negligible latency to your request path. A background daemon thread flushes the buffer to the ingest API every flush_interval seconds, or eagerly when it reaches batch_size. The client never raises into your app: serialization and network failures are caught, optionally reported via on_error, and the batch is re-queued (network error / 5xx) or dropped (4xx). Remaining events are flushed on interpreter exit via atexit.

Identifying the consumer (the "who called me" hook)

consumer is the identity of the caller (an API-key id, user id, tenant…). The simplest option is to read a header:

app.add_middleware(BeaconASGIMiddleware, client=beacon, consumer_header="x-api-key")

For anything richer, pass a consumer_resolver — a callable that receives the framework's request object and returns a string or None:

def resolve_consumer(request) -> str | None:
    # FastAPI/Starlette: `request` is the raw ASGI `scope` dict
    # Flask:             `request` is `flask.request`
    # Django:            `request` is the `HttpRequest`
    return getattr(getattr(request, "state", None), "api_key_id", None)

app.add_middleware(
    BeaconASGIMiddleware, client=beacon, consumer_resolver=resolve_consumer
)

You can also set a default resolver on the client itself (BeaconClient(..., consumer_resolver=...)) so every adapter uses it. Resolution precedence: adapter consumer_resolver → client consumer_resolver → header value. A resolver that raises is treated as "no consumer" — it never breaks the request.

Configuration reference

BeaconClient(...) — option names are the snake_case translation of the JS BeaconClientOptions; defaults match @beaconhq/sdk.

Option	Type	Default	Description
ingest_key	str	— (required)	Per-project ingest key, sent as Authorization: Bearer. The only required argument.
ingest_url	str	https://ingest.beacon.skyware.dev/v1/ingest	Full endpoint or base URL (/v1/ingest appended). Override for self-hosted Beacon.
batch_size	int	100	Buffered events that trigger an eager flush.
flush_interval	float	5.0	Background flush cadence, seconds (JS uses 5000 ms).
max_buffer_size	int	10000	Memory cap; oldest events dropped past this.
timeout	float	5.0	Per-request HTTP timeout, seconds.
enabled	bool	True	When False, capture() is a no-op.
dry_run	bool	False	Buffer + "flush" to an in-memory sink; nothing leaves the process.
debug	bool	False	Attach a default on_error that logs at WARNING.
on_error	Callable[[BaseException], None]	no-op	Observability hook; never required, never re-raised.
consumer_resolver	Callable[[req], str | None]	None	Default consumer hook for all adapters.
transport	Transport	HTTP	Inject a custom transport (tests).
register_atexit	bool	True	Flush remaining events on interpreter exit.

Adapter options (BeaconASGIMiddleware, init_flask, and the Django settings BEACON_CONSUMER_HEADER / BEACON_CONSUMER_RESOLVER): consumer_header and consumer_resolver, as above.

Environment variables (read by BeaconClient.from_env()): BEACON_INGEST_KEY (required), BEACON_INGEST_URL (optional — defaults to the hosted endpoint), BEACON_ENABLED (0/false/no disables).

Event payload

Each captured request becomes one event in the batched POST /v1/ingest body ({"events": [ ... ]}). Field names and types match the ingest contract and the server's validation exactly:

{
  "ts": "2026-06-03T12:00:00.000+00:00",
  "method": "GET",
  "route": "/users/{id}",
  "path": "/users/123",
  "status": 200,
  "duration_ms": 42,
  "consumer": "acme",
  "error": null
}

route is the low-cardinality template (resolved from the framework's routing), which powers per-endpoint aggregation; path is the concrete request path. consumer and error are always sent (as null when absent), matching the JS SDK.

Graceful shutdown

Buffered events are flushed automatically on interpreter exit. To flush explicitly (e.g. in a worker's shutdown hook):

beacon.shutdown()   # stops the timer and flushes remaining events; idempotent

Development & tests

python -m venv .venv && source .venv/bin/activate
pip install -e ".[dev]"
pytest

Core tests run against an in-memory mock transport (no network) and assert the exact payload shape, batching, interval/shutdown flush, and the never-raise/requeue behavior. Adapter smoke tests drive a real request through each framework and skip if the framework isn't installed.

License

MIT © Skyware LLC