import asyncio
import contextlib
import functools
import importlib
import inspect
import json
import logging
import os
from collections.abc import Callable, Mapping
from pathlib import Path
from typing import Any, NamedTuple
__all__ = ["API"]
import uvicorn
from starlette.concurrency import run_in_threadpool
from starlette.datastructures import State
from starlette.exceptions import HTTPException
from starlette.middleware.cors import CORSMiddleware
from starlette.middleware.errors import ServerErrorMiddleware
from starlette.middleware.exceptions import ExceptionMiddleware
from starlette.middleware.gzip import GZipMiddleware
from starlette.middleware.httpsredirect import HTTPSRedirectMiddleware
from starlette.middleware.sessions import SessionMiddleware
from starlette.middleware.trustedhost import TrustedHostMiddleware
from starlette.responses import JSONResponse, PlainTextResponse
from starlette.responses import Response as StarletteResponse
from starlette.types import ASGIApp
from . import status_codes
from .background import BackgroundQueue
from .contracts import (
inferred_response_model,
inferred_stream_model,
response_annotation_is_ignored,
response_status_allows_body,
response_type_adapter,
)
from .errors import (
INTERNAL_SERVER_ERROR,
PROBLEM_JSON,
Problem,
legacy_error_payload,
problem_payload_for_async,
status_title,
)
from .formats import get_formats
from .models import Request, Response
from .params import _Depends
from .routes import Router, _is_pydantic_model, _view_return_hint
from .routing import _AUTH_UNSET as _ROUTER_AUTH_UNSET
from .routing import Router as _IncludableRouter
from .routing import _normalize_prefix, _prefix_scoped_hook
from .staticfiles import StaticFiles
from .statics import (
DEFAULT_CORS_PARAMS,
DEFAULT_MAX_REQUEST_SIZE,
DEFAULT_OPENAPI_THEME,
)
from .templates import Templates
logger = logging.getLogger("responder")
_UNSET = object()
class _MW(NamedTuple):
"""A collected middleware spec, assembled lazily by build_middleware_stack."""
cls: type
options: dict
def _api_json_default_hook(api):
"""The API's ``json.dumps`` ``default=`` hook (user encoder included).
Mirrors ``Response._json_default_hook`` so problem payloads built outside
a ``Response`` (e.g. from a raised :class:`~responder.Problem`) serialize
the same types the media path does.
"""
formats = getattr(api, "formats", None)
json_format = formats.get("json") if formats else None
hook = getattr(json_format, "_responder_default_hook", None)
if hook is not None:
return hook
from .formats import _json_default
return _json_default
async def _negotiated_http_error(request, exc):
"""Render HTTPExceptions (404s and friends) as negotiated framework errors."""
headers = getattr(exc, "headers", None)
if exc.status_code in (204, 304):
return StarletteResponse(status_code=exc.status_code, headers=headers)
api = request.scope.get("api")
if getattr(api, "problem_details", True):
is_problem = isinstance(exc, Problem)
payload = await problem_payload_for_async(
request.scope,
exc.status_code,
exc.detail,
title=(exc.title if is_problem else None) or status_title(exc.status_code),
errors=exc.errors if is_problem else None,
request=request,
exc=exc,
)
if is_problem:
# Explicit Problem members layer on top of problem_handler
# enrichment, mirroring ``resp.problem()``. ``instance`` defaults
# to the request path per the RFC 9457 recommendation.
if exc.type is not None:
payload["type"] = exc.type
if exc.instance is not None:
payload["instance"] = exc.instance
else:
payload.setdefault("instance", request.url.path)
if exc.extensions:
payload.update(exc.extensions)
content = json.dumps(payload, default=_api_json_default_hook(api)).encode(
"utf-8"
)
return StarletteResponse(
content=content,
status_code=exc.status_code,
headers=headers,
media_type=PROBLEM_JSON,
)
return JSONResponse(
payload,
status_code=exc.status_code,
headers=headers,
media_type=PROBLEM_JSON,
)
if "json" in request.headers.get("accept", ""):
return JSONResponse(
legacy_error_payload(exc.status_code, exc.detail),
status_code=exc.status_code,
headers=headers,
)
return PlainTextResponse(exc.detail, status_code=exc.status_code, headers=headers)
async def _negotiated_server_error(request, exc):
"""Render unhandled 500s with the same default error contract."""
if getattr(request.scope.get("api"), "problem_details", True):
return JSONResponse(
await problem_payload_for_async(
request.scope,
500,
INTERNAL_SERVER_ERROR,
title=INTERNAL_SERVER_ERROR,
request=request,
exc=exc,
),
status_code=500,
media_type=PROBLEM_JSON,
)
if "json" in request.headers.get("accept", ""):
return JSONResponse(
legacy_error_payload(500, INTERNAL_SERVER_ERROR), status_code=500
)
return PlainTextResponse(INTERNAL_SERVER_ERROR, status_code=500)
def _read_text_if_exists(path: Path) -> str | None:
"""Return the file's text, or ``None`` if it doesn't exist (runs in a thread)."""
try:
return path.read_text()
except FileNotFoundError:
return None
def _const_provider(value):
"""Wrap a bare value as a zero-parameter dependency provider."""
def provider():
return value
return provider
def _as_tuple(value):
if value is None or value is _UNSET:
return ()
if isinstance(value, (list, tuple)):
return tuple(value)
return (value,)
def _auth_security_requirement(auth: Any) -> str | None:
if hasattr(auth, "security_requirement"):
return auth.security_requirement()
if hasattr(auth, "scheme_name"):
return auth.scheme_name
return None
def _auth_has_security_scheme(auth: Any) -> bool:
if not hasattr(auth, "security_scheme"):
return False
return auth.security_scheme() is not None
def _deep_merge_dicts(base: Mapping[str, Any], override: Mapping[str, Any]) -> dict:
result: dict[str, Any] = dict(base)
for key, value in override.items():
if (
key in result
and isinstance(result[key], Mapping)
and isinstance(value, Mapping)
):
result[key] = _deep_merge_dicts(result[key], value)
else:
result[key] = value
return result
def _openapi_response(value: Any) -> dict[str, Any]:
if isinstance(value, str):
return {"description": value}
if value is None:
return {"description": "Response"}
if not isinstance(value, Mapping):
raise TypeError("OpenAPI response values must be strings or mappings")
return dict(value)
def _typed_response_status(status: Any) -> int:
if isinstance(status, bool):
raise TypeError("Typed response status codes must be integers")
try:
value = int(status)
except (TypeError, ValueError) as exc:
raise TypeError(
"Typed response models require an exact HTTP status code"
) from exc
if str(status) != str(value) or not 100 <= value <= 599:
raise ValueError(f"Invalid typed response status code {status!r}")
return value
def _validate_declared_response_model(model: Any, *, label: str) -> None:
if model is None or model is False:
raise TypeError(f"{label} must be a Pydantic TypeAdapter-compatible type")
try:
response_type_adapter(model).json_schema()
except Exception as exc:
raise TypeError(
f"{label} must be a Pydantic TypeAdapter-compatible type"
) from exc
def _normalize_openapi_responses(
responses: Mapping[Any, Any],
) -> tuple[dict[str, Any], dict[int, Any]]:
if not isinstance(responses, Mapping):
raise TypeError("responses= must be a mapping of status codes to responses")
metadata: dict[str, Any] = {}
models: dict[int, Any] = {}
for status, value in responses.items():
model = _UNSET
if isinstance(value, Mapping):
response = dict(value)
model = response.pop("model", _UNSET)
elif isinstance(value, str) or value is None:
response = _openapi_response(value)
else:
model = value
typed_status = _typed_response_status(status)
response = {"description": status_title(typed_status)}
if model is not _UNSET:
typed_status = _typed_response_status(status)
if not response_status_allows_body(typed_status):
raise ValueError(
f"HTTP {typed_status} cannot declare a response model because "
"that status does not allow a response body"
)
_validate_declared_response_model(
model, label=f"responses={{{typed_status}: ...}} model"
)
models[typed_status] = model
response.setdefault("description", status_title(typed_status))
metadata[str(status)] = response
return metadata, models
def _response_contract_views(endpoint: Any) -> list[Callable[..., Any]]:
if not inspect.isclass(endpoint):
return [endpoint]
names = ["on_request"]
names.extend(
f"on_{method}"
for method in (
"get",
"head",
"post",
"put",
"patch",
"delete",
"options",
"trace",
"connect",
)
)
return [view for name in names if callable(view := getattr(endpoint, name, None))]
def _diagnose_response_annotation(endpoint: Any, route: str | None) -> None:
for view in _response_contract_views(endpoint):
try:
raw = inspect.signature(view).return_annotation
except (TypeError, ValueError):
continue
if raw is inspect.Signature.empty:
continue
resolved = _view_return_hint(view)
if resolved is None and raw not in (None, type(None), "None", "NoneType"):
logger.warning(
"Return annotation %r on %s for route %r could not be resolved; "
"use a Pydantic TypeAdapter-compatible type or pass "
"response_model=False",
raw,
getattr(view, "__qualname__", getattr(view, "__name__", view)),
route,
)
continue
if response_annotation_is_ignored(resolved):
continue
if resolved is not None and inferred_response_model(resolved) is not None:
continue
logger.warning(
"Return annotation %r on %s for route %r cannot be used as a "
"response contract; use a Pydantic TypeAdapter-compatible type "
"or pass response_model=False",
raw,
getattr(view, "__qualname__", getattr(view, "__name__", view)),
route,
)
_OPENAPI_EXAMPLE_FIELDS = frozenset({"summary", "description", "value", "externalValue"})
def _openapi_examples(value: Any) -> dict[str, Any]:
if isinstance(value, Mapping) and value:
maybe_examples = {
str(name): dict(example)
for name, example in value.items()
if isinstance(example, Mapping)
and _OPENAPI_EXAMPLE_FIELDS.intersection(example)
}
if len(maybe_examples) == len(value):
return maybe_examples
return {"default": {"value": value}}
def _add_openapi_examples(
responses: dict[str, Any],
status_code: Any,
examples: Any,
*,
media_type: str = "application/json",
) -> None:
response = responses.setdefault(str(status_code), {})
content = response.setdefault("content", {})
media = content.setdefault(media_type, {})
media["examples"] = _openapi_examples(examples)
def _normalize_openapi_response_examples(
response_examples: Mapping[Any, Any],
) -> dict[str, Any]:
if not isinstance(response_examples, Mapping):
raise TypeError(
"response_examples= must be a mapping of status codes to examples"
)
responses: dict[str, Any] = {}
for status_code, examples in response_examples.items():
_add_openapi_examples(responses, status_code, examples)
return responses
def _registers_as_named_component(model):
"""Whether ``model`` should be auto-registered as a named OpenAPI component.
Parametrized generics (``Page[Item]``) are excluded — their bracketed names
aren't valid component keys, so the spec builder inlines them instead.
"""
meta = getattr(model, "__pydantic_generic_metadata__", None)
return _is_pydantic_model(model) and not (meta and meta.get("args"))
[docs]
def abort(
status_code,
*,
detail=None,
headers=None,
title=None,
type=None, # noqa: A002
instance=None,
errors=None,
**extensions,
):
"""Short-circuit the request with an HTTP error response.
Raises an ``HTTPException`` that Responder renders (as JSON or text per the
client's ``Accept`` header). Use it to bail out from anywhere in a handler
or dependency without importing Starlette directly.
Usage::
from responder import abort
@api.route("/admin")
def admin(req, resp):
if not req.session.get("is_admin"):
abort(403, detail="Forbidden")
Passing any of the RFC 9457 members — ``title``, ``type``, ``instance``,
``errors``, or extension keywords — raises :class:`~responder.Problem`
instead, carrying them into the rendered problem-details payload::
abort(
409,
detail="Plan quota exhausted",
type="https://api.example.com/errors/quota-exceeded",
balance=0,
)
:param status_code: The HTTP status code (e.g. ``404``).
:param detail: Optional error message; defaults to the status phrase.
:param headers: Optional dict of headers to attach to the error response.
:param title: Optional problem summary; defaults to the status phrase.
:param type: Optional URI identifying the problem type.
:param instance: Optional URI for this occurrence; defaults to the
request path in the rendered payload.
:param errors: Optional list of structured error dicts.
:param extensions: Extra keyword arguments become top-level extension
members of the problem payload.
"""
if (
title is None
and type is None
and instance is None
and errors is None
and not extensions
):
raise HTTPException(status_code=status_code, detail=detail, headers=headers)
raise Problem(
status_code,
detail,
title=title,
type=type,
instance=instance,
errors=errors,
headers=headers,
**extensions,
)
[docs]
class API:
"""The primary web-service class.
:param static_dir: The directory to use for static files (default ``static``). Mounted at ``static_route`` only if it exists — it is never created for you. Passing a ``static_dir`` explicitly that doesn't exist raises ``FileNotFoundError``.
:param templates_dir: The directory to use for templates (default ``templates``). It is not created for you.
:param auto_escape: If ``True``, HTML and XML templates will automatically be escaped.
:param enable_hsts: If ``True``, redirect HTTP requests to HTTPS and send a ``Strict-Transport-Security`` header.
:param security_headers: If ``True``, add common security headers (nosniff, X-Frame-Options, Referrer-Policy) to every response; pass a dict of ``SecurityHeadersMiddleware`` options to customize (e.g. ``content_security_policy``).
:param gzip: If ``True`` (the default), compress responses with GZip.
:param openapi_theme: OpenAPI documentation theme, must be one of ``elements``, ``rapidoc``, ``redoc``, ``swagger_ui``
""" # noqa: E501
status_codes = status_codes
def __init__(
self,
*,
debug=False,
title=None,
version=None,
description=None,
terms_of_service=None,
contact=None,
license=None, # noqa: A002
openapi=None,
openapi_servers=None,
openapi_route="/schema.yml",
static_dir=_UNSET,
static_route="/static",
implicit_static_fallback=False,
templates_dir="templates",
auto_escape=True,
secret_key=None,
enable_hsts=False,
security_headers=False,
docs_route=None,
cors=False,
cors_params=DEFAULT_CORS_PARAMS,
allowed_hosts=None,
openapi_theme=DEFAULT_OPENAPI_THEME,
lifespan=None,
gzip=True,
request_id=False,
enable_logging=False,
trust_proxy_headers=False,
csrf=False,
redirect_slashes=True,
max_request_size=DEFAULT_MAX_REQUEST_SIZE,
auto_etag=False,
auto_vary=True,
request_timeout=None,
ws_idle_timeout=None,
trace_dispatch=False,
sessions="auto",
session_backend=None,
session_cookie=None,
session_https_only=None,
session_same_site="lax",
session_max_age=14 * 24 * 3600,
metrics_route=None,
metrics_buckets=None,
health_route=None,
encoder=None,
json_ensure_ascii=False,
json_decimal="string",
problem_details=True,
problem_handler=None,
auth=None,
):
"""Create a new Responder API instance.
:param debug: If ``True``, enable debug mode with verbose error pages.
:param title: The title of the API, used in OpenAPI documentation.
:param version: The version string for the API (e.g. ``"1.0"``).
:param description: A longer description of the API for OpenAPI docs.
:param terms_of_service: URL to the API's terms of service.
:param contact: Contact information dict for the API (``name``, ``url``, ``email``).
:param license: License information dict (``name``, ``url``).
:param openapi: The OpenAPI version string (e.g. ``"3.0.2"``). Enables OpenAPI schema generation.
:param openapi_route: The URL path for the OpenAPI schema (default ``"/schema.yml"``).
:param static_dir: Directory for static files (default ``"static"``). Mounted at ``static_route`` only if the directory exists — it is never created implicitly. A ``static_dir`` passed explicitly that doesn't exist raises ``FileNotFoundError``. Set to ``None`` to disable.
:param static_route: URL prefix for serving static files (default ``"/static"``).
:param implicit_static_fallback: If ``True``, ``add_route(route)`` without an endpoint keeps the legacy static-fallback behavior. The default is ``False``: pass an endpoint explicitly, or serve static assets via ``static_dir``/``static_route``.
:param templates_dir: Directory for Jinja2 templates (default ``"templates"``).
:param auto_escape: If ``True``, auto-escape HTML/XML in templates.
:param secret_key: Secret key for signing cookie-based sessions. **Always set this in production.**
:param enable_hsts: If ``True``, redirect HTTP requests to HTTPS and send a ``Strict-Transport-Security`` header.
:param security_headers: If ``True``, add common security headers to every response; pass a dict of ``SecurityHeadersMiddleware`` options to customize.
:param docs_route: URL path for interactive API docs (e.g. ``"/docs"``). Enables OpenAPI if not already set.
:param cors: If ``True``, enable CORS middleware.
:param cors_params: Dict of CORS configuration (``allow_origins``, ``allow_methods``, etc.).
:param allowed_hosts: List of allowed hostnames (e.g. ``["example.com"]``). Defaults to ``["*"]``.
:param openapi_theme: Documentation UI theme: ``"swagger_ui"``, ``"redoc"``, ``"rapidoc"``, or ``"elements"``.
:param lifespan: An async context manager for startup/shutdown logic.
:param gzip: If ``True`` (the default), compress responses with GZip.
:param request_id: If ``True``, add ``X-Request-ID`` headers to all responses.
:param enable_logging: If ``True``, enable structured logging with per-request context (request ID, method, path, client IP).
:param trust_proxy_headers: If ``True``, honor the forwarding headers set by a reverse proxy — RFC 7239 ``Forwarded``, with fallback to ``X-Forwarded-Proto``/``X-Forwarded-Host``/``X-Forwarded-For``/``X-Real-IP`` — rewriting the request's scheme, host, and client address to what the original client sent (since 9.0; previously this flag only affected the IP recorded by ``enable_logging``). Redirects, URL building, HTTPS detection, and logged IPs are then correct behind nginx/Caddy/a load balancer. Only enable this when every request reaches Responder through a proxy you control that overwrites those headers — otherwise any client can spoof them.
:param csrf: If ``True``, protect unsafe requests (``POST``/``PUT``/``PATCH``/``DELETE``) with a session-bound CSRF token: requests must send the value of ``req.csrf_token`` in an ``X-CSRF-Token`` header or a ``csrf_token`` form field (``{{ req.csrf_input }}`` in templates), or they get a ``403``. Off by default; requires sessions. Override per route with ``@api.route(..., csrf=False)`` (e.g. for webhook endpoints) or ``csrf=True`` to protect a single route without the app-wide switch.
:param redirect_slashes: If ``True`` (the default), requests that miss only by a trailing slash are redirected (``307``) to the matching route.
:param max_request_size: Maximum request body size in bytes, enforced chunk-by-chunk as the body arrives. Bodies larger than this get a ``413`` response. Defaults to 100 MiB (since 9.0); pass ``None`` for the legacy unlimited behavior, or a larger value for big uploads (multipart uploads stream to disk, so a large cap does not mean large memory use).
:param auto_etag: If ``True``, GET responses automatically get a content-hash ``ETag`` and matching ``If-None-Match`` requests receive ``304 Not Modified``.
:param auto_vary: If ``True`` (the default since 6.0), content-negotiated responses get a ``Vary: Accept`` header (correct for shared caches). Pass ``False`` to opt out.
:param request_timeout: Seconds a handler may run before the request is answered with ``504 Gateway Timeout``. ``None`` (the default) means unlimited.
:param ws_idle_timeout: Seconds a WebSocket may wait for the next inbound message before the server closes it (code ``1001``). The deadline resets on every message received, so it bounds *idle* time, not total connection lifetime. ``None`` (the default) means unlimited.
:param trace_dispatch: If ``True``, emit debug logs for the documented route-dispatch order (before hooks, auth, dependencies, handler, after hooks).
:param secret_key: Signing key for cookie sessions. Defaults to ``None``: with ``sessions="auto"`` a random per-process key is generated (with a warning); the old public ``"NOTASECRET"`` default is rejected. Set this (or the ``RESPONDER_SECRET_KEY`` env var) for stable, multi-worker sessions.
:param sessions: ``"auto"`` (default) enables cookie sessions, auto-generating an ephemeral key if none is set; ``True`` requires a real ``secret_key`` (raises otherwise); ``False`` disables sessions entirely (``req.session`` then raises).
:param session_backend: Store session data server-side (e.g. ``MemorySessionBackend()``, ``RedisSessionBackend()`` from ``responder.ext.sessions``) with only an opaque ID in the cookie. ``None`` (the default) keeps signed cookie-payload sessions.
:param session_cookie: Name of the session cookie. ``None`` (the default) keeps the underlying middleware's default name.
:param session_https_only: Mark the session cookie ``Secure`` (only sent over HTTPS). ``None`` (the default) means Secure in production and off under ``debug``.
:param session_same_site: ``SameSite`` policy for the session cookie: ``"lax"`` (default), ``"strict"``, or ``"none"`` (requires a Secure cookie).
:param session_max_age: Session lifetime in seconds (default 14 days).
:param metrics_route: URL path (e.g. ``"/metrics"``) serving request counts and latency histograms in Prometheus text format.
:param metrics_buckets: Ascending histogram bucket upper bounds (in seconds) for the metrics endpoint's latency histogram. ``None`` (the default) uses ``responder.ext.metrics.BUCKETS`` (5ms-10s).
:param health_route: URL path (e.g. ``"/health"``) serving an aggregated readiness check (``200``/``503``); see :meth:`add_health_check`.
:param encoder: Optional ``obj -> serializable`` callable applied across **all** response formats (JSON, YAML, MessagePack) to serialize otherwise-unsupported types. Tried first, then falls back to the built-in conversions for ``datetime``, ``UUID``, ``Decimal``, ``set``, dataclasses, and Pydantic models.
:param json_ensure_ascii: If ``True``, escape non-ASCII in JSON as ``\\uXXXX``; ``False`` (the default since 6.0) emits raw UTF-8.
:param json_decimal: ``"string"`` (default) serializes ``Decimal`` values as precision-preserving strings; ``"float"`` preserves the legacy lossy conversion.
:param problem_details: If ``True`` (the default), framework-generated errors use RFC 9457-style ``application/problem+json`` responses. Pass ``False`` to keep the legacy JSON/plain-text negotiation.
:param problem_handler: Optional callable (sync or ``async def``) that can enrich or replace each problem-details payload. It receives ``(payload, request, exc)``; returning ``None`` means the payload was mutated in place. Async handlers are awaited on the negotiated error path and run to completion on a private event loop when invoked from synchronous call sites such as ``resp.problem()`` or route-level validation/timeout errors.
:param auth: Optional app-level auth helper or list of helpers. Routes inherit it by default; pass ``auth=None`` on a route to make that route public.
""" # noqa: E501
self.background = BackgroundQueue()
self.auth_policies = {}
self._auth = _as_tuple(auth)
# Resolved below if cookie sessions are enabled (else stays None).
self.secret_key = None
#: Application-level state. Set values at startup, read them anywhere
#: (handlers can reach it via ``req.api.state``).
self.state = State()
self.formats = get_formats(
encoder=encoder,
json_ensure_ascii=json_ensure_ascii,
json_decimal=json_decimal,
)
self.router = Router(
lifespan=lifespan,
formats=self.formats,
redirect_slashes=redirect_slashes,
max_request_size=max_request_size,
auto_etag=auto_etag,
auto_vary=auto_vary,
request_timeout=request_timeout,
ws_idle_timeout=ws_idle_timeout,
trace_dispatch=trace_dispatch,
problem_details=problem_details,
csrf=bool(csrf),
)
self.router.api = self
# Drain in-flight background tasks on shutdown rather than abandoning
# them when the process exits.
self.add_event_handler("shutdown", self._drain_background_tasks)
static_dir_explicit = static_dir is not _UNSET
if not static_dir_explicit:
static_dir = "static"
if static_dir is not None:
if static_route is None:
static_route = ""
static_dir = Path(static_dir).resolve()
self.static_dir = static_dir
self.static_route = static_route
self.implicit_static_fallback = bool(implicit_static_fallback)
self.hsts_enabled = enable_hsts
self._security_headers = security_headers
self.cors = cors
self.cors_params = cors_params
self.debug = debug
self.problem_details = bool(problem_details)
self.problem_handler = problem_handler
if not allowed_hosts:
allowed_hosts = ["*"]
self.allowed_hosts = allowed_hosts
# Never create static_dir implicitly: mount it only if it already
# exists. An explicitly-passed static_dir that's missing is an error;
# the default ("static") is simply skipped when absent.
if self.static_dir is not None:
if self.static_dir.is_dir():
self.mount(self.static_route, self.static_app)
elif static_dir_explicit:
raise FileNotFoundError(
f"static_dir {str(self.static_dir)!r} does not exist or is "
"not a directory. Create it, or pass static_dir=None to "
"disable static file serving."
)
self._session = None
self._session_base_url = None
self.default_endpoint = None
# Deferred middleware stack: collect config as data now and assemble the
# ASGI stack lazily on first request. ServerErrorMiddleware then becomes
# the outermost application layer (catching errors from every other
# middleware) while the observability tier (logging / request-id /
# metrics) wraps even it, so 500s still get X-Request-ID and a real
# logged status — the reconciliation v4.1 couldn't reach eagerly.
self._user_middleware: list[_MW] = []
self._middleware_stack: ASGIApp | None = None
self._exception_handlers: dict[Any, Callable] = {
HTTPException: _negotiated_http_error,
Exception: _negotiated_server_error,
}
self._gzip = gzip
self._cors_params = self.cors_params if cors else None
self._enable_logging = bool(enable_logging)
self._request_id = bool(request_id)
self._trust_proxy_headers = bool(trust_proxy_headers)
self._metrics = None
self._session_mw: _MW | None = None
if metrics_route:
from .ext.metrics import MetricsCollector
if metrics_buckets is None:
self.metrics = MetricsCollector()
else:
self.metrics = MetricsCollector(buckets=metrics_buckets)
self._metrics = self.metrics
def _metrics_view(req, resp):
resp.headers["Content-Type"] = "text/plain; version=0.0.4"
resp.content = self.metrics.render()
self.add_route(metrics_route, _metrics_view, static=False)
self._health_checks: dict[str, Callable] = {}
self._health_route = health_route
self._health_route_added = False
if health_route:
self._ensure_health_route()
# Sessions, secure by default. sessions="auto" (default) signs with the
# given key / RESPONDER_SECRET_KEY, else mints a random per-process key
# with a loud warning; sessions=True requires a real key; sessions=False
# disables session middleware entirely. Cookies are Secure in production
# (session_https_only=None) unless debug.
if sessions not in (True, False, "auto"):
raise ValueError("sessions= must be True, False, or 'auto'")
if sessions is False and session_backend is not None:
raise ValueError(
"session_backend was provided but sessions=False. Use "
"sessions='auto' (or True) to enable it, or drop the backend."
)
self.sessions_enabled = sessions is not False
if csrf and not self.sessions_enabled:
raise ValueError(
"csrf=True requires sessions: the CSRF token lives in the "
"session. Drop sessions=False, or disable CSRF protection."
)
if self.sessions_enabled:
effective_https_only = (
(not debug) if session_https_only is None else session_https_only
)
if session_same_site == "none" and not effective_https_only:
raise ValueError(
"session_same_site='none' requires a Secure cookie; set "
"session_https_only=True (browsers reject SameSite=None "
"without Secure)."
)
common_opts = {
"https_only": effective_https_only,
"same_site": session_same_site,
"max_age": session_max_age,
}
if session_backend is not None:
from .ext.sessions import ServerSessionMiddleware
opts = dict(common_opts)
if session_cookie is not None:
opts["cookie_name"] = session_cookie
self._session_mw = _MW(
ServerSessionMiddleware, {"backend": session_backend, **opts}
)
else:
from .ext.sessions import resolve_secret_key
self.secret_key = resolve_secret_key(
secret_key, sessions=sessions, debug=debug
)
opts = dict(common_opts)
if session_cookie is not None:
opts["session_cookie"] = session_cookie
self._session_mw = _MW(
SessionMiddleware, {"secret_key": self.secret_key, **opts}
)
if openapi or docs_route:
try:
from .ext.openapi import OpenAPISchema
except ImportError as ex:
raise ImportError(
"The dependencies for the OpenAPI extension are not installed. "
'Install them using: pip install "responder[orjson]"'
) from ex
self.openapi = OpenAPISchema(
app=self,
title=title,
version=version,
openapi=openapi,
docs_route=docs_route,
description=description,
terms_of_service=terms_of_service,
contact=contact,
license=license,
openapi_route=openapi_route,
static_route=static_route,
openapi_theme=openapi_theme,
servers=openapi_servers,
)
for auth_scheme in self._auth:
if _auth_has_security_scheme(auth_scheme):
self.add_security_scheme(auth_scheme)
self.templates = Templates(directory=templates_dir, autoescape=auto_escape)
# request_id / logging are installed as middleware in the observability
# tier by build_middleware_stack(); here we only configure the logger.
if enable_logging:
import logging as _logging
from .ext.logging import get_logger, setup_logging
log_level = _logging.DEBUG if debug else _logging.INFO
setup_logging(level=log_level)
self.log = get_logger("responder.app")
else:
import logging as _logging
self.log = _logging.getLogger("responder.app")
@property
def requests(self):
"""A test client connected to the ASGI app. Lazily initialized."""
return self._test_client()
@property
def async_requests(self):
"""An async test client wired to the ASGI app (``httpx.AsyncClient``).
The async mirror of :attr:`requests`; ``async with`` also runs the
app's lifespan. See :class:`responder.testing.AsyncTestClient`.
"""
from responder.testing import AsyncTestClient
return AsyncTestClient(self)
@property
def static_app(self):
"""The Starlette ``StaticFiles`` application for serving static assets."""
if not hasattr(self, "_static_app"):
assert self.static_dir is not None
self._static_app = StaticFiles(directory=self.static_dir)
return self._static_app
[docs]
def before_request(self, websocket=False):
"""Register a function to run before every request.
If the hook sets ``resp.status_code``, the route handler is skipped
and the response is sent immediately (short-circuiting).
:param websocket: If ``True``, register as a WebSocket before-request hook instead of HTTP.
Usage::
@api.before_request()
def check_auth(req, resp):
if "Authorization" not in req.headers:
resp.status_code = 401
resp.media = {"error": "unauthorized"}
""" # noqa: E501
# Allow both @api.before_request and @api.before_request().
if callable(websocket):
f = websocket
self.router.before_request(f, websocket=False)
return f
def decorator(f):
self.router.before_request(f, websocket=websocket)
return f
return decorator
[docs]
def dependency(self, name=None, *, scope="request"):
"""Register a dependency provider, injected into views by parameter name.
Any view parameter (beyond ``req`` and ``resp``) whose name matches a
registered dependency receives the provider's value. Providers may be
sync or async functions, or generators — code after ``yield`` runs as
teardown once the response is sent. Providers accepting a parameter
receive the current :class:`Request`. Each dependency is resolved at
most once per request. Path parameters take precedence over
dependencies of the same name.
:param name: The injection name. Defaults to the provider's ``__name__``.
:param scope: ``"request"`` (default) resolves per request;
``"app"`` resolves once on first use and caches for the
application's lifetime — generator teardown then runs at
shutdown. App-scoped providers cannot take parameters.
Usage::
@api.dependency()
async def db():
conn = await create_connection()
yield conn
await conn.close()
@api.route("/users/{id:int}")
async def get_user(req, resp, *, id, db):
resp.media = await db.fetch_user(id)
An app-scoped dependency, shared across all requests::
@api.dependency(scope="app")
async def pool():
pool = await create_pool()
yield pool
await pool.close() # runs at application shutdown
"""
if callable(name): # Used as a bare decorator: @api.dependency
self.router.add_dependency(name.__name__, name)
return name
def decorator(f):
self.router.add_dependency(name or f.__name__, f, scope=scope)
return f
return decorator
[docs]
def add_dependency(self, name, provider, *, scope="request"):
"""Register a dependency provider under an explicit name.
:param name: The view parameter name to inject as.
:param provider: The provider function (sync/async function or generator).
:param scope: ``"request"`` (default) or ``"app"``.
"""
self.router.add_dependency(name, provider, scope=scope)
[docs]
@contextlib.contextmanager
def dependency_overrides(self, **overrides):
"""Temporarily override dependencies (for tests); restores on exit.
Each value may be a provider (a callable, with full sub-dependency and
request injection) or a bare value, which is wrapped automatically.
Overrides are request-scoped, so they replace and bypass the cache of an
``app``-scoped dependency too::
with api.dependency_overrides(db=fake_db):
api.requests.get("/users")
:param overrides: ``name=provider_or_value`` pairs to override.
"""
registry = self.router.dependency_overrides
sentinel = object()
# Snapshot only the keys this block touches, and restore them
# individually — so nested or non-LIFO override blocks don't clobber
# each other's keys.
previous: dict[str, Any] = {
name: registry.get(name, sentinel) for name in overrides
}
for name, value in overrides.items():
provider = value if callable(value) else _const_provider(value)
registry[name] = (provider, "request")
try:
yield
finally:
for name, prev in previous.items():
if prev is sentinel:
registry.pop(name, None)
else:
registry[name] = prev
[docs]
def add_health_check(self, name, check):
"""Register a readiness check run by the health endpoint.
``check`` is a sync or async callable; it passes unless it returns
``False`` or raises. The endpoint returns ``200`` when every check
passes and ``503`` otherwise, with per-check JSON. The route (default
``/health``, or the ``health_route=`` you set) is added on first use.
:param name: A label for the check, used as its key in the JSON body.
:param check: The check callable.
"""
self._health_checks[name] = check
self._ensure_health_route()
def _ensure_health_route(self):
if self._health_route_added:
return
async def _health_view(req, resp):
checks: dict = {}
healthy = True
for name, check in self._health_checks.items():
try:
if inspect.iscoroutinefunction(check):
outcome = await check()
else:
outcome = await run_in_threadpool(check)
if inspect.isawaitable(outcome):
outcome = await outcome
passed = outcome is not False
checks[name] = {"status": "ok" if passed else "error"}
healthy = healthy and passed
except Exception as exc:
healthy = False
checks[name] = {"status": "error", "detail": str(exc)}
resp.status_code = status_codes.HTTP_200 if healthy else 503
resp.media = {"status": "ok" if healthy else "error", "checks": checks}
_health_view._include_in_schema = False # type: ignore[attr-defined]
self.add_route(self._health_route or "/health", _health_view, static=False)
self._health_route_added = True
[docs]
def after_request(self, f=None):
"""Register a function to run after every request.
Works both bare and called: ``@api.after_request`` or
``@api.after_request()``.
Usage::
@api.after_request
def add_request_id(req, resp):
resp.headers["X-Request-ID"] = str(uuid.uuid4())
"""
if callable(f): # used as a bare decorator: @api.after_request
self.router.after_request(f)
return f
def decorator(func):
self.router.after_request(func)
return func
return decorator
@property
def app(self) -> ASGIApp:
"""The assembled ASGI middleware stack, built lazily on first access."""
if self._middleware_stack is None:
self._middleware_stack = self.build_middleware_stack()
return self._middleware_stack
[docs]
def build_middleware_stack(self) -> ASGIApp:
"""Assemble the full ASGI stack from the collected configuration.
Outermost → innermost: logging/request-id → metrics → ServerError →
user middleware → trusted-host → hsts → cors → sessions → gzip →
ExceptionMiddleware → router. ServerErrorMiddleware is the outermost
*application* layer (it catches errors from every middleware below it),
while the observability tier wraps even it so a rendered 500 still
carries ``X-Request-ID`` and is logged with its real status.
"""
debug = self.debug
error_handler = self._exception_handlers.get(500) or self._exception_handlers.get(
Exception
)
exc_handlers = {
k: h for k, h in self._exception_handlers.items() if k not in (500, Exception)
}
app: ASGIApp = self.router
app = ExceptionMiddleware(app, handlers=exc_handlers, debug=debug)
if self._gzip:
app = GZipMiddleware(app)
if self._session_mw is not None:
app = self._session_mw.cls(app, **self._session_mw.options)
if self._cors_params is not None:
app = CORSMiddleware(app, **self._cors_params)
if self._security_headers:
from .middleware import SecurityHeadersMiddleware
opts = (
self._security_headers if isinstance(self._security_headers, dict) else {}
)
app = SecurityHeadersMiddleware(app, **opts)
if self.hsts_enabled:
from .middleware import HSTSMiddleware
app = HTTPSRedirectMiddleware(app)
app = HSTSMiddleware(app)
app = TrustedHostMiddleware(app, allowed_hosts=self.allowed_hosts)
for mw in reversed(self._user_middleware): # index 0 wrapped last = outermost
app = mw.cls(app, **mw.options)
app = ServerErrorMiddleware(app, handler=error_handler, debug=debug)
if self._metrics is not None:
from .ext.metrics import MetricsMiddleware
app = MetricsMiddleware(app, collector=self._metrics)
if self._enable_logging:
from .ext.logging import LoggingMiddleware
app = LoggingMiddleware(app, trust_proxy_headers=self._trust_proxy_headers)
elif self._request_id:
from .ext.logging import RequestIDMiddleware
app = RequestIDMiddleware(app)
if self._trust_proxy_headers:
# Outermost: every layer below (logging IPs, HTTPS redirects,
# trusted-host checks, secure-cookie logic, url building) sees the
# scheme/host/client the original client actually used.
from .middleware import ProxyHeadersMiddleware
app = ProxyHeadersMiddleware(app)
return app
[docs]
def add_middleware(self, middleware_cls, **middleware_config):
"""Add ASGI middleware to the application (valid after construction).
User middleware sits just inside ``ServerErrorMiddleware`` (so its
errors are caught and rendered) and the most-recently-added runs first.
To wrap *everything* (including error rendering), wrap the API object:
``asgi = MyMiddleware(api)``.
:param middleware_cls: A Starlette-compatible middleware class.
:param middleware_config: Keyword arguments passed to the constructor.
"""
self._user_middleware.insert(0, _MW(middleware_cls, middleware_config))
self._middleware_stack = None # rebuild lazily
[docs]
def middleware(self, middleware_type="http"):
"""Register function-style HTTP middleware (decorator).
The decorated function receives the Starlette ``Request`` and a
``call_next`` callable, and returns the response to send — no ASGI
class boilerplate required::
import time
@api.middleware("http")
async def add_timing(request, call_next):
start = time.perf_counter()
response = await call_next(request)
elapsed = time.perf_counter() - start
response.headers["X-Response-Time"] = f"{elapsed:.4f}s"
return response
Synchronous functions work too: they are offloaded to the threadpool
(like sync views), and the ``call_next`` they receive is a plain
blocking callable.
Function middleware registers through the same stack as
:meth:`add_middleware`, so the two compose freely. **Order:** among
user middleware — function-style or class-based alike — the
most-recently-registered is the outermost and sees the request first
(and the response last). Non-HTTP traffic (WebSockets, lifespan)
passes through untouched.
:param middleware_type: Only ``"http"`` is supported (the default).
The decorator may also be applied bare (``@api.middleware``).
"""
from .middleware import FunctionMiddleware
if callable(middleware_type):
# Applied bare: @api.middleware
self.add_middleware(FunctionMiddleware, func=middleware_type)
return middleware_type
if middleware_type != "http":
raise ValueError(
f"Only 'http' middleware is supported, got {middleware_type!r}."
)
def decorator(func):
self.add_middleware(FunctionMiddleware, func=func)
return func
return decorator
[docs]
def add_exception_handler(self, exc_class_or_status_code, handler):
"""Register a handler for an exception type or status code.
``handler`` is a Responder-style ``(req, resp, exc)`` callable (sync or
async). A handler for ``500``/``Exception`` installs the catch-all
server-error handler (ignored under ``debug=True``, which shows the
traceback); any other exception/status routes through the exception
middleware.
"""
self._exception_handlers[exc_class_or_status_code] = self._wrap_exc_handler(
handler
)
self._middleware_stack = None
def _wrap_exc_handler(self, func):
"""Adapt a ``(req, resp, exc)`` handler to a Starlette ``(request, exc)``."""
is_async = inspect.iscoroutinefunction(func)
async def _adapter(request, exc):
req = Request(request.scope, request.receive, api=self, formats=self.formats)
resp = Response(req=req, formats=self.formats)
if is_async:
await func(req, resp, exc)
else:
# Keep a blocking handler off the event loop.
await run_in_threadpool(func, req, resp, exc)
if resp.status_code is None:
resp.status_code = 500
body, headers = await resp.body
return StarletteResponse(
content=body, status_code=resp.status_code, headers=headers
)
return _adapter
[docs]
def exception_handler(self, exception_cls):
"""Register a handler for a specific exception type.
Usage::
@api.exception_handler(ValueError)
async def handle_value_error(req, resp, exc):
resp.status_code = 400
resp.media = {"error": str(exc)}
"""
def decorator(func):
self.add_exception_handler(exception_cls, func)
return func
return decorator
[docs]
def schema(self, name, **options):
"""
Decorator for creating new routes around function and class definitions.
Usage::
from marshmallow import Schema, fields
@api.schema("Pet")
class PetSchema(Schema):
name = fields.Str()
"""
def decorator(f):
self.openapi.add_schema(name=name, schema=f, **options)
return f
return decorator
[docs]
def generate_client(self, path=None, *, class_name="APIClient", language="python"):
"""Generate a client from this app's OpenAPI schema.
Returns source code by default. When ``path`` is provided, writes the
client module there and returns the resulting ``Path``.
Usage::
source = api.generate_client(class_name="AcmeClient")
api.generate_client("clients/acme.py", class_name="AcmeClient")
api.generate_client(
"clients/acme.ts",
class_name="AcmeClient",
language="typescript",
)
"""
if not hasattr(self, "openapi"):
raise RuntimeError(
"OpenAPI is not enabled; pass openapi=... (or docs_route=...) "
"to API() before generating a client."
)
from .ext.clientgen import generate_client, write_client
if path is None:
return generate_client(self.openapi, class_name=class_name, language=language)
return write_client(self.openapi, path, class_name=class_name, language=language)
[docs]
def path_matches_route(self, path):
"""Given a path portion of a URL, tests that it matches against any registered route.
:param path: The path portion of a URL (e.g. ``"/hello"``), to test all
known routes against. An ASGI scope mapping is also accepted.
""" # noqa: E501 (Line too long)
for route in self.router.routes:
scopes: tuple[Any, ...]
if isinstance(path, str):
# The documented contract takes a plain path string; build
# minimal HTTP and websocket scopes for it, matching by path
# regardless of any per-route method restriction. Each route
# type only ever matches the scope of its own kind.
http_scope = {"type": "http", "path": path}
methods = getattr(route, "methods", None)
if methods:
http_scope["method"] = next(iter(methods))
scopes = (http_scope, {"type": "websocket", "path": path})
else:
scopes = (path,)
for scope in scopes:
match, _ = route.matches(scope)
if match:
return route
return None
[docs]
def add_route(
self,
route=None,
endpoint=None,
*,
default=False,
static=True,
check_existing=True,
websocket=False,
before_request=False,
methods=None,
name=None,
):
"""Adds a route to the API.
:param route: A string representation of the route.
:param endpoint: The endpoint for the route -- can be a callable, or a class.
:param default: If ``True``, all unknown requests will route to this view.
:param static: If ``True`` and no endpoint was passed, render
``static/index.html`` as a legacy default route only
when ``implicit_static_fallback=True``.
:param methods: Optional list of HTTP methods (e.g. ``["GET", "POST"]``).
:param name: Optional route name for :meth:`url_for` reverse lookup.
Calling ``add_route()`` without an ``endpoint`` now raises by default.
Pass an endpoint explicitly, or serve static assets via
``static_dir``/``static_route``. Apps that need the old fallback
during migration can pass ``API(implicit_static_fallback=True)``.
""" # noqa: E501
if static and not endpoint:
if not self.implicit_static_fallback:
raise ValueError(
"Calling add_route() without an endpoint requires "
"implicit_static_fallback=True. Pass an endpoint explicitly "
"(with default=True for a catch-all), or serve static assets "
"via static_dir/static_route."
)
if self.static_dir is None:
raise ValueError(
"Cannot add a static fallback route: static_dir is disabled"
)
endpoint = self._static_response
default = True
self.router.add_route(
route,
endpoint,
default=default,
websocket=websocket,
before_request=before_request,
check_existing=check_existing,
methods=methods,
name=name,
)
async def _static_response(self, req, resp):
assert self.static_dir is not None
index = (self.static_dir / "index.html").resolve()
# Read off the event loop — this runs from an async dispatch path.
contents = await run_in_threadpool(_read_text_if_exists, index)
if contents is not None:
resp.html = contents
else:
resp.status_code = status_codes.HTTP_404
resp.text = "Not found."
[docs]
def redirect(
self,
resp: Response,
location: str,
*,
set_text: bool = True,
status_code: int | None = None,
permanent: bool = False,
allow_external: bool = True,
) -> None:
"""
Redirects a given response to a given location.
:param resp: The Response to mutate.
:param location: The location of the redirect.
:param set_text: If ``True``, sets the Redirect body content automatically.
:param status_code: an `API.status_codes` attribute, or an integer,
representing the HTTP status code of the redirect.
Defaults to ``307`` (``308`` with ``permanent=True``).
:param permanent: If ``True``, send a ``308 Permanent Redirect`` instead
of the default ``307``.
:param allow_external: If ``False``, refuse (with a ``400``) to redirect to
an external URL — pass this for user-supplied locations.
"""
resp.redirect(
location,
set_text=set_text,
status_code=status_code,
permanent=permanent,
allow_external=allow_external,
)
[docs]
def on_event(self, event_type: str, **args: Any) -> Callable[..., Any]:
"""Decorator for registering functions or coroutines to run at certain events
Supported events: startup, shutdown
Usage::
@api.on_event('startup')
async def open_database_connection_pool():
...
@api.on_event('shutdown')
async def close_database_connection_pool():
...
"""
def decorator(func):
self.add_event_handler(event_type, func, **args)
return func
return decorator
[docs]
def add_event_handler(self, event_type, handler):
"""Adds an event handler to the API.
:param event_type: A string in ("startup", "shutdown")
:param handler: The function to run. Can be either a function or a coroutine.
"""
self.router.add_event_handler(event_type, handler)
async def _drain_background_tasks(self):
"""Shutdown handler: drain the background pool off the event loop."""
await run_in_threadpool(self.background.shutdown)
[docs]
def add_security_scheme(self, name, scheme=None, *, default=False):
"""Register an OpenAPI security scheme (enables Swagger's Authorize button).
Accepts either an auth helper carrying its own definition
(``add_security_scheme(BearerAuth(...))``) or an explicit ``name`` plus a
scheme dict. With ``default=True`` the scheme becomes a global
requirement applied to every documented operation.
:param name: The scheme name, or an auth helper object.
:param scheme: The OpenAPI security-scheme dict (omit when passing a helper).
:param default: If ``True``, require this scheme on all operations.
"""
if not hasattr(self, "openapi"):
raise RuntimeError(
"OpenAPI is not enabled; pass openapi=... (or docs_route=...) to API()."
)
if scheme is None: # an auth helper carrying its own name + definition
name, scheme = name.scheme_name, name.security_scheme()
self.openapi.add_security_scheme(name, scheme, default=default)
[docs]
def policy(self, name, auth):
"""Create and register a named auth policy.
The returned policy can be passed anywhere ``auth=`` accepts an auth
helper. Naming a policy makes route declarations describe intent while
preserving the wrapped auth scheme's runtime and OpenAPI behavior::
admin = api.policy("admin", bearer.requires("admin"))
@api.get("/admin", auth=admin)
def dashboard(req, resp, *, user): ...
"""
from .ext.auth import AuthPolicy
policy = AuthPolicy(name, auth)
if policy.name in self.auth_policies:
raise ValueError(f"Auth policy {policy.name!r} is already registered")
self.auth_policies[policy.name] = policy
if hasattr(self, "openapi") and _auth_has_security_scheme(policy):
self.add_security_scheme(policy)
return policy
[docs]
def route(
self,
route=None,
*,
response_model=None,
params_model=None,
include_in_schema=True,
security=None,
tags=None,
summary=None,
description=None,
operation_id=None,
deprecated=None,
responses=None,
examples=None,
response_examples=None,
openapi_extra=None,
status_code=None,
before=None,
after=None,
auth=_UNSET,
csrf=None,
dependencies=None,
_stream_mode=None,
_stream_model=None,
_stream_heartbeat=None,
**options,
):
"""Decorator for creating new routes around function and class definitions.
Usage::
@api.route("/hello")
def hello(req, resp):
resp.text = "hello, world!"
A route can declare its default success status with ``status_code=``;
``resp.status_code`` is pre-seeded with it before the handler runs
(assigning ``resp.status_code`` in the handler still wins), and the
OpenAPI document keys the success response under it instead of
``200``::
@api.route("/items", methods=["POST"], status_code=201)
async def create_item(req, resp, *, item: ItemIn):
resp.media = {"id": 1, **item.model_dump()}
Return annotations provide response validation and OpenAPI
documentation without repeating the model in the decorator::
from pydantic import BaseModel
class ItemIn(BaseModel):
name: str
price: float
class ItemOut(BaseModel):
id: int
name: str
price: float
@api.route("/items", methods=["POST"])
async def create_item(req, resp, *, item: ItemIn) -> ItemOut:
return ItemOut(id=1, **item.model_dump())
Pydantic ``TypeAdapter``-compatible annotations such as
``list[ItemOut]`` and ``ItemOut | ErrorOut`` also infer. Pass
``response_model=False`` to disable inference for one route.
Additional statuses can declare their own validated contract with
``responses={404: ErrorOut}``. Use
``responses={404: {"model": ErrorOut, "description": "Not found"}}``
to add OpenAPI metadata to that contract.
Query parameters validate the same way with ``params_model`` —
invalid queries get a ``422``, valid ones land on
``req.state.validated_params``::
class SearchParams(BaseModel):
q: str
limit: int = 10
@api.route("/search", params_model=SearchParams)
async def search(req, resp):
params = req.state.validated_params
resp.media = {"q": params.q, "limit": params.limit}
"""
default_status_code = None
if status_code is not None:
default_status_code = int(status_code)
if not 100 <= default_status_code <= 599:
raise ValueError(
f"status_code= must be a valid HTTP status code (got {status_code!r})"
)
success_key = (
str(default_status_code) if default_status_code is not None else "200"
)
def decorator(f):
if _stream_mode is not None:
if _stream_mode not in ("sse", "ndjson"):
raise ValueError(f"Unknown typed stream mode {_stream_mode!r}")
if options.get("websocket"):
raise ValueError(
"Typed event streams are HTTP routes, not WebSockets"
)
if response_model is not None:
raise TypeError(
"Typed event streams use event_model= or item_model=; "
"response_model= describes buffered responses"
)
if default_status_code is not None and not response_status_allows_body(
default_status_code
):
raise ValueError(
f"HTTP {default_status_code} cannot carry a typed stream"
)
if _stream_heartbeat is not None and (
_stream_mode != "sse"
or isinstance(_stream_heartbeat, bool)
or not isinstance(_stream_heartbeat, (int, float))
or _stream_heartbeat <= 0
):
raise ValueError(
"A typed SSE heartbeat must be a positive number of seconds"
)
target = getattr(f, "on_get", None) if inspect.isclass(f) else f
stream_model = _stream_model
if stream_model is None and target is not None:
stream_model = inferred_stream_model(_view_return_hint(target))
if stream_model is None:
model_option = (
"event_model=" if _stream_mode == "sse" else "item_model="
)
raise TypeError(
f"@api.{_stream_mode} requires an iterator return annotation "
f"or an explicit {model_option} type"
)
_validate_declared_response_model(
stream_model,
label=(
"event_model=" if _stream_mode == "sse" else "item_model="
),
)
f._stream_mode = _stream_mode
f._stream_model = stream_model
f._stream_heartbeat = _stream_heartbeat
else:
# Route metadata is snapshotted during registration. Clearing
# these prevents a shared handler's earlier stream decoration
# from leaking into a later ordinary route.
f._stream_mode = None
f._stream_model = None
f._stream_heartbeat = None
auth_is_explicit = auth is not _UNSET
route_auth = self._auth if not auth_is_explicit else _as_tuple(auth)
# Per-route override of API(csrf=...): False exempts (e.g. a
# webhook receiver), True protects a single route app-wide-off.
# Always record this registration's value (None = "inherit the
# app default") on the view so add_route can snapshot it onto the
# Route — otherwise the attribute lingers on a shared function
# across re-registrations, or is inherited through a CBV's MRO,
# silently rewriting another route's protection.
if csrf:
if options.get("websocket"):
raise ValueError(
f"csrf=True is not supported on the WebSocket route "
f"{route!r}: the CSRF token check only runs on the HTTP "
"dispatch path, so it would silently never enforce. "
"Validate the Origin header in the handler instead."
)
if not self.sessions_enabled:
raise ValueError(
f"csrf=True on route {route!r} requires sessions: the "
"CSRF token lives in the session. Drop sessions=False, "
"or leave the route unprotected."
)
f._csrf = None if csrf is None else bool(csrf)
if before is not None:
f._route_before = _as_tuple(before)
if after is not None:
f._route_after = _as_tuple(after)
if dependencies is not None:
route_dependencies = _as_tuple(dependencies)
invalid = [d for d in route_dependencies if not isinstance(d, _Depends)]
if invalid:
raise TypeError(
"Route dependencies must be declared as Depends(...) markers."
)
f._route_dependencies = route_dependencies
if route_auth:
f._route_auth = route_auth
if security is None:
requirements = []
for a in route_auth:
requirement = _auth_security_requirement(a)
if requirement is None:
continue
if isinstance(requirement, list):
requirements.extend(requirement)
else:
requirements.append(requirement)
if requirements:
f._security = requirements
if hasattr(self, "openapi"):
for auth_scheme in route_auth:
if _auth_has_security_scheme(auth_scheme):
self.add_security_scheme(auth_scheme)
elif auth_is_explicit and security is None:
f._security = []
if response_model is not None:
if response_model is not False:
_validate_declared_response_model(
response_model, label="response_model="
)
f._response_model = response_model
# Generic response models (list[Model], unions, Page[Item], …)
# carry no single component name; the OpenAPI builder unpacks
# them via TypeAdapter.
if hasattr(self, "openapi") and _registers_as_named_component(
response_model
):
self.openapi.add_schema(
response_model.__name__, response_model, check_existing=False
)
if params_model is not None:
f._params_model = params_model
if default_status_code is not None:
f._default_status_code = default_status_code
if security is not None:
f._security = security
meta = {}
if tags is not None:
meta["tags"] = tags
if summary is not None:
meta["summary"] = summary
if description is not None:
meta["description"] = description
if operation_id is not None:
meta["operationId"] = operation_id
if deprecated is not None:
meta["deprecated"] = deprecated
response_models: dict[int, Any] = {}
if responses is not None:
normalized_responses, response_models = _normalize_openapi_responses(
responses
)
meta["responses"] = normalized_responses
if _stream_mode is not None and int(success_key) in response_models:
raise TypeError(
"A typed stream cannot also declare a buffered model for its "
f"{success_key} success response"
)
f._response_models = response_models
response_example_meta: dict[str, Any] = {}
if examples is not None:
_add_openapi_examples(response_example_meta, success_key, examples)
if response_examples is not None:
response_example_meta = _deep_merge_dicts(
response_example_meta,
_normalize_openapi_response_examples(response_examples),
)
if response_example_meta:
known_responses = set(meta.get("responses", {}))
for example_status, response in response_example_meta.items():
if (
example_status != success_key
and example_status not in known_responses
):
response.setdefault("description", "Response")
meta["responses"] = _deep_merge_dicts(
meta.get("responses", {}),
response_example_meta,
)
if openapi_extra is not None:
if not isinstance(openapi_extra, Mapping):
raise TypeError("openapi_extra= must be a mapping")
meta = _deep_merge_dicts(meta, openapi_extra)
if meta:
f._openapi_meta = meta
if not include_in_schema:
f._include_in_schema = False
if (
response_model is None
and _stream_mode is None
and not options.get("websocket")
):
_diagnose_response_annotation(f, route)
self.add_route(route, f, **options)
return f
return decorator
[docs]
def get(self, route=None, **options):
"""Register a route for ``GET`` (sugar for ``route(methods=["GET"])``)."""
return self.route(route, methods=["GET"], **options)
[docs]
def sse(self, route=None, *, event_model=None, heartbeat=None, **options):
"""Register a typed Server-Sent Events ``GET`` route.
The event data contract is inferred from ``AsyncIterator[Item]`` or
``AsyncIterator[SSE[Item]]``. Pass ``event_model=Item`` when the return
annotation cannot express it.
"""
if heartbeat is not None and (
isinstance(heartbeat, bool) or not isinstance(heartbeat, (int, float))
or heartbeat <= 0
):
raise ValueError("heartbeat= must be a positive number of seconds")
if "methods" in options:
raise TypeError("@api.sse always registers a GET route")
return self.route(
route,
methods=["GET"],
_stream_mode="sse",
_stream_model=event_model,
_stream_heartbeat=heartbeat,
**options,
)
[docs]
def ndjson(self, route=None, *, item_model=None, **options):
"""Register a typed newline-delimited JSON ``GET`` route."""
if "methods" in options:
raise TypeError("@api.ndjson always registers a GET route")
return self.route(
route,
methods=["GET"],
_stream_mode="ndjson",
_stream_model=item_model,
**options,
)
[docs]
def post(self, route=None, **options):
"""Register a route for ``POST`` (sugar for ``route(methods=["POST"])``)."""
return self.route(route, methods=["POST"], **options)
[docs]
def put(self, route=None, **options):
"""Register a route for ``PUT`` (sugar for ``route(methods=["PUT"])``)."""
return self.route(route, methods=["PUT"], **options)
[docs]
def patch(self, route=None, **options):
"""Register a route for ``PATCH`` (sugar for ``route(methods=["PATCH"])``)."""
return self.route(route, methods=["PATCH"], **options)
[docs]
def delete(self, route=None, **options):
"""Register a route for ``DELETE`` (sugar for ``route(methods=["DELETE"])``)."""
return self.route(route, methods=["DELETE"], **options)
[docs]
def head(self, route=None, **options):
"""Register a route for ``HEAD`` (sugar for ``route(methods=["HEAD"])``)."""
return self.route(route, methods=["HEAD"], **options)
[docs]
def options(self, route=None, **options):
"""Register a route for ``OPTIONS`` (sugar for ``route(methods=["OPTIONS"])``)."""
return self.route(route, methods=["OPTIONS"], **options)
[docs]
def websocket_route(self, route=None, **options):
"""Register a WebSocket route (sugar for ``route(websocket=True)``)."""
return self.route(route, websocket=True, **options)
[docs]
def graphql(
self,
route="/graphql",
*,
schema,
graphiql=True,
introspection=True,
max_depth=None,
partial_data_status=200,
):
"""Mount a GraphQL API at the given route.
Usage::
import graphene
class Query(graphene.ObjectType):
hello = graphene.String(name=graphene.String(default_value="stranger"))
def resolve_hello(self, info, name):
return f"Hello {name}"
api.graphql("/graphql", schema=graphene.Schema(query=Query))
For production, disable the in-browser IDE and introspection and cap
query depth::
api.graphql(
"/graphql", schema=schema,
graphiql=api.debug, introspection=api.debug, max_depth=10,
)
:param route: The URL path for the GraphQL endpoint.
:param schema: A Graphene schema instance.
:param graphiql: Serve the in-browser GraphiQL IDE for HTML ``GET``
requests (default ``True``).
:param introspection: Allow schema-introspection queries (default ``True``).
:param max_depth: Reject queries nested deeper than this (default unlimited).
:param partial_data_status: HTTP status for GraphQL responses that
contain both ``data`` and ``errors``. ``200`` (default) follows
the GraphQL-over-HTTP spec; ``400`` preserves Responder 8.x
behavior.
"""
from .ext.graphql import GraphQLView
self.add_route(
route,
GraphQLView(
api=self,
schema=schema,
graphiql=graphiql,
introspection=introspection,
max_depth=max_depth,
partial_data_status=partial_data_status,
),
)
[docs]
def mount(self, route, app):
"""Mounts an WSGI / ASGI application at a given route.
:param route: String representation of the route to be used
(shouldn't be parameterized).
:param app: The other WSGI / ASGI app.
"""
self.router.apps.update({route: app})
def _test_client(self, base_url="http://;", **options):
"""Build (or return the cached) Starlette TestClient for this app.
The client is cached per ``base_url``: repeated calls with the same
``base_url`` return the same client, while a different ``base_url``
builds a fresh one instead of silently reusing the old address.
Passing extra ``TestClient`` options builds a fresh client.
"""
if options:
from starlette.testclient import TestClient
return TestClient(self, base_url=base_url, **options)
if self._session is None or self._session_base_url != base_url:
from starlette.testclient import TestClient
self._session = TestClient(self, base_url=base_url)
self._session_base_url = base_url
return self._session
[docs]
def test_client(self, base_url="http://;", **options):
"""Return a Starlette ``TestClient`` connected to this app.
This is the configurable form of :attr:`requests`; pass normal
Starlette ``TestClient`` options such as
``raise_server_exceptions=False`` or a custom ``base_url``.
"""
return self._test_client(base_url=base_url, **options)
[docs]
def url_for(self, endpoint, **params):
"""Given an endpoint, returns a rendered URL for its route.
:param endpoint: The route endpoint you're searching for.
:param params: Data to pass into the URL generator (for parameterized URLs).
"""
return self.router.url_for(endpoint, **params)
[docs]
def template(self, filename, *args, **kwargs):
r"""Render a Jinja2 template file with the provided values.
:param filename: The filename of the jinja2 template, in ``templates_dir``.
:param \*args: Data to pass into the template.
:param \*\*kwargs: Data to pass into the template.
"""
return self.templates.render(filename, *args, **kwargs)
[docs]
def template_string(self, source, *args, **kwargs):
r"""Render a Jinja2 template string with the provided values.
:param source: The template to use, a Jinja2 template string.
:param \*args: Data to pass into the template.
:param \*\*kwargs: Data to pass into the template.
"""
return self.templates.render_string(source, *args, **kwargs)
[docs]
def serve(
self,
*,
address=None,
port=None,
debug=False,
server="uvicorn",
port_precedence="explicit",
**options,
):
"""
Run the application with an ASGI server.
If the ``PORT`` environment variable is set and no explicit ``port=``
is provided, requests will be served on that port automatically to all
known hosts. If both are set and disagree, the explicit ``port=`` wins
by default.
:param address: The address to bind to.
:param port: The port to bind to. If none is provided, one will be selected at random.
:param debug: Whether to run application in debug mode.
:param server: Server backend to use: ``"uvicorn"`` (default) or ``"granian"``.
:param port_precedence: ``"explicit"`` (default) lets an explicit ``port=`` win over a conflicting ``PORT`` environment variable; ``"env"`` preserves the legacy behavior.
:param options: Additional keyword arguments to send to the selected server.
""" # noqa: E501
if port_precedence not in ("env", "explicit"):
raise ValueError("port_precedence= must be 'env' or 'explicit'")
if "PORT" in os.environ:
env_port = int(os.environ["PORT"])
if port is not None and port != env_port:
if port_precedence == "env":
port = env_port
else:
port = env_port
if address is None:
address = "0.0.0.0" # noqa: S104
if address is None:
address = "127.0.0.1"
if port is None:
port = 5042
if debug:
options["log_level"] = "debug"
if server == "uvicorn":
uvicorn.run(self, host=address, port=port, **options)
return
if server == "granian":
self._serve_granian(address=address, port=port, **options)
return
raise ValueError(
f"Unsupported server {server!r}. Expected 'uvicorn' or 'granian'."
)
def _serve_granian(self, *, address, port, **options):
workers = options.pop("workers", None)
if workers not in (None, 1):
raise ValueError(
'api.run(server="granian") uses Granian embedded mode and does '
"not support multiple workers. Use the Granian CLI for "
"multi-worker production deployments."
)
try:
granian_embed = importlib.import_module("granian.server.embed")
granian_constants = importlib.import_module("granian.constants")
except ImportError as exc:
raise RuntimeError(
"Granian is not installed. Install it with: "
'pip install "responder[orjson,server]"'
) from exc
server = granian_embed.Server(
self,
address=address,
port=port,
interface=granian_constants.Interfaces.ASGI,
**options,
)
asyncio.run(server.serve())
[docs]
def run(self, **kwargs):
"""Run the application. Shorthand for :meth:`serve` that inherits the ``debug`` setting.
:param kwargs: Keyword arguments passed through to :meth:`serve`.
""" # noqa: E501
if "debug" not in kwargs:
kwargs.update({"debug": self.debug})
self.serve(**kwargs)
[docs]
def group(self, prefix):
"""Create a route group with a shared URL prefix.
Usage::
v1 = api.group("/v1")
@v1.route("/users")
def list_users(req, resp):
resp.media = []
@v1.route("/users/{id:int}")
def get_user(req, resp, *, id):
resp.media = {"id": id}
For routes declared in separate modules (without an ``API`` instance),
use :class:`responder.Router` with :meth:`include_router` instead.
"""
return RouteGroup(api=self, prefix=prefix)
[docs]
def include_router(
self,
router: _IncludableRouter,
*,
prefix: str = "",
tags: list[str] | None = None,
dependencies: Any = None,
auth: Any = _UNSET,
) -> None:
"""Attach a standalone :class:`responder.Router`'s recorded declarations.
Each recorded route is replayed through :meth:`route`, so auth
inheritance, ``Depends`` guards, and OpenAPI metadata behave exactly as
if the route had been declared on the API directly. Group-level values
compose: prefixes concatenate, ``tags`` merge (group tags first),
``dependencies`` run before route-level ones, and a route's own
``auth=`` wins over the router's, which wins over the one given here.
Inclusion is a snapshot — routes declared on the router afterwards are
not picked up. Including the same router at two prefixes is fine, but
including the same *view function* twice with different effective
auth/tags/dependencies raises, since route metadata is attached to the
view itself.
:param router: The :class:`responder.Router` to include.
:param prefix: URL prefix prepended to the router's own prefix.
:param tags: OpenAPI tags merged into every included route.
:param dependencies: ``Depends(...)`` guards run before every included
route.
:param auth: Auth helper(s) for included routes that don't set their own.
"""
if not isinstance(router, _IncludableRouter):
raise TypeError(
f"include_router() expects a responder.Router, "
f"got {type(router).__name__}"
)
prefix = _normalize_prefix(prefix)
include_tags = tuple(tags) if tags else ()
include_deps = _as_tuple(dependencies)
include_auth = _ROUTER_AUTH_UNSET if auth is _UNSET else auth
for decl in router._effective_routes():
decl = decl.merged_under(prefix, include_tags, include_deps, include_auth)
self._guard_conflicting_route_meta(decl)
kwargs: dict[str, Any] = dict(decl.options)
if decl.tags:
kwargs["tags"] = list(decl.tags)
if decl.dependencies:
kwargs["dependencies"] = list(decl.dependencies)
if decl.auth is not _ROUTER_AUTH_UNSET:
kwargs["auth"] = decl.auth
self.route(decl.route, **kwargs)(decl.endpoint)
for hook_decl in router._effective_hooks():
hook_decl = hook_decl.merged_under(prefix)
hook = hook_decl.hook
if hook_decl.prefix:
hook = _prefix_scoped_hook(hook, hook_decl.prefix)
self.router.before_request(hook, websocket=hook_decl.websocket)
def _guard_conflicting_route_meta(self, decl: Any) -> None:
"""Refuse to silently rewrite a view's route metadata on re-inclusion.
Auth, dependencies, and OpenAPI tags are attached to the view function
itself, so replaying the same view with *different* effective values
would rewrite the earlier registration's behavior in place — whether
it came from a prior inclusion or a direct ``api.route()`` call
(e.g. an include with weaker ``auth=`` must not downgrade a route
that was registered directly with stronger auth).
"""
endpoint = decl.endpoint
effective_auth = (
self._auth if decl.auth is _ROUTER_AUTH_UNSET else _as_tuple(decl.auth)
)
effective = (effective_auth, decl.dependencies, decl.tags)
previous = getattr(endpoint, "_responder_router_meta", None)
if previous is None and self._endpoint_is_registered(endpoint):
# Registered directly via api.route(): reconstruct the metadata
# the view currently carries so this inclusion cannot rewrite it.
previous = (
tuple(getattr(endpoint, "_route_auth", ())),
tuple(getattr(endpoint, "_route_dependencies", ())),
tuple(getattr(endpoint, "_openapi_meta", {}).get("tags", ())),
)
if previous is not None and previous != effective:
name = getattr(endpoint, "__name__", repr(endpoint))
raise ValueError(
f"Cannot include view {name!r} again with different "
"auth/tags/dependencies: route metadata is attached to the "
"view itself, so a second inclusion would silently rewrite "
"the first. Include it with identical settings, or use "
"separate view functions/routers."
)
try:
endpoint._responder_router_meta = effective
except (AttributeError, TypeError): # e.g. functools.partial
pass
def _endpoint_is_registered(self, endpoint: Any) -> bool:
"""Whether ``endpoint`` already backs a route on this API."""
return any(
getattr(route, "endpoint", None) is endpoint for route in self.router.routes
)
async def __call__(self, scope, receive, send):
# Expose the API on the scope up front so error handlers wrapping the
# whole stack (e.g. ServerErrorMiddleware) can read problem_details even
# for failures raised above the router, where scope["api"] is otherwise
# not yet set.
if scope["type"] in ("http", "websocket"):
scope["api"] = self
await self.app(scope, receive, send)
[docs]
class RouteGroup:
"""A group of routes with a shared URL prefix.
Before-request hooks registered on a group only run for requests whose
path falls under the group's prefix.
"""
def __init__(self, api, prefix):
self.api = api
self.prefix = prefix.rstrip("/")
def route(self, route=None, **options):
# A missing path must surface as the usual "a route path is required"
# error, not silently register a literal "/prefixNone" path.
full_route = f"{self.prefix}{route}" if route is not None else None
return self.api.route(full_route, **options)
def get(self, route=None, **options):
return self.route(route, methods=["GET"], **options)
def sse(self, route=None, **options):
full_route = f"{self.prefix}{route}" if route is not None else None
return self.api.sse(full_route, **options)
def ndjson(self, route=None, **options):
full_route = f"{self.prefix}{route}" if route is not None else None
return self.api.ndjson(full_route, **options)
def post(self, route=None, **options):
return self.route(route, methods=["POST"], **options)
def put(self, route=None, **options):
return self.route(route, methods=["PUT"], **options)
def patch(self, route=None, **options):
return self.route(route, methods=["PATCH"], **options)
def delete(self, route=None, **options):
return self.route(route, methods=["DELETE"], **options)
def head(self, route=None, **options):
return self.route(route, methods=["HEAD"], **options)
def options(self, route=None, **options):
return self.route(route, methods=["OPTIONS"], **options)
def websocket_route(self, route=None, **options):
return self.route(route, websocket=True, **options)
def _path_in_group(self, path):
return path == self.prefix or path.startswith(self.prefix + "/")
[docs]
def before_request(self, websocket=False):
"""Register a hook that runs before requests under this group's prefix."""
def decorator(f):
if inspect.iscoroutinefunction(f):
async def hook(target, *rest):
if self._path_in_group(target.url.path):
await f(target, *rest)
else:
def hook(target, *rest):
if self._path_in_group(target.url.path):
f(target, *rest)
functools.wraps(f)(hook)
self.api.router.before_request(hook, websocket=websocket)
return f
return decorator
