reqkey 0.2.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.

Website: reqkey.com · Documentation: reqkey.com/docs

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: await /ingest, then 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.

Denied requests are recorded by default in mode="both", including requests with a missing key. This makes invalid-key attempts, exhausted consumers, and rate-limit hammering visible in analytics. Set ingest_denied_requests=False if you deliberately do not want those events or their ingestion cost.

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.
consumer_name_resolver	—	Resolve a display name from a request header, query parameter, or another trusted source.
client_ip_resolver	automatic	Override client-IP extraction for a trusted proxy setup.
on_error	—	Receive validation or ingestion service failures without exposing request secrets.
ingest_denied_requests	True	Record denied traffic in mode="both".
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
• consumerName when a resolver supplies one

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.

Consumer names

Use consumer_name_resolver when an upstream gateway or your application provides a useful customer label. The resolver may read any part of the framework request and may be synchronous or asynchronous in ASGI/Django:

def consumer_name(request: Request) -> str | None:
    return (
        request.headers.get("X-RapidAPI-User")
        or request.query_params.get("consumerName")
    )


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

The resolved value is sent as analytics metadata. It is not used as the consumer API key and does not change authorization.

For Django, set CONSUMER_NAME_RESOLVER in REQKEY. For WSGI, the resolver receives the WSGI environment, so a header can be read as environ.get("HTTP_X_RAPIDAPI_USER").

Client IPs and trusted proxies

Client-IP capture is opt-in with capture_client_ip=True. By default, the SDK keeps the framework-provided peer address first: request.client.host for ASGI, REMOTE_ADDR for WSGI, and request.META["REMOTE_ADDR"] for Django. It only falls back when that value is absent, checking common proxy headers such as X-Forwarded-For, X-Real-IP, Cloudflare, Vercel, Fly, Fastly, Azure, App Engine, CloudFront, and standardized Forwarded headers.

If a trusted proxy always supplies the real client in a specific header, use a resolver to override that default explicitly:

app.add_middleware(
    ReqKeyMiddleware,
    project_key=os.environ["REQKEY_PROJECT_KEY"],
    api_id="api_payments",
    capture_client_ip=True,
    client_ip_resolver=lambda request: request.headers.get("CF-Connecting-IP"),
)

Only trust forwarding headers when your deployment proxy overwrites them; otherwise a caller can spoof their value.

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

No CORS configuration is needed for server-to-server clients or for the server itself to set these headers. Only cross-origin browser JavaScript that must read them needs the names added to the 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.

Failure alerts

Webhook payload formats differ between Discord, Slack, Teams, and custom alert receivers, so the SDK does not post one supposed universal webhook format. Instead, on_error receives a provider-neutral MiddlewareErrorEvent for ReqKey validation or ingestion failures:

from reqkey import MiddlewareErrorEvent


async def report_reqkey_failure(event: MiddlewareErrorEvent) -> None:
    await send_to_your_alert_provider(
        {
            "operation": event.operation,
            "message": event.message,
            "method": event.method,
            "path": event.path,
            "request_id": event.request_id,
            "status_code": event.status_code,
        }
    )


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

The event intentionally excludes API keys, project credentials, headers, query parameters, and bodies. Exceptions raised by the callback are logged and do not replace the application response. WSGI callbacks are synchronous; ASGI callbacks may be synchronous or asynchronous, and Django follows the sync/async mode of its middleware chain.

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")

Direct ingestion also accepts an optional consumer label:

reqkey.ingest(
    decision.request_id,
    api_id="api_payments",
    method="POST",
    path="/payments",
    status_code=201,
    consumer_name="rapidapi-user",
)

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