HTTP

• Start here: HTTP
• The errors: Errors
• The own-port server: Ops Server

Register ErrorResponses() and micro.install(app) wires the handler, so a rejection raised in a route handler answers the client as an application/problem+json body. Import ProblemDetail to return one of your own, and send_error to write one from a pure-ASGI middleware.

Register IdempotentRequests() and install adds IdempotencyMiddleware, which replays a stored response when a request repeats its Idempotency-Key.

Register ConditionalRequests() and install adds the entity tags, so check_precondition(etag_of(version)) refuses a write whose If-Match moved on. Read Conditional Requests.

Register CachedResponses(), declare CachedResponse(ttl=...) on a route, and a repeated read is answered from the cache instead of the handler. Read Response Cache.

Register RateLimitedRequests(limiter, trusted=...) and a caller over its budget is answered 429 at the edge, with the quota it has left. Read Rate Limit.

Register OpsServer() on a process that serves no HTTP, and the health probes and the Prometheus endpoint answer on a port of their own.

grelmicro.http

HTTP.

Puts grelmicro's rejections on the wire. AdmissionError already covers every "turned away" case, and ProblemDetail renders it as the application/problem+json body of RFC 9457, with the field the client needs next: how long to wait, or that there is nothing to wait for.

Register ErrorResponses() to opt in, and micro.install(app) wires the handler, so a rate limiter, a bulkhead, an open circuit breaker, or an elapsed deadline answers the client without a single except in a route handler. Without it grelmicro changes nothing about how your framework answers an error.

The format comes from the factory you call, never from a variable. RFC 9457 problem details are the default, and ErrorResponses.tmf() renders the TM Forum format for a service answering to a TM Forum Open API platform.

IdempotencyMiddleware replays a stored response when a request repeats its idempotency key, and IdempotentRequests() registers it so micro.install(app) adds it.

ConditionalRequests() answers conditional requests: it puts an ETag on responses, answers 304 to a read that already holds one, and lets check_precondition(etag_of(version)) refuse a write whose If-Match no longer matches.

Read more in the Error Responses and Idempotency Middleware docs.

PROBLEM_MEDIA_TYPE module-attribute

PROBLEM_MEDIA_TYPE = 'application/problem+json'

Media type every problem detail is served with, from RFC 9457.

ERROR_DOCS_BASE module-attribute

ERROR_DOCS_BASE = DOCS_BASE

Where a type URI points, one anchor per rejection.

OpsServer

OpsServer(
    *,
    port: int | None = None,
    host: str | None = None,
    show_details: bool | None = None,
    request_timeout: float | None = None,
    shutdown_timeout: float | None = None,
    max_connections: int | None = None,
    name: str = "default",
    env_prefix: str | None = None,
    env_load: bool | None = None,
)

Serve the health and metrics endpoints on a port of their own.

For a process that runs no web framework: a FastStream consumer, a scheduler, a worker. Kubernetes still needs somewhere to send its probes, and Prometheus still needs somewhere to scrape.

from grelmicro import Grelmicro
from grelmicro.health import HealthChecks
from grelmicro.http import OpsServer
from grelmicro.metrics import Metrics

micro = Grelmicro(
    uses=[
        HealthChecks(auto_health=True),
        Metrics(exporter="prometheus"),
        OpsServer(port=8080),
    ]
)

It serves what the app registers: the three health endpoints when a HealthChecks is registered, and /metrics when a Metrics is. An app that registers neither has nothing to serve, so the server says so instead of listening.

It speaks HTTP/1.1 over the standard library, answers one request per connection, and serves nothing but these endpoints. It speaks no TLS and reads no request body, so give it the pod network rather than an ingress.

Register it first in uses=[...] and it closes last, so the probes keep answering while the rest of the app drains.

Read more in the Ops Server docs.

Initialize the ops server.

PARAMETER	DESCRIPTION
port	Port to listen on. Default: 8080. When unset and env reads are enabled (see env_load and GREL_ENV_LOAD), resolves from the environment variable GREL_OPS_PORT (or GREL_OPS_{NAME_UPPER}_PORT for a named instance) if present, otherwise falls back to the OpsServerConfig default. TYPE: int | None DEFAULT: None
host	Address to bind. Empty binds every interface, IPv4 and IPv6. Default: empty. Resolves from GREL_OPS_HOST the way port does. TYPE: str | None DEFAULT: None
show_details	Whether /healthz includes each check's verbose details field. Default: False. Resolves from GREL_OPS_SHOW_DETAILS the way port does. TYPE: bool | None DEFAULT: None
request_timeout	Seconds one request may take, from the first byte read to the last byte written. Default: 10.0. Resolves from GREL_OPS_REQUEST_TIMEOUT the way port does. TYPE: float | None DEFAULT: None
shutdown_timeout	Seconds in-flight requests get to finish on shutdown. Default: 5.0. Resolves from GREL_OPS_SHUTDOWN_TIMEOUT the way port does. TYPE: float | None DEFAULT: None
max_connections	Connections served at once. Default: 32. Resolves from GREL_OPS_MAX_CONNECTIONS the way port does. TYPE: int | None DEFAULT: None
name	Registration name. Two OpsServer instances may coexist on one Grelmicro under different names, on different ports. TYPE: str DEFAULT: 'default'
env_prefix	Override the auto-derived environment variable prefix. Default: GREL_OPS_ for the default instance, GREL_OPS_{NAME_UPPER}_ for a named one. TYPE: str | None DEFAULT: None
env_load	Whether to read environment variables. When None (the default), follow the process-wide GREL_ENV_LOAD flag. Pass True or False to override the flag for this construction. TYPE: bool | None DEFAULT: None

kind class-attribute

kind: str = 'ops'

name property

name: str

Return the registration name.

config property

config: OpsServerConfig

Return the resolved configuration.

paths property

paths: tuple[str, ...]

Return the paths served.

Empty until the app it belongs to has finished opening, because until then it answers /livez alone and what it will serve is not settled: a Metrics still opening has no exporter to read.

from_config classmethod

from_config(
    config: OpsServerConfig, *, name: str = "default"
) -> Self

Construct an OpsServer from a pre-built OpsServerConfig.

PARAMETER	DESCRIPTION
config	The pre-built ops server configuration. Use this path when the configuration is assembled at startup from a settings tree (for example YAML, Vault, or a pydantic-settings aggregator). The environment path is bypassed and the config is used as-is. TYPE: OpsServerConfig
name	Registration name. Defaults to 'default'. TYPE: str DEFAULT: 'default'

OpsServerConfig

Bases: BaseModel

Ops Server Config.

host class-attribute instance-attribute

host: str = ''

Address to bind. Empty binds every interface, IPv4 and IPv6, which is what the kubelet needs to reach the pod. Set 127.0.0.1 to keep the port on loopback.

port class-attribute instance-attribute

port: int = 8080

Port to listen on.

show_details class-attribute instance-attribute

show_details: bool = False

Whether /healthz includes each check's verbose details field. False strips them.

request_timeout class-attribute instance-attribute

request_timeout: PositiveFloat = 10.0

Seconds one request may take, from the first byte read to the last byte written. A connection that goes quiet is dropped.

shutdown_timeout class-attribute instance-attribute

shutdown_timeout: NonNegativeFloat = 5.0

Seconds in-flight requests get to finish on shutdown before they are cancelled.

max_connections class-attribute instance-attribute

max_connections: PositiveInt = 32

Connections served at once. Beyond it a connection is answered 503 and closed.

OpsServerError

Bases: GrelmicroError

Raised when the ops server cannot serve.

The app registers nothing the server can answer, no Grelmicro app is active, or the port is taken.

ErrorResponses

ErrorResponses(*, name: str = 'default')

Answer every grelmicro rejection in a standard error format.

Register it to opt in, and micro.install(app) wires the handler into FastAPI, Starlette, or Litestar. Without it grelmicro registers nothing, and a rejection reaches the framework's own error handling exactly as any other exception does.

A rate limiter over budget, a full bulkhead, an open circuit breaker, a lock held elsewhere, or an elapsed deadline then answers the client with a body carrying what it can act on, instead of becoming a 500.

The bare constructor renders RFC 9457 problem details, which is the format an HTTP API is expected to speak. ErrorResponses.tmf() renders the TM Forum format instead:

from fastapi import FastAPI

from grelmicro import Grelmicro
from grelmicro.http import ErrorResponses

micro = Grelmicro(uses=[ErrorResponses()])
app = FastAPI()
micro.install(app)

A format is chosen by the factory you call, never by a variable, because it is the shape of the response and not a value to tune. One rendering answers for the whole app, so registering two is refused.

A framework that serves no HTTP, such as FastStream, ignores it.

Read more in the Error Responses docs.

Render rejections as RFC 9457 problem details.

PARAMETER	DESCRIPTION
name	Registration name. Only one may be registered. TYPE: str DEFAULT: 'default'

kind class-attribute

kind: str = 'error_responses'

singleton class-attribute

singleton: bool = True

singleton_reason class-attribute

singleton_reason: str = "One rendering answers for the whole app, so two formats cannot both be registered"

name property

name: str

Return the registration name.

media_type property

media_type: str

Return the content type every rendered error is served with.

model property

model: type[BaseModel]

Return the body model, for publishing the shape in OpenAPI.

document_idempotency(app) reads it, so a schema describes the format the app actually answers in rather than assuming one.

tmf classmethod

tmf(
    *,
    code_prefix: str = DEFAULT_CODE_PREFIX,
    reference_error: str | None = DOCS_BASE,
    name: str = "default",
) -> Self

Render rejections in the TM Forum error format of TMF630.

For a service that answers to a telco or OSS platform built on TM Forum Open APIs, where the client expects code and reason rather than type and title.

The status codes are the same ones RFC 9457 mode returns. TMF630 mandates the IANA registry and names 422, 429 and 503 itself, so nothing is remapped and Retry-After keeps its meaning.

micro = Grelmicro(uses=[ErrorResponses.tmf(code_prefix="SBB")])

Two members do not survive the format. There is no equivalent of instance, and TMF630 defines no extension mechanism, so a retry_after reaches the client only as the Retry-After header.

reference_error=None leaves the documentation URI out, for a service whose responses must name no address outside it.

PARAMETER	DESCRIPTION
code_prefix	Namespace for the code member, which TMF630 makes mandatory and leaves to the API. An application writes its own business codes into the same field, so the prefix says which system defined this one. Set it to fold grelmicro's codes into an operator's catalogue. TYPE: str DEFAULT: DEFAULT_CODE_PREFIX
reference_error	Base the referenceError documentation URI is built on. Pass None to leave the member out, for a service whose responses must name no address outside it. Pass your own base to point at your documentation instead of grelmicro's. TYPE: str | None DEFAULT: DOCS_BASE
name	Registration name. Only one may be registered. TYPE: str DEFAULT: 'default'

problem_details classmethod

problem_details(*, name: str = 'default') -> Self

Render rejections as RFC 9457 problem details.

The same as the bare constructor, for a wiring that says which format it speaks rather than leaving it to the default.

PARAMETER	DESCRIPTION
name	Registration name. Only one may be registered. TYPE: str DEFAULT: 'default'

render_status

render_status(
    status: int,
    *,
    detail: str | None = None,
    instance: str | None = None,
    extensions: dict[str, Any] | None = None,
) -> RenderedError

Return the response for an error the framework raised.

The rejections grelmicro raises go through render, which knows what each one is. This renders one it does not know, so an app's own HTTPException and its request validation failures answer in the same format as everything else.

The status and the message stay as the framework set them. Only the shape changes.

PARAMETER	DESCRIPTION
status	HTTP status code of the response. TYPE: int
detail	Explanation of this occurrence, safe to show a client. TYPE: str | None DEFAULT: None
instance	Request path recorded as the occurrence. TYPE: str | None DEFAULT: None
extensions	Extra members, such as the field errors of a validation failure. TYPE: dict[str, Any] | None DEFAULT: None

render_validation

render_validation(
    field_errors: list[dict[str, Any]],
    *,
    status: int,
    detail: str | None = None,
    instance: str | None = None,
) -> RenderedError

Return the response for a request that failed validation.

A known kind, so a client branches on the identifier rather than on the status, and reads the same identifier whichever framework validated the request.

The status stays the framework's. FastAPI answers 422 and Litestar 400, and which is right is a question those projects have already answered for their users. 422 is the more precise code for a request that is well formed but semantically wrong, and RFC 9110 section 15.5.21 now defines it, but grelmicro reshapes an answer rather than overruling it.

PARAMETER	DESCRIPTION
field_errors	One entry per part of the request that did not match. TYPE: list[dict[str, Any]]
status	Status the framework chose. Kept, not second-guessed. TYPE: int
detail	What the framework said, when it said anything useful. TYPE: str | None DEFAULT: None
instance	Request path recorded as the occurrence. TYPE: str | None DEFAULT: None

render

render(
    exc: BaseException, *, instance: str | None = None
) -> RenderedError | None

Return the response for exc, or None when grelmicro has none.

An error grelmicro did not raise to turn a caller away returns None, so the framework answers it as it always did.

PARAMETER	DESCRIPTION
exc	The rejection to render. TYPE: BaseException
instance	Request path recorded as the occurrence. TYPE: str | None DEFAULT: None

IdempotentRequests

IdempotentRequests(
    *,
    ttl: float | None = None,
    namespace: str = "http",
    cache: TTLCache[Any] | None = None,
    key_header: str | None = None,
    replay_header: str | None = None,
    methods: Collection[str] | None = None,
    key_maker: Callable[[Scope, str], str] | None = None,
    skip: Callable[[StoredResponse], bool] | None = None,
    require_key: bool | None = None,
    fingerprint_body: bool | None = None,
    max_body_size: int | None = None,
    wait_timeout: float | None = None,
    include: tuple[str, ...] | None = None,
    exclude: tuple[str, ...] | None = None,
    reused_status: int | None = None,
    openapi: bool = True,
    name: str = "default",
    env_prefix: str | None = None,
    env_load: bool | None = None,
)

Bases: Reconfigurable[IdempotentRequestsConfig]

Replay repeated requests, wired by micro.install(app).

Register it and install adds IdempotencyMiddleware to the app and describes it in the OpenAPI schema, so the container holds the whole wiring:

from fastapi import FastAPI

from grelmicro import Grelmicro
from grelmicro.cache import Cache
from grelmicro.http import ErrorResponses, IdempotentRequests
from grelmicro.providers.redis import RedisProvider

redis = RedisProvider("redis://localhost:6379/0")
micro = Grelmicro(uses=[Cache(redis), ErrorResponses(), IdempotentRequests()])
app = FastAPI()
micro.install(app)

The bare form stores responses under Idempotency("http"), which keeps them for a day and rides the registered Cache. Pass an Idempotency of your own to set the lifetime, the namespace, or the cache it uses.

Every option of IdempotencyMiddleware is taken here and forwarded, so a registered component and a hand-added middleware answer the same.

A framework that serves no HTTP, such as FastStream, ignores it.

Read more in the Idempotency Middleware docs.

Replay repeated requests through the registered cache.

PARAMETER	DESCRIPTION
ttl	Seconds a stored response replays for. Defaults to a day. Held by the Idempotency this rides, which is named after the namespace, so it is tuned live under GREL_IDEMPOTENCY_HTTP_TTL. TYPE: float | None DEFAULT: None
namespace	Namespace the stored keys sit under, so two sets of rules on one app never read each other's responses. Part of every stored key, so it is not live. TYPE: str DEFAULT: 'http'
cache	The TTLCache responses are stored in. Defaults to the registered Cache component. TYPE: TTLCache[Any] | None DEFAULT: None
key_header	Request header carrying the idempotency key. TYPE: str | None DEFAULT: None
replay_header	Response header marking a replayed response. No standard names one, so pick what the clients already read. TYPE: str | None DEFAULT: None
methods	Methods that take an idempotency key. Every other method passes through. TYPE: Collection[str] | None DEFAULT: None
key_maker	Build the stored key from the ASGI scope and the client key. Set this in any multi-tenant app, folding in the caller identity. TYPE: Callable[[Scope, str], str] | None DEFAULT: None
skip	Predicate receiving the response. Return True to not store it. TYPE: Callable[[StoredResponse], bool] | None DEFAULT: None
require_key	Answer 400 when a method in methods arrives without the header, instead of passing it through. TYPE: bool | None DEFAULT: None
fingerprint_body	Hash the request body and store the hash with the response, so a key reused with a different body gets 422 instead of a wrong replay. TYPE: bool | None DEFAULT: None
max_body_size	Largest body held in memory, in bytes. TYPE: int | None DEFAULT: None
wait_timeout	Seconds a duplicate waits for an execution already in flight, before it is answered with 409. TYPE: float | None DEFAULT: None
include	Paths this middleware acts on. Empty means every path. Name the prefix of a router to select it, as "/payments/*". TYPE: tuple[str, ...] | None DEFAULT: None
exclude	Paths this middleware leaves alone, whatever include says. Same matching. TYPE: tuple[str, ...] | None DEFAULT: None
reused_status	Status answering a key reused with a different payload. 422 is what the Idempotency-Key header draft asks for, and 400 is what some APIs answer instead. TYPE: int | None DEFAULT: None
openapi	Describe both headers and the responses the middleware returns in the OpenAPI schema. Only FastAPI builds one. Read once when the schema is built, so it is not live. TYPE: bool DEFAULT: True
name	Registration name, for a second set of rules on one app. TYPE: str DEFAULT: 'default'
env_prefix	Override the derived prefix, GREL_IDEMPOTENT_REQUESTS_ for the default instance. TYPE: str | None DEFAULT: None
env_load	Whether to read environment variables. None follows the process-wide GREL_ENV_LOAD flag. TYPE: bool | None DEFAULT: None

kind class-attribute

kind: str = 'idempotent_requests'

name property

name: str

Return the registration name.

idempotency property

idempotency: Idempotency[Any]

Return the Idempotency the middleware stores through.

For code that has to reach the store itself, such as clearing a key an operator asked about. Handlers need none of it: the middleware does the storing.

from_config classmethod

from_config(
    config: IdempotentRequestsConfig,
    *,
    name: str = "default",
    namespace: str = "http",
    ttl: float | None = None,
    cache: TTLCache[Any] | None = None,
    key_maker: Callable[[Scope, str], str] | None = None,
    skip: Callable[[StoredResponse], bool] | None = None,
    openapi: bool = True,
) -> IdempotentRequests

Build the component from a configuration that is already whole.

The one declarative door. What you pass is what runs: no environment variable is read, and the instance is not registered for live reload. The store, the key maker and the skip predicate stay here rather than in the config, because they are objects and callables rather than values.

PARAMETER	DESCRIPTION
config	The pre-built idempotent requests configuration. TYPE: IdempotentRequestsConfig
name	Registration name, for a second set of rules on one app. TYPE: str DEFAULT: 'default'
namespace	Namespace the stored keys sit under. TYPE: str DEFAULT: 'http'
ttl	Seconds a stored response replays for. TYPE: float | None DEFAULT: None
cache	The TTLCache responses are stored in. TYPE: TTLCache[Any] | None DEFAULT: None
key_maker	Build the stored key from the scope and the client key. TYPE: Callable[[Scope, str], str] | None DEFAULT: None
skip	Return True to leave one response unstored. TYPE: Callable[[StoredResponse], bool] | None DEFAULT: None
openapi	Describe the headers in the OpenAPI schema. TYPE: bool DEFAULT: True

refresh_routes

refresh_routes(app: Any) -> None

Refresh the same route gates the runtime middleware enforces.

PARAMETER	DESCRIPTION
app	The application whose routes are reported. TYPE: Any

route_is_gated

route_is_gated(method: str, path: str) -> bool

Return whether a default key bypasses this route at runtime.

asgi_middleware

asgi_middleware() -> tuple[type[Any], dict[str, Any]]

Return the middleware class and the arguments to build it with.

micro.install(app) reads this from every registered component that carries it and hands the pair to the integration, which adds the middleware the way its framework takes one. A component without it wires no middleware.

handled_exceptions

handled_exceptions() -> tuple[type[Exception], ...]

Return what this component answers rather than letting through.

The middleware answers what it refuses itself. These are what the block form raises inside a handler, and registering the component is the opt-in for answering those the same way.

document_openapi

document_openapi(app: Any) -> None

Describe the middleware in the app's OpenAPI schema.

Called by the FastAPI integration after the middleware is added. A framework that builds no schema never calls it.

PARAMETER	DESCRIPTION
app	The FastAPI application to describe. TYPE: Any

IdempotencyMiddleware

IdempotencyMiddleware(
    app: ASGIApp,
    *,
    idempotency: Idempotency[Any],
    key_header: str = "Idempotency-Key",
    replay_header: str = _DEFAULT_REPLAY_HEADER,
    methods: Collection[str] = ("POST",),
    key_maker: Callable[[Scope, str], str] | None = None,
    skip: Callable[[StoredResponse], bool] | None = None,
    require_key: bool = False,
    fingerprint_body: bool = False,
    max_body_size: int = 1024 * 1024,
    wait_timeout: float = 10.0,
    include: tuple[str, ...] = (),
    exclude: tuple[str, ...] = (),
    reused_status: int = status,
    live: Live[_State] | None = None,
)

Replay a stored HTTP response when a request repeats its idempotency key.

A request whose method is listed in methods and which carries the key_header runs once. A retry with the same key replays the stored status, headers, and body without reaching the handler, and carries the replay_header marker, Idempotent-Replayed: true by default. A request without the key_header passes straight through, so adding the middleware changes nothing until a client opts in.

from fastapi import FastAPI

from grelmicro import Grelmicro
from grelmicro.http import IdempotencyMiddleware
from grelmicro.cache import TTLCache
from grelmicro.idempotency import Idempotency

micro = Grelmicro(uses=[...])
app = FastAPI()
micro.install(app)

app.add_middleware(
    IdempotencyMiddleware, idempotency=Idempotency("http", ttl=3600)
)

Register IdempotentRequests() instead to have micro.install(app) add it for you, along with the OpenAPI documentation.

Added by hand, it goes before or after micro.install(app). It resolves its Cache through the grelmicro request scope, which install keeps outside every other middleware.

A duplicate that arrives while the first execution is in flight waits for it and replays its response. The wait folds across replicas when a Coordination lock backend is configured, and in-process otherwise. It is bounded by wait_timeout.

Every response the app returns is stored, errors included. A handler that raises an unhandled exception stores nothing, so the framework's 500 never replays.

Without a custom key_maker, a request carrying Authorization or Cookie bypasses idempotency and runs the app. Its response cannot be stored under a key shared across callers, and a replay cannot skip authentication inside the app. Configure an identity-aware key_maker to make authenticated requests idempotent.

Four kinds of response are not stored, and each one lets a retry re-run the handler: one carrying Set-Cookie, one carrying Content-Encoding, one declaring trailers, and one whose body is over max_body_size. All four are logged. Pass skip to add a rule of your own.

Background tasks run after the response is sent, so a replay can be served while the original request's background work is still in flight.

The middleware is pure ASGI and works with any ASGI framework (Starlette, Litestar, ...). It acts on http scopes and passes every other scope through untouched.

Initialize the middleware with the idempotency store and policy.

PARAMETER	DESCRIPTION
app	The next ASGI application in the middleware chain. TYPE: ASGIApp
idempotency	The Idempotency that stores responses. Its ttl sets how long a key replays. TYPE: Idempotency[Any]
key_header	Request header carrying the idempotency key. TYPE: str DEFAULT: 'Idempotency-Key'
replay_header	Response header marking a replayed response. No standard names one, so pick what the clients already read. Idempotent-Replayed is what most APIs answer with. TYPE: str DEFAULT: _DEFAULT_REPLAY_HEADER
methods	Methods that take an idempotency key. Every other method passes through. TYPE: Collection[str] DEFAULT: ('POST',)
key_maker	Build the stored key from the ASGI scope and the client key. Defaults to the authority, scheme, root path, method, route path, query string, and client key, so two public resources never replay each other. Requests carrying Authorization or Cookie bypass that unscoped default. Set this to an identity-aware key in a multi-tenant app that needs authenticated replay. TYPE: Callable[[Scope, str], str] | None DEFAULT: None
skip	Predicate receiving the response. Return True to not store it. Mirrors skip on @cached. Use it for a response that is technically replayable but should not be, such as one whose body embeds a timestamp the caller must not see twice. Responses that are never safe to replay are dropped before this runs. TYPE: Callable[[StoredResponse], bool] | None DEFAULT: None
require_key	Answer 400 when a method in methods arrives without the header, instead of passing it through. TYPE: bool DEFAULT: False
fingerprint_body	Hash the request body and store the hash with the response. A key reused with a different body then gets 422 instead of a wrong replay. Buffers the request body before the handler runs, and answers 413 when it is over max_body_size. TYPE: bool DEFAULT: False
max_body_size	Largest body held in memory, in bytes. A larger response is sent to the client and not stored. With fingerprint_body, a larger request body is answered with 413. TYPE: int DEFAULT: 1024 * 1024
wait_timeout	Seconds a duplicate waits for an execution already in flight. Past it the duplicate is answered with 409 and a Retry-After header. TYPE: float DEFAULT: 10.0
include	Paths this middleware acts on. Empty means every path. Exact match unless the pattern ends with *, which matches as a prefix, so a router mounted under /payments is "/payments/*". TYPE: tuple[str, ...] DEFAULT: ()
exclude	Paths this middleware leaves alone, whatever include says. Same matching. TYPE: tuple[str, ...] DEFAULT: ()
reused_status	Status answering a key reused with a different payload. 422 is what the Idempotency-Key header draft asks for. Pass 400 where the clients expect that instead. The body is the same either way, so a client reading the type identifier is unaffected. TYPE: int DEFAULT: status
live	The cell a registered IdempotentRequests publishes its snapshot into, filled by micro.install(app). Passing it makes the other options the component's to decide. TYPE: Live[_State] | None DEFAULT: None

RAISES	DESCRIPTION
TypeError	If methods is given as a string. tuple("POST") is four one-letter methods, none of which a request carries, so it would meter nothing at all.

app instance-attribute

app = app

StoredResponse

Bases: TypedDict

The response IdempotencyMiddleware is about to store.

Handed to skip so a handler's own rule decides whether a response replays. headers maps lowercased names to their value, keeping the last of a repeated name.

status instance-attribute

status: int

headers instance-attribute

headers: dict[str, str]

body instance-attribute

body: bytes

ConditionalRequests

ConditionalRequests(
    *,
    etag_responses: bool | None = None,
    require_precondition: Sequence[str] | None = None,
    include: tuple[str, ...] | None = None,
    exclude: tuple[str, ...] | None = None,
    max_body_size: int | None = None,
    openapi: bool = True,
    name: str = "default",
    env_prefix: str | None = None,
    env_load: bool | None = None,
)

Bases: Reconfigurable[ConditionalRequestsConfig]

Answer conditional requests, wired by micro.install(app).

Register it and install adds ConditionalRequestsMiddleware, so every request's preconditions are bound for check_precondition(...) and every response carries an ETag:

from fastapi import FastAPI

from grelmicro import Grelmicro
from grelmicro.http import ConditionalRequests, ErrorResponses

micro = Grelmicro(uses=[ErrorResponses(), ConditionalRequests()])
app = FastAPI()
micro.install(app)

Every option of ConditionalRequestsMiddleware is taken here and forwarded, so a registered component and a hand-added middleware answer the same.

A framework that serves no HTTP, such as FastStream, ignores it.

Read more in the Conditional Requests docs.

Answer conditional requests through the registered middleware.

PARAMETER	DESCRIPTION
etag_responses	Add an ETag to a 2xx response that carries a complete body and none of its own. TYPE: bool | None DEFAULT: None
require_precondition	Methods answered 428 when they carry no precondition. ("PUT", "PATCH", "DELETE") refuses every unconditional write in the app. POST is left out of that set on purpose: a create has nothing to match against yet. Empty by default, which leaves the decision to check_precondition() per route. TYPE: Sequence[str] | None DEFAULT: None
include	Paths this middleware acts on. Empty means every path. Exact match unless the pattern ends with *, which matches as a prefix, so a router mounted under /payments is "/payments/*". TYPE: tuple[str, ...] | None DEFAULT: None
exclude	Paths this middleware leaves alone, whatever include says. Same matching. TYPE: tuple[str, ...] | None DEFAULT: None
max_body_size	Largest response body held in memory to hash, in bytes. A larger one is forwarded untouched. TYPE: int | None DEFAULT: None
openapi	Describe If-Match, If-None-Match and the responses they lead to in the OpenAPI schema, so a client built from it sends the headers and Swagger offers the fields. Only FastAPI builds one, and every other framework ignores this. Read once when the schema is built, so it is not live. TYPE: bool DEFAULT: True
name	Registration name, for a second set of rules on one app. TYPE: str DEFAULT: 'default'
env_prefix	Override the derived prefix, GREL_CONDITIONAL_REQUESTS_ for the default instance. TYPE: str | None DEFAULT: None
env_load	Whether to read environment variables. None follows the process-wide GREL_ENV_LOAD flag. TYPE: bool | None DEFAULT: None

kind class-attribute

kind: str = 'conditional_requests'

name property

name: str

Return the registration name.

from_config classmethod

from_config(
    config: ConditionalRequestsConfig,
    *,
    name: str = "default",
    openapi: bool = True,
) -> ConditionalRequests

Build the component from a configuration that is already whole.

The one declarative door. What you pass is what runs: no environment variable is read, and the instance is not registered for live reload.

PARAMETER	DESCRIPTION
config	The pre-built conditional requests configuration. TYPE: ConditionalRequestsConfig
name	Registration name, for a second set of rules on one app. TYPE: str DEFAULT: 'default'
openapi	Describe the headers in the OpenAPI schema. TYPE: bool DEFAULT: True

asgi_middleware

asgi_middleware() -> tuple[type[Any], dict[str, Any]]

Return the middleware class and the arguments to build it with.

The middleware is handed the cell rather than the values, so a live reconfigure reaches it without the stack being rebuilt, which a framework will not do once it is serving.

document_openapi

document_openapi(app: Any) -> None

Describe the conditional headers in the app's OpenAPI schema.

Called by the FastAPI integration after the middleware is added. A framework that builds no schema never calls it.

PARAMETER	DESCRIPTION
app	The FastAPI application to describe. TYPE: Any

handled_exceptions

handled_exceptions() -> tuple[type[Exception], ...]

Return what this component answers rather than letting through.

Registering it is the opt-in: a service that asked for conditional requests gets 412 and 428 on the wire, not a 500, whether or not an ErrorResponses is registered. That component chooses the format, this one decides these two are answered at all.

ConditionalRequestsMiddleware

ConditionalRequestsMiddleware(
    app: ASGIApp,
    *,
    etag_responses: bool = True,
    require_precondition: Sequence[str] = (),
    include: tuple[str, ...] = (),
    exclude: tuple[str, ...] = (),
    max_body_size: int = 1024 * 1024,
    live: Live[_State] | None = None,
)

Bind each request's preconditions and put an ETag on the response.

Three things, none of which a handler should have to write:

• Every If-Match and If-None-Match header is parsed and bound for the request, so check_precondition(...) reads them with no request object in the handler signature.
• A 2xx response that carries a complete body and no ETag of its own gets one, hashed from the body it just produced.
• A GET or HEAD whose If-None-Match matches that tag is answered 304 Not Modified with no body.
• A method named in require_precondition that carries neither precondition header is answered 428, before it reaches a handler.

from grelmicro import Grelmicro
from grelmicro.http import ConditionalRequestsMiddleware

app.add_middleware(ConditionalRequestsMiddleware)

Register ConditionalRequests() instead to have micro.install(app) add it for you.

The 304 saves the response, not the work: the handler has already run by the time the body is there to hash. A handler that knows its resource's version cheaply should compare it itself and skip the load.

A response over max_body_size, or one streamed in chunks past it, is forwarded as it comes and gets no entity tag, so a large download is never held in memory.

The middleware is pure ASGI and works with any ASGI framework (Starlette, Litestar, ...). It acts on http scopes and passes every other scope through untouched.

Initialize the middleware with its entity tag policy.

PARAMETER	DESCRIPTION
app	The next ASGI application in the middleware chain. TYPE: ASGIApp
etag_responses	Add an ETag to a 2xx response that carries a complete body and none of its own. TYPE: bool DEFAULT: True
require_precondition	Methods answered 428 when they carry no precondition. ("PUT", "PATCH", "DELETE") refuses every unconditional write in the app, before it reaches a handler. POST is left out of that set on purpose: a create has nothing to match against yet. Either header counts, so If-None-Match: * still creates under enforcement. Empty by default, which leaves the decision to check_precondition() per route. TYPE: Sequence[str] DEFAULT: ()
include	Paths this middleware acts on. Empty means every path. Exact match unless the pattern ends with *, which matches as a prefix, so a router mounted under /payments is "/payments/*". TYPE: tuple[str, ...] DEFAULT: ()
exclude	Paths this middleware leaves alone, whatever include says. Same matching. TYPE: tuple[str, ...] DEFAULT: ()
max_body_size	Largest response body held in memory to hash, in bytes. A larger one is forwarded untouched. TYPE: int DEFAULT: 1024 * 1024
live	The cell a registered ConditionalRequests publishes its snapshot into, filled by micro.install(app). Passing it makes the other options the component's to decide. TYPE: Live[_State] | None DEFAULT: None

RAISES	DESCRIPTION
TypeError	If require_precondition is given as a string. tuple("PUT") is three one-letter methods, none of which a request carries, so it would enforce nothing at all.

app instance-attribute

app = app

CachedResponses

CachedResponses(
    *,
    ttl: float | None = None,
    include: Mapping[str, float]
    | tuple[str, ...]
    | None = None,
    exclude: tuple[str, ...] | None = None,
    vary_by_headers: tuple[str, ...] | None = None,
    vary_by_query: tuple[str, ...] | _Unset | None = UNSET,
    key: Callable[[Scope], str | None] | None = None,
    skip: Callable[[StoredResponse], bool] | None = None,
    max_body_size: int | None = None,
    cache: TTLCache[Any] | None = None,
    namespace: str = "http",
    name: str = "default",
    env_prefix: str | None = None,
    env_load: bool | None = None,
)

Bases: Reconfigurable[CachedResponsesConfig]

Serve repeated reads from the cache, wired by micro.install(app).

Register it, mark the routes it answers for, and install adds CachedResponsesMiddleware to the app:

from fastapi import FastAPI

from grelmicro import Grelmicro
from grelmicro.cache import Cache
from grelmicro.http import CachedResponses
from grelmicro.integrations.fastapi import CachedResponse
from grelmicro.providers.redis import RedisProvider

redis = RedisProvider("redis://localhost:6379/0")
micro = Grelmicro(uses=[Cache(redis), CachedResponses()])
app = FastAPI()
micro.install(app)

@app.get("/products", dependencies=[CachedResponse(ttl=60)])
async def list_products() -> list[Product]: ...

The bare form caches nothing until a route declares it. include= names URLs instead, for a router whose routes you cannot touch and for a framework that resolves no dependencies grelmicro can read.

It rides the registered Cache, so a response one replica computed answers the callers of every other one. Pass a TTLCache of your own to store them somewhere else.

Register it before ConditionalRequests(), so a hit is answered without entering it.

Every option of CachedResponsesMiddleware is taken here and forwarded, so a registered component and a hand-added middleware answer the same.

A framework that serves no HTTP, such as FastStream, ignores it.

Read more in the Response Cache docs.

Answer repeated reads through the registered middleware.

PARAMETER	DESCRIPTION
ttl	Seconds a response is kept when its route names none. CachedResponse(ttl=...) overrides it per route. TYPE: float | None DEFAULT: None
include	The paths cached, for a route that declares none. A tuple keeps each for ttl, and a mapping gives each its own seconds, as {"/products/*": 60}. Exact match unless the pattern ends with *. TYPE: Mapping[str, float] | tuple[str, ...] | None DEFAULT: None
exclude	Paths never cached, whatever a route or include says. Same matching. TYPE: tuple[str, ...] | None DEFAULT: None
vary_by_headers	Request headers whose value is part of the key. A response whose Vary names a header outside this set is not stored, because one value would answer another. TYPE: tuple[str, ...] | None DEFAULT: None
vary_by_query	Query parameters that are part of the key. None (the default) keys on the whole query string, and passing it says so over any variable. TYPE: tuple[str, ...] | _Unset | None DEFAULT: UNSET
key	Builds the key from the ASGI scope, replacing the path and the vary rules. Return None to leave a request uncached. TYPE: Callable[[Scope], str | None] | None DEFAULT: None
skip	Returns whether one response is left unstored. TYPE: Callable[[StoredResponse], bool] | None DEFAULT: None
max_body_size	Largest response body stored, in bytes. TYPE: int | None DEFAULT: None
cache	The TTLCache responses are stored in. Defaults to one over the registered Cache. TYPE: TTLCache[Any] | None DEFAULT: None
namespace	Namespace the stored keys sit under, so two sets of rules on one app never read each other's responses. Part of every stored key, so it is not live. TYPE: str DEFAULT: 'http'
name	Registration name, for a second set of rules on one app. TYPE: str DEFAULT: 'default'
env_prefix	Override the derived prefix, GREL_CACHED_RESPONSES_ for the default instance. TYPE: str | None DEFAULT: None
env_load	Whether to read environment variables. None follows the process-wide GREL_ENV_LOAD flag. TYPE: bool | None DEFAULT: None

RAISES	DESCRIPTION
SettingsValidationError	If ttl, or one a pattern names, is not a positive number of seconds.

kind class-attribute

kind: str = 'cached_responses'

name property

name: str

Return the registration name.

cache property

cache: TTLCache[Any]

Return the TTLCache the responses are stored in.

from_config classmethod

from_config(
    config: CachedResponsesConfig,
    *,
    name: str = "default",
    namespace: str = "http",
    cache: TTLCache[Any] | None = None,
    key: Callable[[Scope], str | None] | None = None,
    skip: Callable[[StoredResponse], bool] | None = None,
) -> CachedResponses

Build the component from a configuration that is already whole.

The one declarative door. What you pass is what runs: no environment variable is read, and the instance is not registered for live reload. The store and the two callables stay here rather than in the config, because they are objects rather than values.

PARAMETER	DESCRIPTION
config	The pre-built cached responses configuration. TYPE: CachedResponsesConfig
name	Registration name, for a second set of rules on one app. TYPE: str DEFAULT: 'default'
namespace	Namespace the stored keys sit under. TYPE: str DEFAULT: 'http'
cache	The TTLCache responses are stored in. TYPE: TTLCache[Any] | None DEFAULT: None
key	Builds the key from the ASGI scope. TYPE: Callable[[Scope], str | None] | None DEFAULT: None
skip	Returns whether one response is left unstored. TYPE: Callable[[StoredResponse], bool] | None DEFAULT: None

purge async

purge() -> None

Delete every response this component stored.

What a write invalidates, so a handler that changed the resource drops what the cache would go on answering with until the TTL elapsed.

asgi_middleware

asgi_middleware() -> tuple[type[Any], dict[str, Any]]

Return the middleware class and the arguments to build it with.

The middleware is handed the cell rather than the values, so a live reconfigure reaches it without the stack being rebuilt, which a framework will not do once it is serving.

read_routes

read_routes(app: Any) -> None

Read CachedResponse() off every route the app declares.

Called by the integration after the middleware is added. The app is read again when it starts, so a route added between the two counts as well.

PARAMETER	DESCRIPTION
app	The application to read the rules off. TYPE: Any

CachedResponsesMiddleware

CachedResponsesMiddleware(
    app: ASGIApp,
    *,
    cache: TTLCache[Any],
    policies: _Policies | None = None,
    ttl: float = _DEFAULT_TTL,
    include: Mapping[str, float]
    | tuple[str, ...]
    | None = None,
    exclude: tuple[str, ...] = (),
    vary_by_headers: tuple[str, ...] = (),
    vary_by_query: tuple[str, ...] | None = None,
    key: Callable[[Scope], str | None] | None = None,
    skip: Callable[[StoredResponse], bool] | None = None,
    max_body_size: int = _DEFAULT_MAX_BODY_SIZE,
    tag: str = "grelmicro:http:default",
    live: Live[_State] | None = None,
)

Answer a repeated read from the cache instead of the handler.

A GET or HEAD whose path a rule names is looked up first. A hit is answered without reaching the app, carrying the Age it has spent in the cache, and a client that already holds the entity tag is answered 304 Not Modified with no body at all.

from grelmicro.cache import TTLCache
from grelmicro.http import CachedResponsesMiddleware

app.add_middleware(
    CachedResponsesMiddleware,
    cache=TTLCache(),
    include={"/products/*": 60},
)

Register CachedResponses() instead to have micro.install(app) add it for you, and to declare CachedResponse(ttl=...) on a route.

A miss runs the handler once. Every other request for the same key waits for that one and is answered from what it stored, in process and across replicas, so a cold key never fans one computation out to every caller at once.

A request asking for a range is passed through, because what is stored is the whole resource.

A response is stored only when it is safe to hand to somebody else: status 200, no Set-Cookie, no Content-Encoding, no Cache-Control refusing it, and a Vary naming nothing outside vary_by_headers. A request carrying Authorization or Cookie never reads the cache and never fills it. Neither does one an outer ASGI authentication middleware has already marked as authenticated.

A request's own Cache-Control is not read. This answers for the resource rather than for one caller, so a caller that could ask for the handler could spend it at will.

The middleware is pure ASGI and works with any ASGI framework (Starlette, Litestar, ...). It acts on http scopes and passes every other scope through untouched.

Initialize the middleware with the paths it answers for.

PARAMETER	DESCRIPTION
app	The next ASGI application in the middleware chain. TYPE: ASGIApp
cache	The TTLCache responses are stored in. TYPE: TTLCache[Any]
policies	The paths declaring CachedResponse(), filled by micro.install(app). include alone needs none. TYPE: _Policies | None DEFAULT: None
ttl	Seconds a response is kept when its route names none. TYPE: float DEFAULT: _DEFAULT_TTL
include	The paths cached. A tuple keeps each for ttl, and a mapping gives each its own seconds. Exact match unless the pattern ends with *. TYPE: Mapping[str, float] | tuple[str, ...] | None DEFAULT: None
exclude	Paths this middleware leaves alone, whatever else says. Same matching. TYPE: tuple[str, ...] DEFAULT: ()
vary_by_headers	Request headers whose value is part of the key, so one value never answers another. A response whose Vary names a header outside this set is not stored. TYPE: tuple[str, ...] DEFAULT: ()
vary_by_query	Query parameters that are part of the key. None (the default) keys on the whole query string. TYPE: tuple[str, ...] | None DEFAULT: None
key	Builds the key from the ASGI scope, replacing the path and the vary rules. Return None to leave a request uncached. TYPE: Callable[[Scope], str | None] | None DEFAULT: None
skip	Returns whether one response is left unstored. TYPE: Callable[[StoredResponse], bool] | None DEFAULT: None
max_body_size	Largest response body stored, in bytes. A larger one is streamed to the client and not kept. TYPE: int DEFAULT: _DEFAULT_MAX_BODY_SIZE
tag	Tag every entry carries, so purge() deletes them all. TYPE: str DEFAULT: 'grelmicro:http:default'
live	The cell a registered CachedResponses publishes its snapshot into, filled by micro.install(app). Passing it makes the other options the component's to decide. TYPE: Live[_State] | None DEFAULT: None

RAISES	DESCRIPTION
TypeError	If a set of path patterns was given as a string.

app instance-attribute

app = app

RateLimitedRequests

RateLimitedRequests(
    *limiters: RateLimiter,
    trusted: TrustedProxies | None = None,
    key: Callable[[Scope], str | None] | None = None,
    cost: int | None = None,
    max_wait: float | None = None,
    include: tuple[str, ...] | None = None,
    exclude: tuple[str, ...] | None = None,
    legacy_headers: bool | None = None,
    openapi: bool = True,
    name: str = "default",
    env_prefix: str | None = None,
    env_load: bool | None = None,
)

Bases: Reconfigurable[RateLimitedRequestsConfig]

Turn a caller away at the edge, wired by micro.install(app).

Register it and install adds RateLimitMiddleware to the app:

from grelmicro import Grelmicro
from grelmicro.http import ErrorResponses, RateLimitedRequests
from grelmicro.resilience import RateLimiter
from grelmicro.security import TrustedProxies

micro = Grelmicro(
    uses=[
        ErrorResponses(),
        RateLimitedRequests(
            RateLimiter.sliding_window("burst", limit=100, window=60),
            RateLimiter.sliding_window("daily", limit=10000, window=86400),
            trusted=TrustedProxies(["10.0.0.0/8"]),
        ),
    ]
)

A request spends one token of every limiter listed, so a burst limit stands beside a daily one and the response states both.

Every option of RateLimitMiddleware is taken here and forwarded, so a registered component and a hand-added middleware answer the same.

A framework that serves no HTTP, such as FastStream, ignores it.

Read more in the Rate Limit docs.

Meter every request through the registered middleware.

PARAMETER	DESCRIPTION
*limiters	The limiters every request spends, in the order given. TYPE: RateLimiter DEFAULT: ()
trusted	The proxies whose forwarded entries may be believed, for resolving the caller. TYPE: TrustedProxies | None DEFAULT: None
key	Builds the bucket key from the ASGI scope, replacing the resolved caller. Return None to leave a request unmetered. TYPE: Callable[[Scope], str | None] | None DEFAULT: None
cost	Tokens one request spends of every limiter. TYPE: int | None DEFAULT: None
max_wait	Seconds a throttled request waits before it is refused. 0.0 refuses as soon as the budget is spent. TYPE: float | None DEFAULT: None
include	Paths metered. Empty (the default) means every path. Same matching as every other middleware. TYPE: tuple[str, ...] | None DEFAULT: None
exclude	Paths never metered, whatever include says. TYPE: tuple[str, ...] | None DEFAULT: None
legacy_headers	Also send the superseded X-RateLimit-* fields. TYPE: bool | None DEFAULT: None
openapi	Describe the 429 and the RateLimit fields on every operation in the OpenAPI schema. Only FastAPI builds one. Read once when the schema is built, so it is not live. TYPE: bool DEFAULT: True
name	Registration name, for a second set of rules on one app. TYPE: str DEFAULT: 'default'
env_prefix	Override the derived prefix, GREL_RATE_LIMITED_REQUESTS_ for the default instance. TYPE: str | None DEFAULT: None
env_load	Whether to read environment variables. None follows the process-wide GREL_ENV_LOAD flag. TYPE: bool | None DEFAULT: None

RAISES	DESCRIPTION
TypeError	If no limiter is given, or neither trusted nor key says how to key the buckets.
ValueError	If a limiter is named something a RateLimit header cannot carry.

kind class-attribute

kind: str = 'rate_limited_requests'

name property

name: str

Return the registration name.

limiters property

limiters: tuple[RateLimiter, ...]

Return the limiters every request spends.

from_config classmethod

from_config(
    config: RateLimitedRequestsConfig,
    *limiters: RateLimiter,
    name: str = "default",
    trusted: TrustedProxies | None = None,
    key: Callable[[Scope], str | None] | None = None,
    openapi: bool = True,
) -> RateLimitedRequests

Build the component from a configuration that is already whole.

The one declarative door. What you pass is what runs: no environment variable is read, and the instance is not registered for live reload. The limiters stay here rather than in the config, because they are objects with buckets of their own, each tuned under its own GREL_RATELIMITER_ address.

PARAMETER	DESCRIPTION
config	The pre-built rate limited requests configuration. TYPE: RateLimitedRequestsConfig
*limiters	The limiters every request spends, in the order given. TYPE: RateLimiter DEFAULT: ()
name	Registration name, for a second set of rules on one app. TYPE: str DEFAULT: 'default'
trusted	The proxies whose forwarded entries may be believed. TYPE: TrustedProxies | None DEFAULT: None
key	Builds the bucket key from the ASGI scope. TYPE: Callable[[Scope], str | None] | None DEFAULT: None
openapi	Describe the 429 in the OpenAPI schema. TYPE: bool DEFAULT: True

asgi_middleware

asgi_middleware() -> tuple[type[Any], dict[str, Any]]

Return the middleware class and the arguments to build it with.

handled_exceptions

handled_exceptions() -> tuple[type[Exception], ...]

Return what this component answers rather than letting through.

The middleware answers what it refuses itself. This is what the limiter raises inside a handler, and registering the component is the opt-in for answering those the same way.

document_openapi

document_openapi(app: Any) -> None

Describe the 429 a metered app can answer with, in the schema.

Every operation, not the metered ones. Which paths are metered is tuned while the service runs, and the schema is built once, so naming the current set would publish a document that stops being true the first time an operator narrows it. A 429 says only what a client may be answered with, never what it must send, so stating it everywhere stays true whichever paths are metered.

Called by the FastAPI integration after the middleware is added. A framework that builds no schema never calls it.

PARAMETER	DESCRIPTION
app	The FastAPI application to describe. TYPE: Any

RateLimitMiddleware

RateLimitMiddleware(
    app: ASGIApp,
    *,
    limiters: Sequence[RateLimiter],
    trusted: TrustedProxies | None = None,
    key: Callable[[Scope], str | None] | None = None,
    cost: int = 1,
    max_wait: float = 0.0,
    include: tuple[str, ...] = (),
    exclude: tuple[str, ...] = (),
    legacy_headers: bool = False,
    live: Live[_State] | None = None,
)

Turn a caller away at the edge, and say what it has left.

Every request spends one token of every limiter it is given, keyed by the caller. A request the first of them refuses is answered 429 without reaching the app, through the same AdmissionError path a handler's own refusal takes, so one exception handler still covers every way a caller is turned away.

from grelmicro.http import RateLimitMiddleware
from grelmicro.resilience import RateLimiter
from grelmicro.security import TrustedProxies

app.add_middleware(
    RateLimitMiddleware,
    limiters=[RateLimiter.sliding_window("api", limit=100, window=60)],
    trusted=TrustedProxies(["10.0.0.0/8"]),
)

Register RateLimitedRequests(...) instead to have micro.install(app) add it for you.

Allowed or refused, the response carries what the caller has left, in the two fields of the RateLimit header specification: RateLimit with r= remaining and t= seconds to reset, and RateLimit-Policy with q= quota and w= window for every limiter that has a window. A refusal adds Retry-After.

The caller is the resolved client address, which is the socket peer unless a trusted proxy vouched for another one. A limiter keyed on the peer would meter the ingress rather than the caller, and one keyed on a raw X-Forwarded-For would meter whatever the caller wrote there.

The middleware is pure ASGI and works with any ASGI framework (Starlette, Litestar, ...). It acts on http scopes and passes every other scope through untouched.

Initialize the middleware with the limiters it spends.

PARAMETER	DESCRIPTION
app	The next ASGI application in the middleware chain. TYPE: ASGIApp
limiters	The limiters every request spends. A request passes all of them or is refused by the first that says no. TYPE: Sequence[RateLimiter]
trusted	The proxies whose forwarded entries may be believed, for resolving the caller. Give this or key: without either, the only bucket left is the socket peer, which behind an ingress is the ingress. An address ClientAddressMiddleware already resolved is reused, and this says which proxies to believe when it has not. TYPE: TrustedProxies | None DEFAULT: None
key	Builds the bucket key from the ASGI scope, replacing the resolved caller. Return None to let a request through unmetered. TYPE: Callable[[Scope], str | None] | None DEFAULT: None
cost	Tokens one request spends of every limiter. TYPE: int DEFAULT: 1
max_wait	Seconds a throttled request waits for tokens before it is refused. 0.0 (the default) refuses as soon as the budget is spent, because waiting at the edge holds the connection open. TYPE: float DEFAULT: 0.0
include	Paths this middleware meters. Empty (the default) means every path. Exact match unless the pattern ends with *, which matches as a prefix. TYPE: tuple[str, ...] DEFAULT: ()
exclude	Paths this middleware leaves alone, whatever include says. Same matching. TYPE: tuple[str, ...] DEFAULT: ()
legacy_headers	Also send X-RateLimit-Limit, -Remaining and -Reset, the superseded fields, for a client that reads only those. They carry the limiter closest to being spent. TYPE: bool DEFAULT: False
live	The cell a registered RateLimitedRequests publishes its snapshot into, filled by micro.install(app). Passing it makes the other options the component's to decide. TYPE: Live[_State] | None DEFAULT: None

RAISES	DESCRIPTION
TypeError	If no limiter is given, or the caller cannot be resolved because neither trusted nor key was given.
ValueError	If a limiter is named something a RateLimit header cannot carry, or cost is more than one of them can ever serve.

app instance-attribute

app = app

PreconditionError

Bases: GrelmicroError

Base error for a conditional request that could not proceed.

Catch it to handle both halves of optimistic concurrency with one except: a precondition that failed, and one the service required and the client did not send.

PreconditionFailedError

PreconditionFailedError()

Bases: PreconditionError

Raised when the client's precondition does not match the resource.

The If-Match header carried an entity tag that is not the one the resource has now, so another writer landed in between and the write would have overwritten their change. Answers 412.

Raise it yourself when the conditional write itself comes back empty, which is the case a check before the write cannot catch:

result = await session.execute(
    update(Cart)
    .where(Cart.id == cart_id, Cart.version == expected)
    .values(items=items, version=expected + 1)
)
if result.rowcount == 0:
    raise PreconditionFailedError

Initialize the error.

PreconditionRequiredError

PreconditionRequiredError()

Bases: PreconditionError

Raised when a write that must be conditional carried no precondition.

The service requires If-Match on this write, so a client that sends none is told to fetch the resource and come back with its entity tag, rather than being allowed to overwrite whatever is there. Answers 428.

Initialize the error.

RenderedError dataclass

RenderedError(
    status: int,
    media_type: str,
    headers: dict[str, str],
    body: bytes,
)

One rejection, ready to send, in whichever format was chosen.

What every integration needs and nothing more, so a framework wires the same three values whatever standard the body follows.

status instance-attribute

status: int

HTTP status code of the response.

media_type instance-attribute

media_type: str

Content type the body is written with.

headers instance-attribute

headers: dict[str, str]

Headers beside the content type, Retry-After included.

body instance-attribute

body: bytes

Serialized body.

ProblemDetail

Bases: BaseModel

An error response body, as defined by RFC 9457.

The five standard members are declared, and anything else passed is kept as an extension member, which is where the useful part of a problem detail lives:

from grelmicro.http import ProblemDetail

problem = ProblemDetail(
    type="https://example.com/problems/insufficient-funds",
    title="Insufficient funds",
    status=409,
    balance=30,
)

Declare it as a response model to publish the shape in OpenAPI:

@app.post("/charge", responses={429: {"model": ProblemDetail}})
async def charge() -> Charge: ...

Read more in the Error Responses docs.

type class-attribute instance-attribute

type: str = 'about:blank'

URI identifying the problem kind. Stable, so a client can branch on it without reading the prose.

title instance-attribute

title: str

Short summary of the problem kind, the same for every occurrence.

status instance-attribute

status: int

HTTP status code of the response.

detail class-attribute instance-attribute

detail: str | None = None

Explanation of this occurrence, safe to show a client.

instance class-attribute instance-attribute

instance: str | None = None

URI reference identifying the occurrence, the request path here.

TMFError

Bases: BaseModel

An error response body, as defined by TMF630.

Declare it as a response model to publish the shape in OpenAPI:

@app.post("/charge", responses={429: {"model": TMFError}})
async def charge() -> Charge: ...

Read more in the Error Responses docs.

model_config class-attribute instance-attribute

model_config = ConfigDict(
    extra="allow", populate_by_name=True
)

code instance-attribute

code: str

Application code for the error, namespaced by the prefix the component was built with.

reason instance-attribute

reason: str

Short summary of the error kind, safe to show a client user.

message class-attribute instance-attribute

message: str | None = None

Explanation of this occurrence and what to do about it.

reference_error class-attribute instance-attribute

reference_error: str | None = Field(
    default=None, alias="referenceError"
)

URI of the documentation describing this error kind.

type_ class-attribute instance-attribute

type_: str = Field(default='Error', alias='@type')

Class type of the representation, Error for this one.

check_precondition

check_precondition(
    version: object = _UNSET,
    *,
    etag: str | None = None,
    require: bool = True,
) -> None

Refuse a write whose precondition no longer holds.

Load the resource, hand over what identifies its version, and write only if this returns:

@app.put("/carts/{cart_id}")
async def replace(cart_id: str, body: CartIn) -> Cart:
    cart = await repo.load(cart_id)
    check_precondition(cart.version)
    return await repo.save(cart.apply(body))

The entity tag is built for you, the same way etag_of builds one, so a version column is all a handler has to hand over. Pass the whole resource where there is no version column, and etag= where the tag is already a tag.

This is a check, not a lock. Between it and the write, another request can land, so the write itself has to be conditional too. Read Conditional Requests for the three ways to do that, and which one to reach for.

The request's headers come from ConditionalRequests(), so nothing is threaded through the handler signature and the same line works on every framework.

PARAMETER	DESCRIPTION
version	What identifies the version the resource carries now. Whatever etag_of takes: a version token (int, str, UUID, datetime) or a representation (bytes, JSON data, a pydantic model). None says the resource does not exist, which is what makes If-None-Match: * a create. TYPE: object DEFAULT: _UNSET
etag	An entity tag that is already one, quotes included. For a tag read from a store or an upstream service rather than built here. Pass this or the version, never both. TYPE: str | None DEFAULT: None
require	Answer 428 when the request carries no precondition at all. Default True: a handler that checks is a handler whose write must be conditional. TYPE: bool DEFAULT: True

RAISES	DESCRIPTION
PreconditionFailedError	If the client's entity tag is not the one the resource carries. Answers 412.
PreconditionRequiredError	If require and the request carried no precondition. Answers 428.
OutOfContextError	If ConditionalRequests() is not registered, so no request headers were ever read.
TypeError	If neither the version nor etag is given, or both are.

etag_of

etag_of(value: object, *, weak: bool = False) -> str

Build an entity tag for a resource.

check_freshness and check_precondition build one for you, so a handler reaches for this only when it needs the tag itself: to set one on a response the middleware will not tag, to hand one to check_precondition(etag=...), to store one beside a cached object, or to compare tags in a client of another service.

Two ways in, because a service has one of two things at hand:

etag_of(cart.version)  # a version column: "7"
etag_of(cart)  # a pydantic model: "b1946ac9..."

A version token is used as it stands, since it already identifies the version and hashing it would only make it longer. A representation is serialized and hashed with SHA-256.

Value	Tag
int, str, UUID	the value, quoted
datetime	its ISO 8601 form, quoted
bytes	SHA-256 of the bytes
a pydantic model, dict, list	SHA-256 of the canonical JSON

weak=True marks the tag weak, which says equivalent rather than byte for byte. A weak tag still answers 304 on a read and never satisfies an If-Match, which takes strong comparison.

Prefer a version token. A hash of the representation changes whenever the serialization does, so adding a field to the model changes every entity tag your service has ever issued, and every client holding one gets 412 until it fetches again.

The serialization is canonical: sorted keys, no spaces, and never the JSON library that happens to be installed, so every replica produces the same tag for the same value. A pydantic model goes through model_dump(mode="json") first.

Read more in the Conditional Requests docs.

PARAMETER	DESCRIPTION
value	What identifies this version of the resource. A version token (int, str, UUID, datetime) becomes the entity tag itself, and a representation (bytes, a mapping, a sequence, or a pydantic model) is hashed into one. TYPE: object
weak	Mark the tag weak. A weak tag answers 304 on a read and never matches an If-Match, which takes strong comparison. TYPE: bool DEFAULT: False

RAISES	DESCRIPTION
TypeError	If the value is a bool, or is neither a version token nor something that serializes to JSON.
ValueError	If a version token holds a quote or a control character, which an entity tag cannot carry. An entity tag that is already one goes to check_precondition(etag=...) instead.

send_error async

send_error(send: Send, rendered: RenderedError) -> None

Write a rendered error as a complete ASGI response.

For a middleware that refuses a request itself, before the app runs and before any exception handler can see it. It takes what the app's ErrorResponses produced rather than a body of its own, so a middleware cannot answer in a format the rest of the app does not speak.

class Gate:
    def __init__(self, app):
        self.app = app

    async def __call__(self, scope, receive, send):
        try:
            await self.app(scope, receive, send)
        except Exception as exc:
            errors = Grelmicro.current().error_responses
            rendered = errors.render(exc, instance=scope["path"])
            if rendered is None:
                # Not a rejection grelmicro renders, so it stays the
                # framework's to answer.
                raise
            await send_error(send, rendered)

PARAMETER	DESCRIPTION
send	The ASGI send callable of the request being refused. TYPE: Send
rendered	What to write, from ErrorResponses.render. TYPE: RenderedError

merge_headers

merge_headers(
    rendered: RenderedError,
    theirs: Mapping[str, str] | None,
) -> dict[str, str]

Merge the app's headers over ours, keeping the safety ones ours.

A header the app set on an exception is part of the answer, WWW-Authenticate on a 401 above all, so it outranks what the component produced. Retry-After is data too, and an app that knows better than the limiter may say so.

The two safety headers are not data. grelmicro adds them because of the body it renders: no-store because a refusal is about one client at one moment, and nosniff because that body reflects the request path back. Letting them be overridden would take away a guarantee the docs make.

Names are lowercased first. They are case insensitive, and ours are lowercase, so a canonical-cased Cache-Control would otherwise survive beside our cache-control and emit two contradictory directives.

PARAMETER	DESCRIPTION
rendered	What the component produced, safety headers included. TYPE: RenderedError
theirs	Headers the exception carried, keyed however the app wrote them. TYPE: Mapping[str, str] | None