在 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

    这里nameage是查询参数,当客户端发送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}

    此代码实现了一个简单的登录接口,接收表单提交的usernamepassword数据

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_unsetresponse_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 请求与响应机制的强大功能

最后修改:2025 年 06 月 05 日
如果觉得我的文章对你有用,请随意赞赏