Middleware¶

Middleware lets you intercept and modify every request and response through FastHTTP — without changing handler code.

What middleware can do¶

Automatically add authorization headers
Log all requests and responses
Add timing and tracing headers
Retry requests on specific response codes
Transform response data

How it works¶

request  →  mw1.request → mw2.request → mw3.request → [HTTP]
response ←  mw1.response ← mw2.response ← mw3.response ← [HTTP]

Middleware executes in __priority__ order on the way in and in reverse order on the way out.

Creating Middleware¶

Create a class inheriting from BaseMiddleware:

from fasthttp import FastHTTP
from fasthttp.middleware import BaseMiddleware
from fasthttp.response import Response


class MyMiddleware(BaseMiddleware):
    __return_type__ = None
    __priority__ = 0
    __methods__ = None
    __enabled__ = True

    async def request(self, method, url, kwargs):
        kwargs["headers"] = kwargs.get("headers") or {}
        kwargs["headers"]["X-Custom"] = "value"
        return kwargs

    async def response(self, response):
        return response

    async def on_error(self, error, route, config):
        print(f"Error: {error}")

Attaching to the app¶

ListPipeSingle

app = FastHTTP(middleware=[AuthMiddleware(), LoggingMiddleware()])

app = FastHTTP(middleware=AuthMiddleware() | LoggingMiddleware())

app = FastHTTP(middleware=MyMiddleware())

Examples¶

Authentication¶

from fasthttp import FastHTTP
from fasthttp.middleware import BaseMiddleware


class AuthMiddleware(BaseMiddleware):
    __return_type__ = bool
    __priority__ = 0
    __methods__ = None
    __enabled__ = True

    def __init__(self, token: str):
        self.token = token

    async def request(self, method, url, kwargs):
        kwargs["headers"] = kwargs.get("headers") or {}
        kwargs["headers"]["Authorization"] = f"Bearer {self.token}"
        return kwargs


app = FastHTTP(middleware=[AuthMiddleware(token="your-token")])

Adding Trace ID¶

import uuid
from fasthttp import FastHTTP
from fasthttp.middleware import BaseMiddleware


class TraceMiddleware(BaseMiddleware):
    __return_type__ = None
    __priority__ = 0
    __methods__ = None
    __enabled__ = True

    async def request(self, method, url, kwargs):
        kwargs["headers"] = kwargs.get("headers") or {}
        kwargs["headers"]["X-Trace-ID"] = str(uuid.uuid4())
        return kwargs


app = FastHTTP(middleware=[TraceMiddleware()])

Logging¶

import time
from contextvars import ContextVar
from fasthttp import FastHTTP
from fasthttp.middleware import BaseMiddleware


class LoggingMiddleware(BaseMiddleware):
    __return_type__ = None
    __priority__ = 99
    __methods__ = None
    __enabled__ = True

    def __init__(self) -> None:
        self._start: ContextVar[float] = ContextVar("log_start", default=0.0)

    async def request(self, method, url, kwargs):
        print(f"→ {method} {url}")
        self._start.set(time.monotonic())
        return kwargs

    async def response(self, response):
        elapsed = time.monotonic() - self._start.get()
        print(f"← {response.status} ({elapsed:.2f}s)")
        return response


app = FastHTTP(middleware=[LoggingMiddleware()])

Caching¶

FastHTTP comes with built-in CacheMiddleware:

from fasthttp import FastHTTP, CacheMiddleware

app = FastHTTP(
    middleware=[CacheMiddleware(ttl=3600, max_size=100)]
)

Caches GET requests in memory with LRU eviction.

Retry¶

FastHTTP comes with built-in RetryMiddleware for automatic retries with exponential backoff:

from fasthttp import FastHTTP, RetryMiddleware

app = FastHTTP(
    middleware=RetryMiddleware(
        max_retries=3,
        retry_on={429, 500, 502, 503, 504},
        backoff_factor=0.5,
    )
)

Automatically retries failed requests on connection errors, timeouts, or specific HTTP status codes.

Parameter	Type	Default	Description
`max_retries`	`int`	`3`	Maximum number of retry attempts
`retry_on`	`set[int]`	`{429, 500, 502, 503, 504}`	HTTP status codes that trigger a retry
`backoff_factor`	`float`	`0.5`	Multiplier for exponential backoff delay
`max_delay`	`float`	`30.0`	Maximum delay between retries in seconds
`retry_exceptions`	`tuple[type[Exception], ...]`	`(Exception,)`	Exception types that trigger a retry

Response modification¶

from fasthttp import FastHTTP
from fasthttp.middleware import BaseMiddleware


class ResponseModifierMiddleware(BaseMiddleware):
    __return_type__ = None
    __priority__ = 0
    __methods__ = None
    __enabled__ = True

    async def response(self, response):
        response.headers["X-Custom-Response"] = "value"
        return response


app = FastHTTP(middleware=[ResponseModifierMiddleware()])

Class attributes¶

Attribute	Type	Description
`__return_type__`	`type \\| None`	Type this middleware operates on
`__priority__`	`int`	Execution order — lower runs first
`__methods__`	`list[str] \\| None`	HTTP methods to intercept. `None` = all methods
`__enabled__`	`bool`	`False` skips without removing from chain

Runtime toggle¶

debug = LoggingMiddleware()
app = FastHTTP(middleware=[debug])

debug.__enabled__ = False   # disable
debug.__enabled__ = True    # re-enable

Event Hooks¶

Event hooks provide a simpler alternative to middleware for common tasks like logging, timing, and error tracking. They run as decorators on your app, router, or session.

`on_request`¶

Runs before each request:

from fasthttp import FastHTTP

app = FastHTTP()

@app.on_request
async def log_request(route, config):
    print(f"→ {route.method} {route.url}")

`on_response`¶

Runs after each successful response:

@app.on_response
async def log_response(response):
    print(f"← {response.status}")

`on_error`¶

Runs when an error occurs:

@app.on_error
async def track_error(error, route):
    print(f"✖ {error} on {route.url}")

`exception_handler`¶

FastAPI-style handler for a specific exception type. Unlike on_error, its return value replaces the route's result instead of the request failing silently — the request is "recovered" rather than just logged.

from fasthttp.exceptions import FastHTTPTimeoutError

@app.exception_handler(FastHTTPTimeoutError)
async def handle_timeout(route, exc):
    return {"error": "timeout", "url": route.url}

The handler receives (route, exc) — same order as FastAPI's (request, exc). If several handlers match (through inheritance), the most specific registered type wins:

@app.exception_handler(Exception)
async def fallback(route, exc):
    return {"error": "unexpected"}

@app.exception_handler(FastHTTPTimeoutError)
async def timeout(route, exc):
    return {"error": "timeout"}  # wins for FastHTTPTimeoutError, fallback still applies to everything else

Only exceptions that actually propagate out of a request reach a handler — for HTTP status errors this means the route (or the app) must have raise_for_status=True, otherwise FastHTTPBadStatusError is never raised and there's nothing to intercept.

With Router¶

Event hooks on a router, including exception_handler, are merged into the app via include_router():

from fasthttp import FastHTTP, Router

router = Router(base_url="https://api.example.com")

@router.on_request
async def router_hook(route, config):
    print(f"[router] → {route.url}")

@router.exception_handler(FastHTTPTimeoutError)
async def router_timeout(route, exc):
    return {"error": "timeout", "url": route.url}

app = FastHTTP()
app.include_router(router)
# both hooks will run for all requests from this router

With AsyncSession¶

from fasthttp import AsyncSession

async with AsyncSession() as session:
    @session.on_request
    async def inject_auth(route, config):
        config["headers"]["Authorization"] = "Bearer token"

    resp = await session.get("https://api.example.com")

exception_handler is only available on FastHTTP and Router — AsyncSession does not support it, since it returns responses directly to the caller instead of routing through handlers.

Middleware vs Event Hooks¶

Feature	Middleware	Event Hooks
Modify request	✅ Yes	✅ Yes
Modify response	✅ Yes	❌ No
Error handling	✅ Yes	✅ Yes
Execution order control	✅ Priority	❌ Registration order
Method filtering	✅ `__methods__`	❌ No
Complexity	Higher	Lower

Comparison with Dependencies¶

Feature	Middleware	Dependencies
Global application	✅ Yes	❌ No
Specific request	❌ No	✅ Yes
Response modification	✅ Yes	❌ No
Error handling	✅ Yes	❌ No
Complexity	Higher	Lower