I. Why Do Requests Time Out?¶
Imagine ordering takeout and waiting 15 minutes for delivery—you’d see a “delivery timeout” message. Similarly, if your FastAPI interface takes too long to respond, users will feel “waited too long” and even abandon the request. The essence of a timeout is the server taking longer to process the request than the client’s wait threshold. Common causes include:
- User network lag: Unstable client network terminates the request prematurely.
- Server overload: Too many simultaneous requests overwhelm the server, causing delays.
- Slow interface logic: IO operations like database queries or file reads/writes block the code, increasing processing time.
The consequences are clear: poor user experience, failed requests, and even cascading failures in downstream services. Thus, we need to proactively set timeouts in FastAPI and use asynchronous processing to speed up responses.
II. How to Set Request Timeouts in FastAPI?¶
FastAPI simplifies timeout configuration with the timeout parameter to control maximum processing time per request. Here are two common scenarios:
1. Route-Level Timeout¶
In asynchronous routes, set timeout (in seconds) directly in the async def function. FastAPI returns a timeout error if processing exceeds this duration.
from fastapi import FastAPI
import asyncio
app = FastAPI()
# Simulate a 15-second async task (e.g., DB query, API call)
@app.get("/slow-task")
async def slow_task(timeout: int = 10): # Timeout set to 10 seconds
try:
await asyncio.sleep(15) # Simulate long operation
return {"status": "Task completed"}
except TimeoutError:
return {"error": "Request timed out! Server took >10 seconds"}
2. Global Timeout¶
For uniform timeout across all requests, use middleware. Here’s a simple global timeout middleware example:
from fastapi import Request, Response
from fastapi.responses import JSONResponse
class TimeoutMiddleware:
def __init__(self, app):
self.app = app
async def __call__(self, scope, receive, send):
# Global timeout set to 10 seconds (requires async framework)
timeout = 10
try:
return await asyncio.wait_for(
self.app(scope, receive, send),
timeout=timeout
)
except asyncio.TimeoutError:
return JSONResponse(
status_code=504, # 504 = Gateway Timeout
content={"error": "Global timeout! Please try again later"}
)
app.add_middleware(TimeoutMiddleware)
III. Asynchronous Processing: The Secret to Faster FastAPI¶
Why does async solve timeouts? Synchronous processing forces “idling,” while async allows handling other requests during waiting (like setting a timer and doing other tasks until the water boils).
1. Synchronous vs. Asynchronous: Code Comparison¶
- Synchronous routes (blocking, for CPU-bound tasks):
import time
def sync_task():
time.sleep(5) # Blocks the entire thread
return "Sync completed"
- Asynchronous routes (non-blocking, for IO-bound tasks):
import asyncio
async def async_task():
await asyncio.sleep(5) # Non-blocking wait
return "Async completed"
2. Asynchronous Route Syntax in FastAPI¶
FastAPI recommends async def for async routes with await for async operations:
@app.get("/async-route")
async def async_route():
# Call async IO operations (e.g., async DB queries)
await asyncio.sleep(3) # Simulate async delay
return {"message": "Async processing done"}
3. Asynchronous Background Tasks¶
For non-critical tasks (e.g., sending emails, generating reports), use BackgroundTasks or asyncio.create_task:
from fastapi import BackgroundTasks
@app.post("/send-email")
async def send_email(background_tasks: BackgroundTasks):
# Add task to background (non-blocking)
background_tasks.add_task(send_async, "user@example.com")
return {"status": "Email queued; response sent immediately"}
async def send_async(to_email):
await asyncio.sleep(10) # Async email delivery (async library required)
print(f"Email sent to {to_email}")
IV. Performance Optimization: From Code to Deployment¶
Optimization requires multi-layered improvements beyond timeouts and async:
1. Caching: Reduce Redundant Computation¶
Cache frequent, rarely changing data (e.g., popular product lists) with lru_cache or Redis:
from functools import lru_cache
# Cache last 100 calls
@lru_cache(maxsize=100)
def get_cached_data(param: int):
# Simulate DB query
return param * 2
@app.get("/cached-data/{param}")
async def cached_data(param: int):
data = get_cached_data(param)
return {"data": data}
2. Database Connection Pooling¶
Avoid frequent connection overhead with connection pools for database operations:
import asyncpg
async def init_db_pool():
return await asyncpg.create_pool(
user="user",
password="password",
database="mydb",
host="localhost",
max_size=20 # Adjust based on server capacity
)
@app.get("/db-data")
async def db_data(pool=Depends(init_db_pool)):
async with pool.acquire() as connection:
result = await connection.fetchrow("SELECT * FROM users LIMIT 10")
return {"data": result}
3. Database Optimization¶
- Avoid N+1 queries: Use
SELECT ... JOINinstead of multiple single queries. - Indexing: Add indexes to frequently queried fields (e.g., user IDs, order numbers).
- Batch operations: Use
INSERT INTO ... VALUES (), ()instead of looping single inserts.
4. Deployment: Load Balancing¶
For single-server limitations, use:
- Uvicorn multi-process: uvicorn main:app --workers 4 --reload (4 workers).
- Load balancers: Distribute traffic across servers (e.g., AWS ELB, Azure SLB).
V. Summary: How to Optimize Holistically?¶
- Set timeouts first: Use
timeout(10-30 seconds) to prevent user frustration. - Async for IO-bound tasks: Use async libraries (e.g.,
asyncpg,httpx.AsyncClient) for DB/API calls. - Cache frequently accessed data: Use
lru_cacheor Redis to reduce redundant work. - Optimize database connections: Configure connection pools.
- Scale deployment: Multi-process and load balancing for high concurrency.
By combining these strategies, your FastAPI endpoints will minimize timeouts, handle high traffic efficiently, and improve both user experience and system stability.