reqkey 0.1.0

The official Python SDK for ReqKey API key validation and usage analytics

ReqKey Python SDK

The official Python SDK for API key validation, credit metering, consumer rate limits, and correlated API traffic analytics with ReqKey.

One distribution contains the shared sync/async client and optional framework adapters. FastAPI, Starlette, Django, Flask, and generic ASGI/WSGI applications can all use ready-made middleware without implementing ReqKey request logic.

Supported integrations

Application	Install	Middleware import
FastAPI / Starlette	reqkey[fastapi]	reqkey.fastapi.ReqKeyMiddleware
Generic ASGI	reqkey[asgi]	reqkey.asgi.ReqKeyMiddleware
Django (sync or async)	reqkey[django]	reqkey.django.ReqKeyMiddleware
Flask	reqkey[flask]	reqkey.flask.ReqKeyMiddleware
Bottle / Pyramid / generic WSGI	reqkey or reqkey[wsgi]	reqkey.wsgi.ReqKeyMiddleware
Scripts, workers, and custom frameworks	reqkey	reqkey.ReqKey / reqkey.AsyncReqKey

Framework dependencies are optional. Installing plain reqkey installs only the universal client and its HTTP dependency; it does not install Django, Flask, FastAPI, or Starlette.

This release provides ready-made middleware for incoming requests in the frameworks listed above. It does not yet include native Tornado or AWS Lambda adapters, nor automatic instrumentation of outgoing calls made with requests or httpx. Those applications can still use the direct ReqKey and AsyncReqKey clients, but framework-specific request extraction remains application code until a dedicated adapter is added.

Install

Core client:

pip install reqkey

FastAPI middleware:

pip install "reqkey[fastapi]"

Django middleware:

pip install "reqkey[django]"

Flask middleware:

pip install "reqkey[flask]"

Every supported framework integration:

pip install "reqkey[all]"

Complete FastAPI integration

import os

from fastapi import FastAPI, Request
from reqkey.fastapi import ReqKeyMiddleware

app = FastAPI()

app.add_middleware(
    ReqKeyMiddleware,

    # ReqKey project
    project_key=os.environ["REQKEY_PROJECT_KEY"],
    api_id="api_payments",

    # "validate", "ingest", or "both"
    mode="both",
    enabled=True,

    # Where your consumer sends their ReqKey-issued key
    key_location="header",
    key_name="X-StartupName-Key",
    key_scheme="raw",

    # Usage cost
    credits=1,

    # No validation or analytics on these routes
    exclude_paths=("/health", "/docs", "/openapi.json", "/redoc", "/cron/*"),

    # Privacy-safe analytics defaults
    capture_query_params=False,
    capture_request_headers=False,
    capture_response_headers=False,
    capture_response_body=False,
)


@app.get("/health")
async def health() -> dict[str, bool]:
    return {"ok": True}


@app.post("/payments")
async def create_payment(request: Request) -> dict[str, object]:
    decision = request.state.reqkey
    return {
        "created": True,
        "credits_remaining": decision.credits_remaining,
    }

The application contains no direct /key/validate or /ingest requests. The middleware owns that internal work.

Complete Django integration

Set the project credential in the server environment:

export REQKEY_PROJECT_KEY="your_project_key"

Then configure settings.py:

REQKEY = {
    "API_ID": "api_payments",
    "MODE": "both",
    "KEY_LOCATION": "header",
    "KEY_NAME": "X-StartupName-Key",
    "KEY_SCHEME": "raw",
    "CREDITS": 1,
    "EXCLUDE_PATHS": ("/health", "/admin/*", "/static/*"),
    "CAPTURE_QUERY_PARAMS": False,
    "CAPTURE_REQUEST_HEADERS": False,
    "CAPTURE_RESPONSE_HEADERS": False,
    "CAPTURE_RESPONSE_BODY": False,
}

MIDDLEWARE = [
    "django.middleware.security.SecurityMiddleware",
    "reqkey.django.ReqKeyMiddleware",
    # The rest of your middleware...
]

The same middleware class detects whether Django supplied a synchronous or asynchronous request chain and uses ReqKey or AsyncReqKey accordingly. After successful validation, a Django view can read:

def create_payment(request):
    decision = request.reqkey
    return JsonResponse(
        {
            "created": True,
            "credits_remaining": decision.credits_remaining,
        }
    )

All direct middleware keyword options are available in Django's REQKEY mapping using uppercase names. PROJECT_KEY may be specified there, but an environment variable is recommended so credentials do not enter source control.

Complete Flask integration

import os

from flask import Flask, request
from reqkey.flask import ReqKeyMiddleware

app = Flask(__name__)
app.wsgi_app = ReqKeyMiddleware(
    app.wsgi_app,
    project_key=os.environ["REQKEY_PROJECT_KEY"],
    api_id="api_payments",
    mode="both",
    key_name="X-StartupName-Key",
    exclude_paths=("/health", "/static/*"),
)


@app.post("/payments")
def create_payment():
    decision = request.environ["reqkey.decision"]
    return {
        "created": True,
        "credits_remaining": decision.credits_remaining,
    }

Flask remains the application object; only its wsgi_app callable is wrapped. The same reqkey.wsgi.ReqKeyMiddleware class works with Bottle, Pyramid, and other PEP 3333 WSGI applications.

Generic ASGI and WSGI

For any ASGI 3 application:

import os

from reqkey.asgi import ReqKeyMiddleware

application = ReqKeyMiddleware(
    application,
    project_key=os.environ["REQKEY_PROJECT_KEY"],
    api_id="api_payments",
)

For any WSGI application:

import os

from reqkey.wsgi import ReqKeyMiddleware

application = ReqKeyMiddleware(
    application,
    project_key=os.environ["REQKEY_PROJECT_KEY"],
    api_id="api_payments",
)

Request lifecycle

With mode="both":

Consumer request
  → extract consumer key
  → await /key/validate
  → denied: return 401 / 402 / 403 / 429
  → approved: run the FastAPI endpoint
  → collect response metadata
  → await /ingest with the validation requestId
  → release the response to the consumer

The framework adapters keep ingestion inside the request lifecycle instead of starting an unreliable background task:

• ASGI awaits ingestion before sending the final response-completion message.
• WSGI ingests when the response iterable finishes, without buffering the full response.
• Django's native middleware ingests before returning the response object.

Streaming responses are forwarded chunk by chunk. The middleware keeps only enough data to reconstruct the first 1,000 response-body characters and holds the final ASGI completion message until ingestion finishes. It does not retain the complete stream in memory. An indefinitely open stream cannot produce a completed analytics event until that stream eventually closes.

Django streaming responses are never consumed or buffered by the native middleware. Their status and headers can be recorded, but response-body capture is omitted because Django requires middleware to preserve the streaming iterator.

Configuration

Input	Default	Purpose
project_key	required	Project credential sent to ReqKey as the Bearer token.
root_key	—	Backward-compatible alias for project_key; never pass both.
api_id	required	ReqKey API being protected or observed.
mode	"both"	"validate", "ingest", or "both".
enabled	True	Bypass the integration entirely when false.
key_location	"header"	"header", "query", or "cookie".
key_name	"X-API-Key"	Customer-facing header, query parameter, or cookie name.
key_scheme	"raw"	"raw" or "bearer" for headers.
credits	1	Static cost or sync/async cost resolver.
exclude_paths	empty	Exact paths or trailing-* prefix patterns.
skip_methods	("OPTIONS",)	Methods that bypass ReqKey.
failure_mode	"closed"	Deny or allow when ReqKey itself is unavailable.
error_messages	built in	Override customer-facing denial messages.
timeout	2.0	Timeout for each ReqKey operation.

REQKEY_PROJECT_KEY is the preferred environment-variable name. REQKEY_ROOT_KEY remains supported by ReqKey.from_env() and AsyncReqKey.from_env().

The project credential must remain server-side. Do not expose it in browser code or commit it to source control.

For a local-development switch, pass a boolean without changing the middleware layout:

enabled=os.getenv("ENABLE_REQKEY", "true").lower() == "true"

Choose validation, analytics, or both

Validation only:

app.add_middleware(
    ReqKeyMiddleware,
    project_key=os.environ["REQKEY_PROJECT_KEY"],
    api_id="api_payments",
    mode="validate",
)

Traffic analytics only:

app.add_middleware(
    ReqKeyMiddleware,
    project_key=os.environ["REQKEY_PROJECT_KEY"],
    api_id="api_payments",
    mode="ingest",
)

In ingest-only mode, the endpoint runs without ReqKey authentication and the event is associated with api_id. If another middleware already performed validation, provide its request ID for consumer correlation:

app.add_middleware(
    ReqKeyMiddleware,
    project_key=os.environ["REQKEY_PROJECT_KEY"],
    api_id="api_payments",
    mode="ingest",
    request_id_resolver=lambda request: getattr(
        request.state,
        "reqkey_request_id",
        None,
    ),
)

Validation plus correlated analytics:

app.add_middleware(
    ReqKeyMiddleware,
    project_key=os.environ["REQKEY_PROJECT_KEY"],
    api_id="api_payments",
    mode="both",
)

Choose where the consumer key comes from

Custom header, recommended:

key_location="header"
key_name="X-StartupName-Key"
key_scheme="raw"

Authorization Bearer token:

key_location="header"
key_name="Authorization"
key_scheme="bearer"

Query parameter:

key_location="query"
key_name="api_key"

Query parameters are supported for compatibility, but headers are recommended because URLs are often retained in browser history, access logs, and proxy logs.

Cookie:

key_location="cookie"
key_name="startup_key"

A custom sync or async extractor can handle another source:

async def get_consumer_key(request: Request) -> str | None:
    return request.headers.get("X-Custom-Key")


app.add_middleware(
    ReqKeyMiddleware,
    project_key=os.environ["REQKEY_PROJECT_KEY"],
    api_id="api_payments",
    get_consumer_key=get_consumer_key,
)

Avoid reading an API key from the request body. Authentication middleware runs before the endpoint and consuming the body can interfere with downstream body parsing.

Exclude or select endpoints

Exact and prefix exclusions:

exclude_paths=(
    "/health",
    "/openapi.json",
    "/docs/*",
    "/cron/*",
)

For complete control, use a sync or async resolver:

def should_protect(request: Request) -> bool:
    return request.url.path.startswith("/api/")


app.add_middleware(
    ReqKeyMiddleware,
    project_key=os.environ["REQKEY_PROJECT_KEY"],
    api_id="api_payments",
    should_protect=should_protect,
)

The same selection controls validation and ingestion. Excluded traffic is not validated, charged, or recorded.

Dynamic credit costs

def credits_for(request: Request) -> int:
    if request.method == "POST" and request.url.path == "/images":
        return 5
    if request.url.path.startswith("/reports/"):
        return 2
    return 1


app.add_middleware(
    ReqKeyMiddleware,
    project_key=os.environ["REQKEY_PROJECT_KEY"],
    api_id="api_payments",
    credits=credits_for,
)

Analytics capture

Metadata sent by default:

• requestId when validation produced one
• apiId
• method and normalized endpoint path
• response status
• endpoint processing latency
• user agent

Additional data is opt-in:

app.add_middleware(
    ReqKeyMiddleware,
    project_key=os.environ["REQKEY_PROJECT_KEY"],
    api_id="api_payments",
    mode="both",
    capture_query_params=True,
    capture_request_headers=True,
    capture_response_headers=True,
    capture_response_body=True,
    capture_client_ip=True,
    excluded_headers=(
        "X-RapidAPI-Proxy-Secret",
        "X-Vercel-OIDC-Token",
    ),
)

Authorization, cookies, Set-Cookie, common API-key headers, and the configured consumer-key header are always removed from captured headers. Response-body capture is limited to textual content and a hard maximum of 1,000 characters. The limit is enforced again by the HTTP client before /ingest is called.

Custom denial messages

app.add_middleware(
    ReqKeyMiddleware,
    project_key=os.environ["REQKEY_PROJECT_KEY"],
    api_id="api_payments",
    error_messages={
        "missing_api_key": "Send your key in X-StartupName-Key.",
        "invalid_api_key": "That key is invalid or disabled.",
        "insufficient_credits": "Your account has no credits remaining.",
        "access_denied": "This key cannot access the requested API.",
        "rate_limited": "Too many requests. Please retry shortly.",
        "reqkey_unavailable": "Authentication is temporarily unavailable.",
    },
)

The middleware returns stable error identifiers alongside the messages and preserves Retry-After for rate-limited decisions.

Request state and response headers

After successful validation, FastAPI and Starlette routes can access:

request.state.reqkey
request.state.reqkey_request_id

Django views receive the same values as request attributes:

request.reqkey
request.reqkey_request_id

WSGI applications, including Flask, receive namespaced environment values:

request.environ["reqkey.decision"]
request.environ["reqkey.request_id"]

The response includes these values when available:

• X-ReqKey-Request-ID
• X-ReqKey-Credits-Limit
• X-ReqKey-Credits-Remaining
• X-ReqKey-Validation-Time-Ms

If browser JavaScript needs these headers, add them to the FastAPI CORS middleware's expose_headers list.

Availability behavior

Validation fails closed by default. When ReqKey times out or cannot be reached, the middleware returns 503 without running the endpoint.

failure_mode="open"

Fail-open applies only to ReqKey transport/service errors. Explicitly invalid, disabled, exhausted, forbidden, and rate-limited keys are always denied. In mode="both", fail-open requests can still produce an API-level traffic event, but it will not be correlated to a validated consumer.

The SDK does not automatically retry /key/validate, because validation can deduct credits. Safe automatic retries require server-side idempotency.

Direct sync and async clients

Async:

from reqkey import AsyncReqKey

async with AsyncReqKey(project_key=os.environ["REQKEY_PROJECT_KEY"]) as reqkey:
    decision = await reqkey.verify(
        "consumer_key_...",
        api_id="api_payments",
        credits=1,
        resource="/payments",
    )

Synchronous:

from reqkey import ReqKey

with ReqKey(project_key=os.environ["REQKEY_PROJECT_KEY"]) as reqkey:
    decision = reqkey.verify("consumer_key_...", api_id="api_payments")

ASGI and FastAPI use the asynchronous client and await both network operations. WSGI uses the synchronous client. Django automatically selects the correct client based on its middleware chain, avoiding blocking HTTP calls inside an asynchronous Django deployment.

Development

python -m venv .venv
. .venv/bin/activate
pip install -e ".[dev]"
ruff check .
mypy src/reqkey
pytest
python -m build