python 44 lines · 6 steps

Per-route request timeouts in FastAPI

A decorator wraps async handlers with a deadline and turns overruns into a clean 504 response.

Explained by highlit
1from functools import wraps
2import asyncio
3 
4from fastapi import APIRouter, FastAPI, Request
5from fastapi.responses import JSONResponse
6from starlette.status import HTTP_504_GATEWAY_TIMEOUT
7 
8router = APIRouter()
9 
10 
11def with_timeout(seconds: float):
12 def decorator(handler):
13 @wraps(handler)
14 async def wrapper(*args, **kwargs):
15 try:
16 return await asyncio.wait_for(handler(*args, **kwargs), timeout=seconds)
17 except asyncio.TimeoutError:
18 raise RequestTimeout(seconds)
19 return wrapper
20 return decorator
21 
22 
23class RequestTimeout(Exception):
24 def __init__(self, seconds: float):
25 self.seconds = seconds
26 
27 
28def register_timeout_handler(app: FastAPI) -> None:
29 @app.exception_handler(RequestTimeout)
30 async def _handle(request: Request, exc: RequestTimeout):
31 return JSONResponse(
32 status_code=HTTP_504_GATEWAY_TIMEOUT,
33 content={
34 "detail": f"Request exceeded the {exc.seconds:g}s limit",
35 "path": request.url.path,
36 },
37 )
38 
39 
40@router.get("/reports/{report_id}")
41@with_timeout(3.0)
42async def get_report(report_id: int):
43 report = await reports_service.render(report_id)
44 return {"id": report_id, "body": report}
01 / 01
STEP 01

Walkthrough

Space play step click any line
Three takeaways
  1. 1Wrapping a coroutine in asyncio.wait_for gives any handler a hard deadline without touching its logic.
  2. 2Raising a custom exception lets you separate the timeout mechanism from how the error is rendered.
  3. 3A registered exception handler centralizes error formatting so every timed-out route returns a consistent 504.

Related explainers

Share this explainer

Here's the card — post it anywhere.

Per-route request timeouts in FastAPI — share card
Made with highlit — turn any snippet into a walkthrough like this in about a minute.
Explain your code