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
FastAPI Intermediate decorators async timeouts exception-handling middleware-pattern
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

python 
from fastapi import FastAPI, WebSocket, WebSocketDisconnect
 
app = FastAPI()
 

Building a WebSocket chat with FastAPI

websockets broadcast connection-management
Intermediate 9 steps
python 
import time
import uuid
 
from django.utils.deprecation import MiddlewareMixin

Attaching per-request context in Django

middleware request lifecycle multi-tenancy
Intermediate 7 steps
rust 
use axum::{
    extract::{Path, State},
    response::sse::{Event, KeepAlive, Sse},
};

Streaming import progress with SSE in Axum

server-sent-events streams watch-channel
Advanced 7 steps
javascript 
import { useState, useEffect, useCallback, useRef } from 'react';
 
const cache = new Map();
const inflight = new Map();

Building a stale-while-revalidate hook in React

caching request-deduplication custom-hooks
Advanced 10 steps
python 
import random
from typing import Iterator, List
 
 

How reservoir sampling picks k items

reservoir-sampling streaming randomness
Intermediate 5 steps
python 
import secrets
 
from django.contrib.auth import authenticate, login
from django.core.cache import cache

Two-factor login with OTP in Django

two-factor-auth one-time-passwords caching
Intermediate 9 steps

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