在 Web 开发中，API 的请求与响应过程如同系统的 “脉搏”，直接决定了服务与客户端交互的效率和质量。作为 Python 领域备受青睐的高性能 Web 框架，FastAPI 凭借其强大且灵活的请求与响应处理机制，为开发者提供了高效构建 API 的解决方案。本文将深入剖析 FastAPI 的请求与响应机制，从请求数据的接收、处理，到响应数据的生成、返回，全面揭示其运行原理与实践技巧
一、FastAPI 请求处理：数据接收与解析
1.1 请求参数类型
在 FastAPI 中，请求参数主要分为路径参数、查询参数、请求体参数和表单参数，每种参数都有其独特的定义和使用方式。
路径参数：通过在路由路径中使用大括号{}定义，用于标识资源的特定标识符。例如：
from fastapi import FastAPI app = FastAPI() @app.get("/users/{user_id}") def read_user(user_id: int): return {"user_id": user_id}上述代码中，
{user_id}是路径参数，FastAPI 会自动将请求 URL 中的对应值提取出来，并根据函数参数的类型注解进行类型转换查询参数：作为 URL 的一部分，通过问号?与路径分隔，以键值对形式传递。在函数中通过普通参数定义，可设置默认值。如：
from fastapi import FastAPI app = FastAPI() @app.get("/users/") def search_users(name: str = None, age: int = None): result = {} if name: result["name"] = name if age: result["age"] = age return result这里
name和age是查询参数，当客户端发送GET请求如/users/?name=Alice&age=25时，参数值会被正确传递给函数请求体参数：用于发送大量数据，如 JSON 格式数据。借助 Pydantic 模型定义请求体结构，FastAPI 会自动解析请求体数据并验证其有效性。示例如下：
from fastapi import FastAPI from pydantic import BaseModel app = FastAPI() class User(BaseModel): name: str age: int @app.post("/users/") def create_user(user: User): return user在上述代码中，
User类定义了请求体的数据结构，客户端发送包含符合该结构的JSON数据的POST请求，FastAPI 会将其解析为User对象传递给create_user函数表单参数：在处理 HTML 表单数据时使用，通过Form类定义。需要注意，使用Form时需要安装python-multipart库。例如：
from fastapi import FastAPI, Form app = FastAPI() @app.post("/login/") def login(username: str = Form(...), password: str = Form(...)): return {"username": username, "password": password}此代码实现了一个简单的登录接口，接收表单提交的
username和password数据
1.2 请求头与 Cookie
除了上述常见参数，FastAPI 还支持获取请求头和 Cookie 信息
请求头：通过函数参数直接定义，参数名对应请求头字段名（自动处理大小写转换）。例如获取User-Agent请求头
from fastapi import FastAPI, Header app = FastAPI() @app.get("/headers/") def read_headers(user_agent: str = Header(None)): return {"User-Agent": user_agent}Cookie：使用Cookie类定义，获取客户端发送的 Cookie 数据。示例：
from fastapi import FastAPI, Cookie app = FastAPI() @app.get("/cookies/") def read_cookies(session_id: str = Cookie(None)): return {"session_id": session_id}
二、FastAPI 响应处理：数据生成与返回
2.1 响应状态码设置
FastAPI 允许开发者通过status_code参数轻松设置响应的状态码。例如，创建资源成功时返回201 Created状态码：
from fastapi import FastAPI
from pydantic import BaseModel
app = FastAPI()
class Item(BaseModel):
name: str
price: float
@app.post("/items/", response_model=Item, status_code=201)
def create_item(item: Item):
return item除了常见的状态码，对于自定义的错误情况，可通过抛出HTTPException来返回特定状态码和错误信息。如：
from fastapi import FastAPI, HTTPException
app = FastAPI()
@app.get("/items/{item_id}")
def read_item(item_id: int):
if item_id not in [1, 2, 3]:
raise HTTPException(status_code=404, detail="Item not found")
return {"item_id": item_id}2.2 响应模型定义
使用response_model参数可以指定响应的数据模型，FastAPI 会自动根据该模型对返回数据进行序列化，并在 API 文档中生成相应的结构说明。结合 Pydantic 模型，能有效保证响应数据的格式规范。例如：
from fastapi import FastAPI
from pydantic import BaseModel
app = FastAPI()
class Message(BaseModel):
detail: str
@app.get("/messages/", response_model=Message)
def get_message():
return {"detail": "This is a sample message"}此外，还可以对响应模型进行过滤，只返回部分字段。通过response_model_exclude_unset、response_model_exclude_defaults等参数实现
2.3 响应内容类型与媒体类型
默认情况下，FastAPI 返回 JSON 格式的响应。但在某些场景下，可能需要返回其他格式的数据，如 HTML、XML、二进制数据等。通过Response类及其子类，可以灵活设置响应的内容类型和媒体类型
返回 HTML 内容：
from fastapi import FastAPI, Response app = FastAPI() @app.get("/html/", response_class=Response) def get_html(): html_content = "<html><body><h1>Hello, FastAPI!</h1></body></html>" return Response(content=html_content, media_type="text/html")返回二进制数据（如图片）：
from fastapi import FastAPI, Response app = FastAPI() @app.get("/html/", response_class=Response) def get_html(): html_content = "<html><body><h1>Hello, FastAPI!</h1></body></html>" return Response(content=html_content, media_type="text/html")
三、实战案例：构建一个完整的用户管理 API
为了更直观地展示 FastAPI 请求与响应机制的实际应用，我们构建一个包含用户注册、登录、信息获取和更新功能的用户管理 API
from fastapi import FastAPI, HTTPException, Depends
from pydantic import BaseModel
from typing import Optional
from passlib.context import CryptContext
from jose import JWTError, jwt
# 模拟数据库
fake_users_db = {
"johndoe": {
"username": "johndoe",
"full_name": "John Doe",
"email": "johndoe@example.com",
"hashed_password": "$2b$12$EixZaYVK1fsbw1ZfbX3OXePaWxn96p36WQoeG6Lruj3vjPGga31lW", # 密码: "secret"
"disabled": False
}
}
app = FastAPI()
# 密码加密
pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
# JWT密钥和算法
SECRET_KEY = "09d25e094faa6ca2556c818166b7a9563b93f7099f6f0f4caa6cf63b88e8d3e7"
ALGORITHM = "HS256"
# 用户基础模型
class User(BaseModel):
username: str
email: Optional[str] = None
full_name: Optional[str] = None
disabled: Optional[bool] = None
# 用户创建模型
class UserCreate(User):
password: str
# 用户登录模型
class UserLogin(BaseModel):
username: str
password: str
# 令牌模型
class Token(BaseModel):
access_token: str
token_type: str
# 数据库中获取用户
def get_user(db, username: str):
if username in db:
user_dict = db[username]
return User(**user_dict)
# 验证密码
def verify_password(plain_password, hashed_password):
return pwd_context.verify(plain_password, hashed_password)
# 获取哈希密码
def get_password_hash(password):
return pwd_context.hash(password)
# 生成令牌
def create_access_token(data: dict):
to_encode = data.copy()
encoded_jwt = jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)
return encoded_jwt
# 用户注册
@app.post("/users/")
def create_new_user(user: UserCreate):
hashed_password = get_password_hash(user.password)
fake_users_db[user.username] = {
"username": user.username,
"email": user.email,
"full_name": user.full_name,
"hashed_password": hashed_password,
"disabled": user.disabled
}
return User(**fake_users_db[user.username])
# 用户登录
@app.post("/login/", response_model=Token)
def login_for_access_token(user: UserLogin):
user_in_db = get_user(fake_users_db, user.username)
if not user_in_db or not verify_password(user.password, user_in_db.hashed_password):
raise HTTPException(
status_code=401,
detail="Incorrect username or password",
headers={"WWW-Authenticate": "Bearer"},
)
access_token = create_access_token(data={"sub": user.username})
return {"access_token": access_token, "token_type": "bearer"}
# 获取用户信息
@app.get("/users/me/", response_model=User)
def read_users_me(current_user: User = Depends(get_current_active_user)):
return current_user
# 更新用户信息
@app.put("/users/me/")
def update_user_me(user: User, current_user: User = Depends(get_current_active_user)):
if current_user.username in fake_users_db:
fake_users_db[current_user.username].update(user.dict())
return User(**fake_users_db[current_user.username])
raise HTTPException(status_code=404, detail="User not found")在这个案例中，综合运用了请求参数的各种类型，包括请求体参数用于接收用户注册、登录和更新的数据；同时，通过设置响应状态码、定义响应模型，实现了规范且符合业务需求的响应返回，充分展示了 FastAPI 请求与响应机制的强大功能