Official Python SDK for Clicktag API analytics. Add one middleware and see traffic, latency percentiles, error rates and top consumers per endpoint.
- Works with FastAPI, Starlette, Flask and any ASGI or WSGI app
- Zero dependencies, Python 3.9+
- Requests are aggregated per minute in memory and sent in small batches from one background thread. Recording never blocks or raises in your request path.
pip install clicktagCreate an API key in your Clicktag dashboard and set it as CLICKTAG_API_KEY.
from fastapi import FastAPI
from clicktag.fastapi import ClicktagMiddleware
app = FastAPI()
app.add_middleware(ClicktagMiddleware)Starlette works the same way with from clicktag.starlette import ClicktagMiddleware.
from flask import Flask
from clicktag.flask import ClicktagMiddleware
app = Flask(__name__)
ClicktagMiddleware(app)With an app factory, create clicktag = ClicktagMiddleware() once and call clicktag.init_app(app).
from clicktag.asgi import ClicktagMiddleware # or clicktag.wsgi
app = ClicktagMiddleware(app)This covers Django, Litestar, Quart, Falcon and friends. These frameworks don't expose a route template to middleware, so the raw path is sent and Clicktag groups it into routes (/users/42 becomes /users/:id). Pass a route callback to send your own templates.
Pass options as keyword arguments, e.g. app.add_middleware(ClicktagMiddleware, consumer=..., ignore=...).
| Option | Default | Description |
|---|---|---|
api_key |
CLICKTAG_API_KEY |
API key, or a zero-argument function returning it |
consumer |
none | consumer(request) returns who made the call (customer, tenant, API key id) |
route |
framework route | route(request) returns a route template, None keeps the default |
ignore |
none | ignore(request) returns True to skip a request, e.g. health checks |
endpoint |
https://clicktag.io/api/ingest |
Ingest URL |
flush_interval |
10.0 |
Seconds between flushes |
max_batch_rows |
1000 |
Rows per batch (1 to 5000). A full buffer flushes right away |
error_samples |
True |
Send up to 50 5xx and 20 4xx samples per flush |
debug |
False |
Log diagnostics through the clicktag logger |
Callbacks receive a Starlette Request (FastAPI, Starlette), flask.request (Flask), the ASGI scope or the WSGI environ.
app.add_middleware(
ClicktagMiddleware,
consumer=lambda request: request.headers.get("x-customer-id"),
ignore=lambda request: request.url.path == "/health",
)Per minute, method, route and status: request count, total duration, a latency histogram, User-Agent and consumer. Error samples add the path without its query string, the duration and, for unhandled exceptions, ExceptionType: message.
Never sent: request or response bodies, headers other than User-Agent, query strings, cookies or IP addresses.
- Timing. Duration runs from the request until the response starts, so streamed bodies don't count. Unhandled exceptions are recorded as 500, and ASGI clients that disconnect before a response as 499.
- Workers. Every process (gunicorn or uvicorn workers) keeps and flushes its own buffer. Forked workers start empty.
- Shutdown. Buffers flush on ASGI lifespan shutdown and at interpreter exit, within 5 seconds.
- Serverless. A frozen process can't flush in the background, so call
clicktag.flush()before your handler returns. It sends everything buffered, waits for delivery and returnsFalseif itstimeoutran out. - Delivery. Failed batches are retried up to 3 times with backoff, using the same body and
batch_idso the server drops duplicates. Oversized batches are split. An invalid key logs one warning per process and data is dropped.
from clicktag import Clicktag
client = Clicktag(integration="my-server")
client.record(method="GET", path="/users/42", route="/users/:id", status=200, duration_ms=12.5)pip install -e ".[test]"
pytest
ruff check . && ruff format --check .tests/fixtures/conformance.json is generated by the JS SDK, so both SDKs produce identical batches:
npx tsx scripts/generate_fixtures.mts ../clicktag-js > tests/fixtures/conformance.jsonMIT