FastAPI is a game-changer for building high-performance APIs. Let’s dive into some advanced techniques to take your FastAPI skills to the next level, focusing on complex routes and custom middleware for security and caching.
First up, let’s talk about building complex API routes. FastAPI makes it easy to create nested and dynamic routes that can handle a variety of request types. Here’s a simple example to get us started:
from fastapi import FastAPI, Path
app = FastAPI()
@app.get("/items/{item_id}")
async def read_item(item_id: int = Path(..., ge=1)):
return {"item_id": item_id}
This route uses path validation to ensure the item_id is a positive integer. But we can get much fancier than that. Let’s create a more complex route that handles multiple parameters and query strings:
from fastapi import FastAPI, Path, Query
from typing import Optional
app = FastAPI()
@app.get("/users/{user_id}/orders")
async def get_user_orders(
user_id: int = Path(..., ge=1),
skip: int = Query(0, ge=0),
limit: int = Query(10, le=100),
order_status: Optional[str] = Query(None, regex="^(pending|completed|cancelled)$")
):
# Imagine we're fetching orders from a database here
return {
"user_id": user_id,
"orders": [
{"id": 1, "status": "pending"},
{"id": 2, "status": "completed"}
],
"skip": skip,
"limit": limit,
"order_status": order_status
}
This route is much more powerful. It allows us to fetch a user’s orders with pagination (skip and limit) and optional filtering by order status. The Path and Query validators ensure we’re getting valid data before our function even runs.
Now, let’s talk about custom middleware. Middleware in FastAPI allows you to add custom code to process requests before they reach your route handlers, and to process responses before they’re sent back to the client. This is super useful for things like authentication, logging, and caching.
Let’s start with a simple timing middleware that measures how long each request takes:
import time
from fastapi import FastAPI, Request
app = FastAPI()
@app.middleware("http")
async def add_process_time_header(request: Request, call_next):
start_time = time.time()
response = await call_next(request)
process_time = time.time() - start_time
response.headers["X-Process-Time"] = str(process_time)
return response
This middleware adds an X-Process-Time header to every response, showing how long the request took to process. Pretty neat, right?
But we can do even cooler things with middleware. Let’s create a caching middleware that stores responses in memory for a short time:
from fastapi import FastAPI, Request
from fastapi.responses import JSONResponse
import time
app = FastAPI()
# Our simple in-memory cache
cache = {}
@app.middleware("http")
async def cache_middleware(request: Request, call_next):
# Only cache GET requests
if request.method == "GET":
cache_key = request.url.path + "?" + request.url.query
if cache_key in cache:
cached_response, timestamp = cache[cache_key]
# Return cached response if it's less than 60 seconds old
if time.time() - timestamp < 60:
return JSONResponse(content=cached_response)
response = await call_next(request)
# Cache the response if it's a JSON response
if request.method == "GET" and isinstance(response, JSONResponse):
cache[cache_key] = (response.body, time.time())
return response
This middleware caches GET requests for 60 seconds. It’s a simple implementation, but it shows how powerful middleware can be. In a real-world scenario, you’d probably want to use a more robust caching solution like Redis.
Security is another area where custom middleware shines. Let’s implement a basic API key authentication middleware:
from fastapi import FastAPI, Request, HTTPException
from fastapi.security.api_key import APIKeyHeader
from starlette.status import HTTP_403_FORBIDDEN
API_KEY = "my_secret_api_key"
API_KEY_NAME = "X-API-Key"
api_key_header = APIKeyHeader(name=API_KEY_NAME, auto_error=False)
app = FastAPI()
@app.middleware("http")
async def api_key_middleware(request: Request, call_next):
api_key = request.headers.get(API_KEY_NAME)
if api_key != API_KEY:
raise HTTPException(
status_code=HTTP_403_FORBIDDEN, detail="Could not validate credentials"
)
return await call_next(request)
This middleware checks for a valid API key in the X-API-Key header of each request. If the key isn’t present or doesn’t match, it raises a 403 Forbidden error.
These examples just scratch the surface of what’s possible with FastAPI. The framework’s flexibility allows you to build incredibly powerful and efficient APIs. One of the things I love most about FastAPI is how it encourages you to write self-documenting code. The type hints and validation you add to your routes automatically generate OpenAPI (Swagger) documentation.
Let’s take our API to the next level by adding some more advanced features. How about we implement a rate limiting middleware? This can help protect our API from abuse:
import time
from fastapi import FastAPI, Request, HTTPException
from fastapi.responses import JSONResponse
app = FastAPI()
# Store request counts for each IP
request_counts = {}
# Maximum requests per minute
RATE_LIMIT = 60
@app.middleware("http")
async def rate_limit_middleware(request: Request, call_next):
ip = request.client.host
current_time = time.time()
if ip in request_counts:
last_request_time, count = request_counts[ip]
if current_time - last_request_time < 60: # Within the last minute
if count >= RATE_LIMIT:
raise HTTPException(status_code=429, detail="Rate limit exceeded")
request_counts[ip] = (last_request_time, count + 1)
else:
request_counts[ip] = (current_time, 1)
else:
request_counts[ip] = (current_time, 1)
response = await call_next(request)
return response
This middleware tracks the number of requests from each IP address and blocks requests if they exceed the rate limit. It’s a simple implementation, but it shows how you can use middleware to add powerful features to your API.
Now, let’s combine some of these concepts into a more complex API. We’ll create a simple blog API with post creation, retrieval, and a cached list of recent posts:
from fastapi import FastAPI, HTTPException, Depends
from pydantic import BaseModel
from typing import List, Optional
import time
app = FastAPI()
# Our "database"
posts = []
# Our cache
cache = {}
class Post(BaseModel):
id: Optional[int]
title: str
content: str
author: str
@app.post("/posts", response_model=Post)
async def create_post(post: Post):
post.id = len(posts) + 1
posts.append(post)
# Invalidate the cache when a new post is created
cache.pop("recent_posts", None)
return post
@app.get("/posts/{post_id}", response_model=Post)
async def get_post(post_id: int):
if post_id < 1 or post_id > len(posts):
raise HTTPException(status_code=404, detail="Post not found")
return posts[post_id - 1]
@app.get("/posts", response_model=List[Post])
async def get_recent_posts(limit: int = 10):
cache_key = f"recent_posts:{limit}"
if cache_key in cache:
cached_posts, timestamp = cache[cache_key]
if time.time() - timestamp < 60: # Cache for 60 seconds
return cached_posts
recent_posts = posts[-limit:][::-1] # Get last 'limit' posts in reverse order
cache[cache_key] = (recent_posts, time.time())
return recent_posts
# Add our middleware
@app.middleware("http")
async def add_process_time_header(request: Request, call_next):
start_time = time.time()
response = await call_next(request)
process_time = time.time() - start_time
response.headers["X-Process-Time"] = str(process_time)
return response
This API demonstrates several advanced concepts:
- It uses Pydantic models for request and response validation.
- It implements a simple caching mechanism for the list of recent posts.
- It includes our timing middleware to measure request processing time.
- It shows how to handle route parameters and query parameters.
One of the things I love about FastAPI is how it makes it easy to implement best practices. The use of type hints not only provides better documentation and IDE support, but it also allows FastAPI to do automatic data validation and serialization.
When building complex APIs, it’s important to think about error handling. FastAPI makes it easy to raise HTTP exceptions with custom status codes and error messages. You can even create custom exception handlers for more fine-grained control over error responses.
Here’s an example of a custom exception handler:
from fastapi import FastAPI, Request, HTTPException
from fastapi.responses import JSONResponse
app = FastAPI()
class CustomException(Exception):
def __init__(self, name: str):
self.name = name
@app.exception_handler(CustomException)
async def custom_exception_handler(request: Request, exc: CustomException):
return JSONResponse(
status_code=418,
content={"message": f"Oops! {exc.name} did something. There goes a rainbow..."},
)
@app.get("/unicorn/{name}")
async def read_unicorn(name: str):
if name == "yolo":
raise CustomException(name=name)
return {"unicorn_name": name}
This creates a custom exception and an exception handler that returns a whimsical error message when the exception is raised. It’s a silly example, but it demonstrates how you can create custom error responses for different situations in your API.
As your API grows more complex, you might want to start thinking about structuring your code better. FastAPI supports the use of APIRouter for grouping related routes together. This can help keep your code organized as your API expands:
from fastapi import APIRouter, FastAPI
app = FastAPI()
router = APIRouter()
@router.get("/users", tags=["users"])
async def read_users():
return [{"username": "Rick"}, {"username": "Morty"}]
@router.get("/users/{username}", tags=["users"])
async def read_user(username: str):
return {"username": username}
app.include_router(router)
This approach allows you to split your API into logical sections, making it easier to manage as it grows.
Another advanced feature of FastAPI is its support for background tasks. These allow you to perform operations after returning a response to the client, which can be great for things like sending emails or processing data:
from fastapi import BackgroundTasks, FastAPI
app = FastAPI()
def write_log(message: str):
with open("log.txt", mode="a") as log:
log.write(message)
@app.post("/send-notification/{email}")
async def send_notification(email: str, background_tasks: BackgroundTasks):
background_tasks.add_task(write_log, f"Notification sent to {email}")
return {"message": "Notification sent in the background"}
This endpoint returns immediately, but continues to perform the log-writing task in the background.
As you can see, FastAPI provides a rich set of tools for building complex, high-performance APIs. Its combination of modern Python features, automatic OpenAPI