Async Request Logging in FastAPI
Request logging tracks every HTTP request and response, capturing method, path, status code, latency, and user context. For debugging production issues, rich logs are often your only window into what happened. FastAPI middleware is the ideal place to add logging—it runs before and after every route, has access to request/response details, and can measure total request time including downstream services. This guide shows you how to implement production-grade logging that's both informative and efficient.
I've debugged production outages where logs were either too sparse (missing context) or too verbose (drowning in noise). This article teaches you the Goldilocks approach: right level of detail at the right times.
Simple Synchronous Request Logging
Start with a basic logging middleware that captures essential request/response information:
from fastapi import FastAPI
from starlette.middleware.base import BaseHTTPMiddleware
from starlette.requests import Request
from starlette.responses import Response
import time
import logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
class LoggingMiddleware(BaseHTTPMiddleware):
async def dispatch(self, request: Request, call_next) -> Response:
method = request.method
path = request.url.path
start_time = time.time()
try:
response = await call_next(request)
process_time = time.time() - start_time
logger.info(
f"{method} {path} {response.status_code} {process_time:.3f}s"
)
return response
except Exception as exc:
process_time = time.time() - start_time
logger.error(
f"{method} {path} ERROR {process_time:.3f}s: {str(exc)}"
)
raise
app = FastAPI()
app.add_middleware(LoggingMiddleware)
This logs one line per request with method, path, status, and duration. Quick and useful for development.
Structured Logging with Dictionaries
For production, use structured logging—logs as JSON dictionaries that tools can parse:
import logging
import time
import json
from pythonjsonlogger import jsonlogger
from starlette.middleware.base import BaseHTTPMiddleware
from starlette.requests import Request
# Configure JSON logging
handler = logging.StreamHandler()
formatter = jsonlogger.JsonFormatter()
handler.setFormatter(formatter)
logger = logging.getLogger(__name__)
logger.addHandler(handler)
logger.setLevel(logging.INFO)
class StructuredLoggingMiddleware(BaseHTTPMiddleware):
async def dispatch(self, request: Request, call_next) -> Response:
method = request.method
path = request.url.path
query_string = str(request.url.query) if request.url.query else ""
client_ip = request.client.host if request.client else "unknown"
start_time = time.time()
try:
response = await call_next(request)
duration_ms = (time.time() - start_time) * 1000
logger.info(
"request_completed",
extra={
"method": method,
"path": path,
"query": query_string,
"status_code": response.status_code,
"duration_ms": round(duration_ms, 2),
"client_ip": client_ip,
"user_agent": request.headers.get("user-agent")
}
)
return response
except Exception as exc:
duration_ms = (time.time() - start_time) * 1000
logger.error(
"request_failed",
extra={
"method": method,
"path": path,
"query": query_string,
"error_type": type(exc).__name__,
"error_message": str(exc),
"duration_ms": round(duration_ms, 2),
"client_ip": client_ip
},
exc_info=True
)
raise
app = FastAPI()
app.add_middleware(StructuredLoggingMiddleware)
Output is JSON:
{"message": "request_completed", "method": "GET", "path": "/users", "status_code": 200, "duration_ms": 15.45, "client_ip": "192.168.1.1"}
{"message": "request_failed", "method": "POST", "path": "/users", "error_type": "ValidationError", "error_message": "Invalid email", "duration_ms": 5.23}
Logging systems like Datadog, ELK, and CloudWatch parse these JSON logs, allowing you to search, filter, and alert on specific fields.
Request ID Propagation
Correlate logs from the same request by adding a unique request ID:
import uuid
from starlette.middleware.base import BaseHTTPMiddleware
class RequestIDMiddleware(BaseHTTPMiddleware):
async def dispatch(self, request: Request, call_next) -> Response:
# Extract or generate request ID
request_id = request.headers.get("X-Request-ID") or str(uuid.uuid4())
request.state.request_id = request_id
response = await call_next(request)
response.headers["X-Request-ID"] = request_id
return response
class StructuredLoggingMiddleware(BaseHTTPMiddleware):
async def dispatch(self, request: Request, call_next) -> Response:
request_id = request.state.request_id
# ... logging code ...
logger.info(
"request_completed",
extra={
"request_id": request_id,
"method": request.method,
"path": request.url.path,
"status_code": response.status_code
}
)
app = FastAPI()
app.add_middleware(StructuredLoggingMiddleware)
app.add_middleware(RequestIDMiddleware)
Now all logs for a single request share the same request_id, making it easy to trace a request through your system.
Logging Request and Response Bodies
For debugging, log request/response bodies (but be careful with sensitive data):
class DetailedLoggingMiddleware(BaseHTTPMiddleware):
async def dispatch(self, request: Request, call_next) -> Response:
request_id = request.state.request_id
# Capture request body
body = b""
if request.method in ("POST", "PUT", "PATCH"):
body = await request.body()
# Re-wrap body for the route handler
async def receive():
return {"type": "http.request", "body": body}
request._receive = receive
# Log request
logger.debug(
"request_received",
extra={
"request_id": request_id,
"method": request.method,
"path": request.url.path,
"body": body.decode() if body else ""
}
)
response = await call_next(request)
# Capture response body (trickier—response already sent)
response_body = b""
async for chunk in response.body_iterator:
response_body += chunk
logger.debug(
"response_sent",
extra={
"request_id": request_id,
"status_code": response.status_code,
"body": response_body.decode() if response_body else ""
}
)
# Re-wrap response so FastAPI sends it
from starlette.responses import StreamingResponse
return StreamingResponse(
iter([response_body]),
status_code=response.status_code,
headers=dict(response.headers)
)
Be cautious: logging request bodies can expose passwords or PII. Filter sensitive fields:
def redact_sensitive_fields(data: dict) -> dict:
"""Redact passwords, API keys, etc. from logs."""
sensitive_keys = ["password", "secret", "token", "api_key", "ssn", "credit_card"]
for key in sensitive_keys:
if key in data:
data[key] = "***REDACTED***"
return data
# In logging middleware
body_data = json.loads(body) if body else {}
redacted_body = redact_sensitive_fields(body_data)
logger.debug(
"request_received",
extra={"body": json.dumps(redacted_body)}
)
Contextual Logging with User Information
Include user context in logs for audit trails and debugging:
class ContextLoggingMiddleware(BaseHTTPMiddleware):
async def dispatch(self, request: Request, call_next) -> Response:
request_id = request.state.request_id
# Extract user info if authenticated
user_id = request.state.get("user_id")
user_email = request.state.get("user_email")
start_time = time.time()
response = await call_next(request)
duration_ms = (time.time() - start_time) * 1000
logger.info(
"request_completed",
extra={
"request_id": request_id,
"method": request.method,
"path": request.url.path,
"status_code": response.status_code,
"duration_ms": round(duration_ms, 2),
"user_id": user_id,
"user_email": user_email,
"client_ip": request.client.host if request.client else None
}
)
return response
app = FastAPI()
app.add_middleware(ContextLoggingMiddleware)
@app.get("/profile")
async def get_profile(request: Request):
# Middleware logs this request with user_id from request.state
return {"user_id": request.state.user_id}
Logs now show who made the request, making it easy to audit actions or debug user-specific issues.
Performance Monitoring Middleware
Log slow requests separately for performance analysis:
class PerformanceLoggingMiddleware(BaseHTTPMiddleware):
SLOW_REQUEST_THRESHOLD_MS = 1000 # Log requests > 1 second
async def dispatch(self, request: Request, call_next) -> Response:
start_time = time.time()
response = await call_next(request)
duration_ms = (time.time() - start_time) * 1000
if duration_ms > self.SLOW_REQUEST_THRESHOLD_MS:
logger.warning(
"slow_request_detected",
extra={
"method": request.method,
"path": request.url.path,
"duration_ms": round(duration_ms, 2),
"threshold_ms": self.SLOW_REQUEST_THRESHOLD_MS,
"status_code": response.status_code
}
)
return response
This helps identify performance bottlenecks. Routes that consistently exceed the threshold are candidates for optimization.
Key Takeaways
- Implement structured logging (JSON) for production; log fields that tools can parse and search.
- Use request IDs to correlate logs across services and middleware.
- Log request/response details (method, path, status, duration) for every request.
- Redact sensitive data (passwords, API keys) before logging.
- Include user context (user ID, email) for audit trails.
- Log slow requests to a separate channel for performance monitoring.
Frequently Asked Questions
How do I log without affecting performance?
Use async logging libraries and write to disk or a service asynchronously. Set log level to INFO in production (not DEBUG, which is noisy).
Should I log response bodies for all requests?
No—it's expensive and creates privacy risks. Log request bodies selectively (with redaction), and response bodies only for debugging specific issues.
How do I prevent logs from growing too large?
Rotate logs (e.g., daily) using RotatingFileHandler. Use a logging service (Datadog, CloudWatch) instead of local files.
Can I log to multiple destinations?
Yes. Add multiple handlers to your logger:
file_handler = logging.FileHandler("app.log")
console_handler = logging.StreamHandler()
logger.addHandler(file_handler)
logger.addHandler(console_handler)
How do I correlate logs across microservices?
Pass request IDs (e.g., X-Request-ID header) through all service calls. Each service logs with that ID.