FastAPI Request and Response
Overview
HTTP requests and responses are the core of web APIs. FastAPI provides powerful request processing and response generation mechanisms, supporting automatic validation, serialization, documentation generation, and more. This chapter will深入探讨 how to handle various types of request data and customize response formats.
📥 Request Body Processing
Pydantic Model Request Body
python
from fastapi import FastAPI, HTTPException, status
from pydantic import BaseModel, Field, validator
from typing import Optional, List
from datetime import datetime
from enum import Enum
app = FastAPI()
class Priority(str, Enum):
LOW = "low"
MEDIUM = "medium"
HIGH = "high"
URGENT = "urgent"
class TaskCreate(BaseModel):
title: str = Field(..., min_length=1, max_length=200, description="Task title")
description: Optional[str] = Field(None, max_length=1000, description="Task description")
priority: Priority = Field(Priority.MEDIUM, description="Task priority")
due_date: Optional[datetime] = Field(None, description="Due date")
tags: List[str] = Field(default_factory=list, description="Task tags")
estimated_hours: Optional[float] = Field(None, ge=0.1, le=1000, description="Estimated hours")
@validator('tags')
def validate_tags(cls, v):
if len(v) > 10:
raise ValueError('Cannot have more than 10 tags')
return [tag.strip().lower() for tag in v if tag.strip()]
@validator('due_date')
def validate_due_date(cls, v):
if v and v < datetime.now():
raise ValueError('Due date cannot be in the past')
return v
class TaskResponse(BaseModel):
id: int
title: str
description: Optional[str]
priority: Priority
due_date: Optional[datetime]
tags: List[str]
estimated_hours: Optional[float]
created_at: datetime
updated_at: datetime
is_completed: bool = False
class Config:
from_attributes = True
# Create task
@app.post("/tasks/", response_model=TaskResponse, status_code=status.HTTP_201_CREATED)
async def create_task(task: TaskCreate):
# Simulate creating task
task_data = task.dict()
task_data.update({
"id": 1,
"created_at": datetime.now(),
"updated_at": datetime.now(),
"is_completed": False
})
return TaskResponse(**task_data)Nested Models
python
class Address(BaseModel):
street: str = Field(..., min_length=1, max_length=200)
city: str = Field(..., min_length=1, max_length=100)
state: str = Field(..., min_length=2, max_length=50)
postal_code: str = Field(..., regex=r"^\d{5}(-\d{4})?$")
country: str = Field(default="US", max_length=50)
class Contact(BaseModel):
email: str = Field(..., regex=r"^[^@]+@[^@]+\.[^@]+$")
phone: Optional[str] = Field(None, regex=r"^\+?1?\d{9,15}$")
class UserCreate(BaseModel):
username: str = Field(..., min_length=3, max_length=50, regex="^[a-zA-Z0-9_]+$")
full_name: str = Field(..., min_length=1, max_length=100)
address: Address
contact: Contact
preferences: Optional[dict] = Field(default_factory=dict)
@validator('preferences')
def validate_preferences(cls, v):
allowed_keys = {'theme', 'language', 'notifications', 'timezone'}
if not all(key in allowed_keys for key in v.keys()):
raise ValueError(f'preferences can only contain: {allowed_keys}')
return v
@app.post("/users/", response_model=dict)
async def create_user(user: UserCreate):
return {
"message": "User created successfully",
"user": user.dict(),
"summary": {
"username": user.username,
"location": f"{user.address.city}, {user.address.state}",
"contact_email": user.contact.email
}
}Multiple Request Body Parameters
python
class Item(BaseModel):
name: str
price: float
description: Optional[str] = None
class User(BaseModel):
username: str
full_name: str
class Metadata(BaseModel):
source: str = "api"
timestamp: datetime = Field(default_factory=datetime.now)
@app.post("/items/{item_id}")
async def update_item_with_user(
item_id: int,
item: Item,
user: User,
metadata: Metadata,
importance: int = Field(..., ge=1, le=5, description="Importance 1-5")
):
return {
"item_id": item_id,
"item": item.dict(),
"user": user.dict(),
"metadata": metadata.dict(),
"importance": importance,
"operation": "update_with_context"
}📤 Response Models
Basic Response Models
python
from typing import Union
class ErrorResponse(BaseModel):
error: str
message: str
details: Optional[dict] = None
class SuccessResponse(BaseModel):
success: bool = True
message: str
data: Optional[dict] = None
class PaginatedResponse(BaseModel):
items: List[dict]
total: int
page: int
per_page: int
pages: int
@validator('page')
def validate_page(cls, v, values):
if 'pages' in values and v > values['pages']:
raise ValueError('Page number cannot exceed total pages')
return v
# Use Union type for response
@app.get("/items/{item_id}", response_model=Union[TaskResponse, ErrorResponse])
async def get_item(item_id: int):
if item_id <= 0:
return ErrorResponse(
error="invalid_id",
message="Item ID must be greater than 0",
details={"provided_id": item_id}
)
# Simulate getting item
if item_id == 999:
return ErrorResponse(
error="not_found",
message="Specified item not found"
)
return TaskResponse(
id=item_id,
title="Sample task",
description="This is a sample task",
priority=Priority.MEDIUM,
created_at=datetime.now(),
updated_at=datetime.now()
)Response Model Inheritance
python
class BaseResponse(BaseModel):
timestamp: datetime = Field(default_factory=datetime.now)
api_version: str = "1.0"
class UserResponse(BaseResponse):
id: int
username: str
email: str
is_active: bool
class UserListResponse(BaseResponse):
users: List[UserResponse]
total_count: int
class UserDetailResponse(UserResponse):
full_name: str
created_at: datetime
last_login: Optional[datetime]
profile: Optional[dict]
@app.get("/users/", response_model=UserListResponse)
async def list_users():
return UserListResponse(
users=[
UserResponse(id=1, username="alice", email="alice@example.com", is_active=True),
UserResponse(id=2, username="bob", email="bob@example.com", is_active=False)
],
total_count=2
)
@app.get("/users/{user_id}", response_model=UserDetailResponse)
async def get_user_detail(user_id: int):
return UserDetailResponse(
id=user_id,
username="alice",
email="alice@example.com",
is_active=True,
full_name="Alice Johnson",
created_at=datetime.now(),
last_login=datetime.now(),
profile={"bio": "Software developer", "location": "San Francisco"}
)🎭 Response Customization
Multiple Response Formats
python
from fastapi import Response
from fastapi.responses import JSONResponse, HTMLResponse, PlainTextResponse
@app.get("/data/{format}")
async def get_data_in_format(format: str, item_id: int = 1):
data = {"id": item_id, "name": "Sample Item", "value": 42}
if format == "json":
return JSONResponse(content=data)
elif format == "html":
html_content = f"""
<html>
<body>
<h1>Item {data['id']}</h1>
<p>Name: {data['name']}</p>
<p>Value: {data['value']}</p>
</body>
</html>
"""
return HTMLResponse(content=html_content)
elif format == "text":
text_content = f"ID: {data['id']}, Name: {data['name']}, Value: {data['value']}"
return PlainTextResponse(content=text_content)
else:
raise HTTPException(
status_code=400,
detail="Unsupported format, supported formats: json, html, text"
)Conditional Response
python
from fastapi.responses import RedirectResponse
@app.get("/items/{item_id}")
async def get_item_conditional(
item_id: int,
format: str = "json",
redirect_if_inactive: bool = False
):
# Simulate item data
item = {
"id": item_id,
"name": "Sample Item",
"is_active": item_id % 2 == 0, # Even IDs are active
"created_at": datetime.now().isoformat()
}
# Conditional redirect
if not item["is_active"] and redirect_if_inactive:
return RedirectResponse(
url=f"/items/active?suggested_id={item_id + 1}",
status_code=status.HTTP_307_TEMPORARY_REDIRECT
)
# Conditional response format
if format == "summary":
return {"id": item["id"], "name": item["name"]}
elif format == "full":
return item
else:
# Default format
return {"id": item["id"], "active": item["is_active"]}📋 Status Code Management
Custom Status Codes
python
@app.post("/items/", status_code=status.HTTP_201_CREATED)
async def create_item_with_status(item: Item):
return {"message": "Item created successfully", "item": item.dict()}
@app.put("/items/{item_id}")
async def update_item_with_conditional_status(item_id: int, item: Item):
# Simulate checking if item exists
item_exists = item_id <= 100
if not item_exists:
# Create new item
return JSONResponse(
status_code=status.HTTP_201_CREATED,
content={
"message": "Item does not exist, created new item",
"item_id": item_id,
"item": item.dict()
}
)
else:
# Update existing item
return JSONResponse(
status_code=status.HTTP_200_OK,
content={
"message": "Item updated successfully",
"item_id": item_id,
"item": item.dict()
}
)
@app.delete("/items/{item_id}", status_code=status.HTTP_204_NO_CONTENT)
async def delete_item(item_id: int):
# Delete successfully, return 204 No Content
return Response(status_code=status.HTTP_204_NO_CONTENT)Error Response Handling
python
class ValidationErrorResponse(BaseModel):
error_type: str = "validation_error"
message: str
field_errors: List[dict]
class NotFoundErrorResponse(BaseModel):
error_type: str = "not_found"
message: str
resource: str
@app.get("/items/{item_id}")
async def get_item_with_error_handling(item_id: int):
if item_id <= 0:
return JSONResponse(
status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
content=ValidationErrorResponse(
message="Item ID validation failed",
field_errors=[{
"field": "item_id",
"error": "Must be greater than 0",
"provided_value": item_id
}]
).dict()
)
if item_id == 404:
return JSONResponse(
status_code=status.HTTP_404_NOT_FOUND,
content=NotFoundErrorResponse(
message="Item not found",
resource="item"
).dict()
)
return {"item_id": item_id, "name": "Sample Item"}🔧 Request Headers and Response Headers
Handling Request Headers
python
from fastapi import Header, Request
@app.get("/info/")
async def get_request_info(
request: Request,
user_agent: Optional[str] = Header(None),
accept_language: Optional[str] = Header(None, alias="accept-language"),
x_api_key: Optional[str] = Header(None, alias="x-api-key"),
authorization: Optional[str] = Header(None)
):
return {
"client_ip": request.client.host,
"user_agent": user_agent,
"accept_language": accept_language,
"has_api_key": x_api_key is not None,
"has_auth": authorization is not None,
"all_headers": dict(request.headers)
}
@app.post("/upload/")
async def upload_file(
content_type: str = Header(...),
content_length: int = Header(...),
x_file_name: Optional[str] = Header(None, alias="x-file-name")
):
if content_length > 10 * 1024 * 1024: # 10MB
raise HTTPException(
status_code=status.HTTP_413_REQUEST_ENTITY_TOO_LARGE,
detail="File size cannot exceed 10MB"
)
return {
"content_type": content_type,
"content_length": content_length,
"file_name": x_file_name,
"status": "ready_to_process"
}Setting Response Headers
python
from fastapi import Response
@app.get("/download/{file_id}")
async def download_file(file_id: int, response: Response):
# Set download response headers
response.headers["Content-Disposition"] = f"attachment; filename=file_{file_id}.txt"
response.headers["Content-Type"] = "text/plain"
response.headers["X-Download-ID"] = str(file_id)
return {"content": f"This is content of file {file_id}"}
@app.get("/api/data/")
async def get_api_data():
content = {"data": "sample", "timestamp": datetime.now().isoformat()}
headers = {
"X-API-Version": "1.0",
"X-Rate-Limit": "100",
"X-Rate-Remaining": "95",
"Cache-Control": "public, max-age=300"
}
return JSONResponse(content=content, headers=headers)🍪 Cookie Processing
Reading and Setting Cookies
python
from fastapi import Cookie
@app.get("/profile/")
async def get_profile(
session_id: Optional[str] = Cookie(None),
user_preferences: Optional[str] = Cookie(None, alias="user-preferences")
):
return {
"has_session": session_id is not None,
"session_id": session_id,
"preferences": user_preferences,
"message": "Get user preferences"
}
@app.post("/login/")
async def login(username: str, password: str, response: Response):
# Simulate login authentication
if username == "admin" and password == "secret":
# Set session cookie
response.set_cookie(
key="session_id",
value="abc123xyz",
max_age=3600, # 1 hour
httponly=True,
secure=True, # HTTPS only
samesite="strict"
)
# Set preferences cookie
response.set_cookie(
key="user-preferences",
value="theme=dark;lang=zh",
max_age=86400 * 30 # 30 days
)
return {"message": "Login successful", "username": username}
else:
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Invalid username or password"
)
@app.post("/logout/")
async def logout(response: Response):
# Delete cookie
response.delete_cookie(key="session_id")
response.delete_cookie(key="user-preferences")
return {"message": "Logged out successfully"}📊 Data Validation and Transformation
Custom Validators
python
from pydantic import BaseModel, validator, root_validator
import re
class ProductCreate(BaseModel):
name: str = Field(..., min_length=2, max_length=100)
sku: str = Field(..., description="Product SKU")
price: float = Field(..., gt=0, description="Product price")
category: str
specifications: dict = Field(default_factory=dict)
@validator('sku')
def validate_sku(cls, v):
# SKU format validation: letters start, followed by numbers and letters
if not re.match(r'^[A-Z]{2,3}\d{3,6}$', v.upper()):
raise ValueError('SKU format error, should be 2-3 letters followed by 3-6 numbers')
return v.upper()
@validator('category')
def validate_category(cls, v):
allowed_categories = ['electronics', 'clothing', 'books', 'home', 'sports']
if v.lower() not in allowed_categories:
raise ValueError(f'Category must be one of: {", ".join(allowed_categories)}')
return v.lower()
@root_validator
def validate_product(cls, values):
# Cross-field validation
category = values.get('category')
price = values.get('price')
if category == 'electronics' and price < 10:
raise ValueError('Electronics product price cannot be less than 10')
if category == 'books' and price > 1000:
raise ValueError('Books product price cannot exceed 1000')
return values
@app.post("/products/", response_model=dict)
async def create_product(product: ProductCreate):
return {
"message": "Product created successfully",
"product": product.dict(),
"generated_id": hash(product.sku) % 10000
}Data Transformation and Formatting
python
class EventCreate(BaseModel):
title: str
start_time: datetime
duration_minutes: int = Field(..., ge=15, le=480) # 15 minutes to 8 hours
location: str
attendees: List[str] = Field(default_factory=list)
@validator('title')
def clean_title(cls, v):
# Clean and format title
return ' '.join(v.strip().split())
@validator('attendees', pre=True)
def parse_attendees(cls, v):
# Handle comma-separated string
if isinstance(v, str):
return [email.strip() for email in v.split(',') if email.strip()]
return v
@validator('attendees')
def validate_emails(cls, v):
email_pattern = r'^[^@]+@[^@]+\.[^@]+$'
for email in v:
if not re.match(email_pattern, email):
raise ValueError(f'Invalid email address: {email}')
return v
@app.post("/events/")
async def create_event(event: EventCreate):
# Calculate end time
end_time = event.start_time + timedelta(minutes=event.duration_minutes)
return {
"event": event.dict(),
"computed_fields": {
"end_time": end_time.isoformat(),
"duration_hours": event.duration_minutes / 60,
"attendee_count": len(event.attendees)
}
}Summary
This chapter深入介绍了FastAPI's request and response processing:
- ✅ Request Body Processing: Pydantic models, nested models, multiple parameters
- ✅ Response Models: Basic response, model inheritance, conditional response
- ✅ Response Customization: Multiple formats, status code management
- ✅ HTTP Header Processing: Request header reading, response header setting
- ✅ Cookie Management: Reading, setting, deleting cookies
- ✅ Data Validation: Custom validators, data transformation
These features enable FastAPI to handle complex API requirements, providing type safety and automatic documentation generation while maintaining code simplicity and maintainability.
Request Response Design Recommendations
- Use Pydantic models to ensure data type safety
- Provide clear error response formats
- Use appropriate HTTP status codes
- Add appropriate response header information
- Consider API backward compatibility
- Write detailed model documentation
In the next chapter, we will learn FastAPI's form processing and file upload functionality.