Python SDK
The official Python SDK for API key validation, credit metering, consumer rate limits, and correlated traffic analytics. One package contains the shared sync/async client and optional middleware for FastAPI, Starlette, Flask, Django, and any ASGI or WSGI application.
Install
Framework dependencies are optional extras: plain reqkey installs only the universal client and its HTTP dependency — it never pulls in Django, Flask, FastAPI, or Starlette.
# Core client only (scripts, workers, custom frameworks)
pip install reqkey
# With a framework adapter
pip install "reqkey[fastapi]" # FastAPI / Starlette
pip install "reqkey[flask]" # Flask
pip install "reqkey[django]" # Django (sync or async)
pip install "reqkey[asgi]" # any ASGI 3 app
pip install "reqkey[wsgi]" # any WSGI app (Bottle, Pyramid, ...)
# Everything
pip install "reqkey[all]"The SDK reads your project credential from the REQKEY_PROJECT_KEY environment variable. Copy the real key from the dashboard, and keep it out of source control:
export REQKEY_PROJECT_KEY="reqkey_your_project_key"Quickstart
Add the middleware once, where your application is created. Every route is guarded automatically — your endpoints stay untouched, and the application never calls /key/validate or /ingest directly.
import os
from fastapi import FastAPI, Request
from reqkey.fastapi import ReqKeyMiddleware
app = FastAPI()
app.add_middleware(
ReqKeyMiddleware,
project_key=os.environ["REQKEY_PROJECT_KEY"],
api_id="api_payments",
mode="both", # validate keys AND record analytics
key_name="X-MyStartup-Key", # where consumers send their key
exclude_paths=("/health", "/docs"),
)
@app.post("/payments")
async def create_payment(request: Request):
decision = request.state.reqkey
return {"created": True, "credits_remaining": decision.credits_remaining}The FastAPI adapter also covers Starlette; the WSGI adapter covers Bottle, Pyramid, and any PEP 3333 app. Django's middleware detects sync vs async chains and picks the matching client automatically. For scripts, workers, and custom frameworks, use the direct ReqKey / AsyncReqKey clients.
Reading the decision
After successful validation, the full decision — the consumer, credits remaining, and a traceable request id — is attached to the request:
request.state.reqkey and request.state.reqkey_request_idrequest.reqkey and request.reqkey_request_idrequest.environ["reqkey.decision"] and request.environ["reqkey.request_id"]scope["state"]["reqkey"]Responses also carry the decision back to your consumer when available:
X-ReqKey-Request-ID— traceable id for support and log correlation.X-ReqKey-Credits-LimitandX-ReqKey-Credits-Remaining— the consumer's balance.X-ReqKey-Validation-Time-Ms— how long the key check took.
Modes
One switch decides what the middleware does per request:
mode="both" # validate keys AND record analytics (default)
mode="validate" # key checks and credit deduction only — nothing recorded
mode="ingest" # analytics only — no key checks at allIn mode="both", denied requests — missing keys, exhausted credits, rate-limit hits — are recorded too, so abuse is visible in your dashboards. Set ingest_denied_requests=False if you deliberately don't want those events. In ingest-only mode, another middleware's validation can be correlated via request_id_resolver.
Where the key comes from
Match however your consumers already send credentials:
# Custom header (recommended)
key_location="header", key_name="X-MyStartup-Key", key_scheme="raw"
# Authorization: Bearer <key>
key_location="header", key_name="Authorization", key_scheme="bearer"
# ?api_key=<key> — easy to test, but URLs end up in logs
key_location="query", key_name="api_key"
# Cookie — sent automatically by browsers
key_location="cookie", key_name="api_key"
# Or take full control with a sync or async extractor
get_consumer_key=lambda request: request.headers.get("X-Custom-Key")Paths & credit costs
Excluded traffic is never validated, charged, or recorded — for anyone. Credit costs can be a constant or a per-request resolver:
# Exact paths or trailing-* prefixes: never validated, charged, or recorded
exclude_paths=("/health", "/openapi.json", "/docs/*", "/cron/*")
# Or decide per request with a sync or async resolver
should_protect=lambda request: request.url.path.startswith("/api/")
# Charge different endpoints differently
def credits_for(request):
if request.method == "POST" and request.url.path == "/images":
return 5
return 1
app.add_middleware(ReqKeyMiddleware, ..., credits=credits_for)Analytics capture
By default each event records the method, normalized path, status, latency, and user agent — correlated to the validated consumer. Everything else is opt-in:
app.add_middleware(
ReqKeyMiddleware,
project_key=os.environ["REQKEY_PROJECT_KEY"],
api_id="api_payments",
mode="both",
# All off by default — turn on what your dashboards need
capture_query_params=True,
capture_request_headers=True,
capture_response_headers=True,
capture_response_body=True, # textual content, first 1,000 chars
capture_client_ip=True,
# Strip additional headers beyond the built-in redaction list
excluded_headers=("X-Internal-Trace",),
)Authorization, cookies, Set-Cookie, common API-key headers, and your configured consumer-key header are always stripped before anything leaves your server — even with header capture enabled. Response bodies are capped at 1,000 characters of textual content.Failure behavior
Validation fails closed by default: if ReqKey times out or is unreachable, the middleware returns 503 without running your endpoint. Prefer availability? Fail open instead — transport failures let requests through, while explicitly invalid, exhausted, or rate-limited keys are still always denied. Either way, on_error tells you it happened:
from reqkey import MiddlewareErrorEvent
async def report_reqkey_failure(event: MiddlewareErrorEvent) -> None:
# Provider-neutral: post it to Slack, Discord, PagerDuty, ...
await send_to_your_alert_provider({
"operation": event.operation, # "validate" or "ingest"
"message": event.message,
"method": event.method,
"path": event.path,
})
app.add_middleware(
ReqKeyMiddleware,
project_key=os.environ["REQKEY_PROJECT_KEY"],
api_id="api_payments",
failure_mode="closed", # or "open": let requests through
on_error=report_reqkey_failure,
)The SDK never auto-retries /key/validate, because validation deducts credits — safe retries would require server-side idempotency.
Configuration reference
Every option below is a keyword argument to any ReqKeyMiddleware. Django takes the same options in its REQKEY settings mapping, uppercased — key_name becomes "KEY_NAME".
project_keystrrequiredYour project credential, sent to ReqKey as the Bearer token. Read it from the REQKEY_PROJECT_KEY environment variable — never hard-code it.
api_idstrrequiredThe ReqKey API this application serves — validation and analytics are scoped to it.
mode"both" | "validate" | "ingest"What the middleware does per request: guard keys and record analytics (the default), key checks only, or analytics only.
enabledboolOne switch to bypass ReqKey entirely — handy as a local-development toggle that leaves your middleware layout untouched.
root_keystrBackward-compatible alias for project_key. Never pass both.
key_location"header" | "query" | "cookie"Where the key travels. Headers are recommended — query strings end up in logs and browser history.
key_namestrThe customer-facing header, query parameter, or cookie name. Make it yours: x-startup-key.
key_scheme"raw" | "bearer"For headers: the raw key value, or a standard Authorization: Bearer token.
get_consumer_keyresolverFull control — a sync or async function that extracts the key from anywhere in the request.
creditsint | resolverThe cost per successful request. Pass a function to charge heavy endpoints more than light ones.
exclude_pathstuple[str, ...]Paths that stay public — never validated, charged, or recorded. Exact matches or trailing-* prefixes.
should_protectresolverA per-request predicate for when patterns are not enough — protect exactly the routes you choose.
skip_methodstuple[str, ...]HTTP methods that bypass ReqKey. Defaults to ("OPTIONS",) so CORS preflights stay free.
capture_query_paramsboolRecord query strings with each analytics event. Off by default.
capture_request_headersboolRecord request headers — auth headers, cookies, and consumer keys are always stripped first.
capture_response_headersboolRecord response headers, with the same always-on redaction.
capture_response_bodyboolRecord textual response bodies, capped at the first 1,000 characters.
capture_client_ipboolRecord the caller’s IP address, proxy-aware out of the box.
excluded_headerstuple[str, ...]Strip additional headers from captured analytics, beyond the built-in redaction list.
consumer_name_resolverresolverLabel events with a human-readable customer name from a header or another trusted source.
client_ip_resolverresolverBehind a trusted proxy? Tell the SDK exactly which header holds the real client IP.
request_id_resolverresolverIn ingest-only mode: correlate events to a validation performed by another middleware.
ingest_denied_requestsboolRecord denied traffic too — invalid keys, exhausted credits, rate-limit hits — so abuse is visible. On by default.
failure_mode"closed" | "open"Fail closed (block with a 503, the default) or fail open (let requests through unvalidated). Invalid and exhausted keys are always denied either way.
on_errorcallbackCalled on ReqKey service failures with a provider-neutral event — wire it to Slack, Discord, or PagerDuty. No request secrets included.
error_messagesdict[str, str]Rewrite the customer-facing denial messages in your own voice, per error type.
timeoutfloatSeconds allowed for each ReqKey operation before the failure mode kicks in. Defaults to 2.0.
Direct clients
ReqKey and AsyncReqKey expose verify() and ingest() for full manual control, plus ReqKey.from_env() to construct from REQKEY_PROJECT_KEY. The decision object carries valid, reason, credits_remaining, and request_id.