FastAPI Response Filtering
When building APIs, you often need to control exactly what data gets sent back to the client. Response filtering allows you to show or hide specific fields based on factors like user permissions, API versions, or performance requirements. In this tutorial, we'll explore how to implement response filtering in FastAPI.
Introduction to Response Filtering
Response filtering is the practice of customizing API responses by including or excluding certain fields before sending them to the client. Common reasons for filtering responses include:
- Hiding sensitive information
- Reducing payload size for better performance
- Providing different views of data based on user roles
- Supporting different API versions or client requirements
FastAPI provides several elegant ways to implement response filtering, which we'll explore in this guide.
Basic Response Filtering with Pydantic Models
The simplest way to filter responses in FastAPI is to use different Pydantic models for input and output data.
Creating Response Models
Let's start with a basic example of a user model:
from fastapi import FastAPI
from pydantic import BaseModel
app = FastAPI()
# Database model (complete data)
class UserInDB(BaseModel):
id: int
username: str
email: str
hashed_password: str
is_active: bool
# Response model (filtered data)
class User(BaseModel):
id: int
username: str
email: str
is_active: bool
@app.get("/users/{user_id}", response_model=User)
async def get_user(user_id: int):
# Simulating database retrieval
user_data = {
"id": user_id,
"username": "johndoe",
"email": "[email protected]",
"hashed_password": "secrethash123", # This will be filtered out
"is_active": True
}
return UserInDB(**user_data) # FastAPI will filter using the response_model
In this example, even though our function returns a UserInDB instance with the hashed_password field, FastAPI will automatically filter the response to match the User model, which doesn't include the password field.
Dynamic Response Filtering
Sometimes you need to filter responses dynamically based on certain conditions. Let's explore how to do that.
Using Response Model with include and exclude
FastAPI allows you to use the response_model_include and response_model_exclude parameters to dynamically control which fields to include or exclude:
from fastapi import FastAPI, Query
from typing import Optional
from pydantic import BaseModel
app = FastAPI()
class Item(BaseModel):
name: str
description: str
price: float
tax: float
inventory_count: int
supplier_info: str
@app.get("/items/{item_id}", response_model=Item)
async def read_item(
item_id: int,
include_supplier: bool = Query(False, description="Include supplier information")
):
item = {
"name": "Fancy Widget",
"description": "A fancy widget with amazing features",
"price": 29.99,
"tax": 2.50,
"inventory_count": 100,
"supplier_info": "WidgetCorp International"
}
# Dynamically exclude supplier_info based on the query parameter
if not include_supplier:
return Item(**item, supplier_info="") # Or filter it at the response model level
return item
For more complex filtering, you can use the response_model_exclude parameter:
from fastapi import FastAPI, Depends
from pydantic import BaseModel
from typing import List, Optional
app = FastAPI()
class Product(BaseModel):
id: int
name: str
price: float
description: str
inventory: int
supplier_code: str
internal_notes: str
def get_fields_to_exclude(is_admin: bool = False):
# Regular users don't see inventory, supplier_code, and internal_notes
if not is_admin:
return {"inventory", "supplier_code", "internal_notes"}
# Admins see everything
return set()
@app.get("/products/{product_id}")
async def get_product(
product_id: int,
is_admin: bool = False,
exclude_fields: set = Depends(get_fields_to_exclude)
):
product = {
"id": product_id,
"name": "Awesome Product",
"price": 49.99,
"description": "This is an awesome product!",
"inventory": 42,
"supplier_code": "SUP-123",
"internal_notes": "Restock needed soon"
}
# Create a new dict excluding the specified fields
filtered_product = {k: v for k, v in product.items() if k not in exclude_fields}
return filtered_product
Role-Based Response Filtering
A common scenario is filtering responses based on user roles. Here's how you can implement role-based filtering:
from fastapi import FastAPI, Depends, HTTPException, status
from pydantic import BaseModel
from typing import Optional
app = FastAPI()
# User roles
class UserRole:
REGULAR = "regular"
ADMIN = "admin"
GUEST = "guest"
# Different response models for different roles
class ProductBase(BaseModel):
id: int
name: str
description: str
class ProductGuest(ProductBase):
pass # Guests only see basic info
class ProductRegular(ProductBase):
price: float
rating: float
class ProductAdmin(ProductRegular):
inventory_count: int
cost_price: float
supplier_id: str
# Simulate a database
products_db = {
1: {
"id": 1,
"name": "Smart Watch",
"description": "A smartwatch with health tracking features",
"price": 199.99,
"rating": 4.7,
"inventory_count": 156,
"cost_price": 89.99,
"supplier_id": "SUP-42"
}
}
# Dependency to get current user role
def get_current_user_role(role: str = "guest"):
# In a real app, this would verify a token and return the user's role
return role
@app.get("/products/{product_id}")
async def get_product(product_id: int, user_role: str = Depends(get_current_user_role)):
if product_id not in products_db:
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Product not found")
product_data = products_db[product_id]
# Return different data based on the user's role
if user_role == UserRole.ADMIN:
return ProductAdmin(**product_data)
elif user_role == UserRole.REGULAR:
return ProductRegular(**product_data)
else: # Guest or any other role
return ProductGuest(**product_data)
To test this endpoint with different roles, you'd make requests like:
/products/1?role=admin
/products/1?role=regular
/products/1 # defaults to guest role
Advanced Filtering with Response Middleware
For more advanced cases, you might want to implement a custom response middleware that can modify any response:
from fastapi import FastAPI, Request
from fastapi.responses import JSONResponse
from starlette.middleware.base import BaseHTTPMiddleware
import json
app = FastAPI()
class ResponseFilterMiddleware(BaseHTTPMiddleware):
async def dispatch(self, request: Request, call_next):
# Process the request and get the response
response = await call_next(request)
# Only filter JSON responses
if response.headers.get("content-type") == "application/json":
# Get the response body
body = b""
async for chunk in response.body_iterator:
body += chunk
# Parse and filter the response
data = json.loads(body.decode())
# Apply filtering rules
# For example, remove any key ending with "_secret"
if isinstance(data, dict):
filtered_data = {k: v for k, v in data.items() if not k.endswith("_secret")}
else:
filtered_data = data
# Create a new response with the filtered data
return JSONResponse(
status_code=response.status_code,
content=filtered_data,
headers=dict(response.headers),
)
return response
# Add the middleware
app.add_middleware(ResponseFilterMiddleware)
@app.get("/sensitive-data")
async def get_sensitive_data():
return {
"public_name": "John Doe",
"email": "[email protected]",
"api_key_secret": "abc123xyz456", # This will be filtered out
"notes": "Regular customer"
}
This middleware will strip out any keys ending with "_secret" from all JSON responses.
Conditional Field Inclusion
Sometimes you want to include fields only if they meet certain conditions. Pydantic models can help here:
from fastapi import FastAPI
from pydantic import BaseModel, Field
from typing import Optional
app = FastAPI()
class UserProfile(BaseModel):
id: int
username: str
email: str
full_name: str
# Only include if not None
bio: Optional[str] = None
# Custom conditional logic using computed fields
premium_member: bool
@property
def special_features(self) -> Optional[list]:
# Only premium members see special features
if self.premium_member:
return ["advanced_search", "priority_support", "ad_free"]
return None
@app.get("/profile/{user_id}", response_model=UserProfile)
async def get_profile(user_id: int):
# Simulate fetching user data
user_data = {
"id": user_id,
"username": "techuser42",
"email": "[email protected]",
"full_name": "Tech User",
"bio": "I love technology!",
"premium_member": True
}
return UserProfile(**user_data)
Versioned API Responses
When supporting multiple API versions, you might want to filter responses differently:
from fastapi import FastAPI, Header, HTTPException
from typing import Optional, Dict, Any
from pydantic import BaseModel
app = FastAPI()
class ProductV1(BaseModel):
id: int
name: str
price: float
class ProductV2(ProductV1):
description: str
category: str
@app.get("/products/{product_id}")
async def get_product(
product_id: int,
api_version: Optional[str] = Header(None, convert_underscores=False)
):
# Simulated product data
product_data = {
"id": product_id,
"name": "Wireless Headphones",
"price": 79.99,
"description": "High-quality wireless headphones with noise cancellation",
"category": "Electronics"
}
# Return different data based on API version
if not api_version or api_version == "1.0":
return ProductV1(**product_data)
elif api_version == "2.0":
return ProductV2(**product_data)
else:
raise HTTPException(status_code=400, detail="Unsupported API version")
Summary
Response filtering in FastAPI gives you fine-grained control over what data you send to clients. We've explored several approaches:
- Pydantic Response Models - Using different models for input and output
- Dynamic Filtering - Using
response_model_includeandresponse_model_exclude - Role-Based Filtering - Returning different data based on user roles
- Response Middleware - Custom middleware for global response modifications
- Conditional Fields - Including fields based on specific conditions
- API Versioning - Filtering responses based on API version
By implementing these techniques, you can build more efficient, secure, and client-friendly APIs that send only the data that's needed and appropriate for each request.
Additional Resources and Exercises
Additional Resources
- FastAPI Official Documentation on Response Models
- Pydantic Fields Documentation
- Starlette Middleware Documentation
Exercises
-
Basic Filtering: Create an API endpoint for a blog post that returns different fields for authenticated vs. unauthenticated users.
-
Role-Based Permissions: Build an e-commerce product API that shows different product details based on whether the user is a customer, store manager, or administrator.
-
Performance Optimization: Create an API endpoint that can optionally include or exclude large nested data structures based on query parameters to optimize response size.
-
Custom Middleware: Write a middleware that automatically removes any field named "password" or containing the string "secret" from all JSON responses.
-
Versioned API: Implement an API endpoint with three different versions (v1, v2, v3) that progressively include more data fields in the response.
Remember that proper response filtering not only improves security by hiding sensitive information but also enhances performance by reducing the size of API responses.
If you spot any mistakes on this website, please let me know at [email protected]. I’d greatly appreciate your feedback! :)