请求处理

请求处理是 FastAPI 的核心功能之一。本章将详细介绍如何处理各种类型的请求参数，包括路径参数、查询参数、请求体、表单数据、文件上传等。

参数识别规则

FastAPI 通过以下规则自动识别参数来源：

参数类型	识别规则
路径参数	参数名出现在路由路径中，如 `/items/{item_id}`
查询参数	单一类型（`int`、`str`、`float`、`bool` 等），不在路径中
请求体	Pydantic 模型类型
表单数据	使用 `Form()` 声明
文件	使用 `File()` 声明

理解这些规则，能让你更清晰地组织 API 参数。

路径参数

路径参数是 URL 路径的一部分，用于标识资源。例如 /users/123 中的 123 就是路径参数。

基本使用

from fastapi import FastAPI

app = FastAPI()

@app.get("/items/{item_id}")
async def read_item(item_id: int):
    # item_id 会自动从字符串转换为整数
    return {"item_id": item_id}

访问 /items/42，返回：

{"item_id": 42}

注意返回值是整数 42，而不是字符串 "42"。FastAPI 根据类型注解自动进行类型转换。

类型转换与验证

如果传入的值无法转换为声明的类型，FastAPI 会返回清晰的错误信息：

@app.get("/items/{item_id}")
async def read_item(item_id: int):
    return {"item_id": item_id}

访问 /items/foo，返回 422 错误：

{
  "detail": [
    {
      "type": "int_parsing",
      "loc": ["path", "item_id"],
      "msg": "Input should be a valid integer, unable to parse string as an integer",
      "input": "foo"
    }
  ]
}

这种即时反馈极大提高了调试效率。

路径参数类型

from fastapi import FastAPI

app = FastAPI()

# 整数类型
@app.get("/items/{item_id}")
async def read_item(item_id: int):
    return {"item_id": item_id, "type": "int"}

# 浮点数类型
@app.get("/price/{amount}")
async def read_price(amount: float):
    return {"amount": amount, "type": "float"}

# 字符串类型（默认）
@app.get("/users/{username}")
async def read_user(username: str):
    return {"username": username, "type": "str"}

# 布尔类型
@app.get("/flag/{enabled}")
async def read_flag(enabled: bool):
    return {"enabled": enabled, "type": "bool"}

# 路径类型（包含斜杠）
@app.get("/files/{file_path:path}")
async def read_file(file_path: str):
    # 访问 /files/home/user/document.txt
    # file_path = "home/user/document.txt"
    return {"file_path": file_path}

路径参数验证

使用 Path 函数添加验证规则和元数据：

from typing import Annotated
from fastapi import FastAPI, Path

app = FastAPI()

@app.get("/items/{item_id}")
async def read_item(
    item_id: Annotated[int, Path(
        title="商品 ID",
        description="商品的唯一标识符，必须是正整数",
        gt=0,        # 大于 0
        le=10000,    # 小于等于 10000
        example=123  # 文档中的示例值
    )]
):
    return {"item_id": item_id}

数值验证参数

参数	含义	示例
`gt`	大于	`gt=0` 值必须大于 0
`ge`	大于等于	`ge=1` 值必须大于等于 1
`lt`	小于	`lt=100` 值必须小于 100
`le`	小于等于	`le=1000` 值必须小于等于 1000
`multiple_of`	是某数的倍数	`multiple_of=5` 必须是 5 的倍数

字符串验证参数

@app.get("/users/{username}")
async def read_user(
    username: Annotated[str, Path(
        min_length=3,
        max_length=20,
        pattern=r'^[a-zA-Z0-9_]+$',  # 正则表达式
        description="用户名，只能包含字母、数字和下划线"
    )]
):
    return {"username": username}

预定义值（枚举）

当路径参数只能是几个固定值之一时，使用枚举：

from enum import Enum
from fastapi import FastAPI

class ModelName(str, Enum):
    alexnet = "alexnet"
    resnet = "resnet"
    lenet = "lenet"

app = FastAPI()

@app.get("/models/{model_name}")
async def get_model(model_name: ModelName):
    # 可以使用枚举成员比较
    if model_name is ModelName.alexnet:
        return {"model": model_name, "message": "Deep Learning FTW!"}
    
    # 也可以使用值比较
    if model_name.value == "lenet":
        return {"model": model_name, "message": "LeCNN all the images"}
    
    return {"model": model_name, "message": "Have some residuals"}

枚举值会自动显示在 API 文档的下拉列表中，方便用户选择。

路由顺序很重要

FastAPI 按定义顺序匹配路由，更具体的路由应放在前面：

# 正确顺序
@app.get("/users/me")
async def read_me():
    return {"user": "当前用户"}

@app.get("/users/{user_id}")
async def read_user(user_id: int):
    return {"user_id": user_id}

# 错误顺序：/users/me 会被 {user_id} 匹配
# @app.get("/users/{user_id}")
# async def read_user(user_id: int):
#     return {"user_id": user_id}
# 
# @app.get("/users/me")  # 永远不会被匹配到！
# async def read_me():
#     return {"user": "当前用户"}

查询参数

查询参数是 URL 中 ? 后面的键值对，用于过滤、排序、分页等。

基本使用

from fastapi import FastAPI

app = FastAPI()

fake_items_db = [
    {"item_name": "Foo"},
    {"item_name": "Bar"},
    {"item_name": "Baz"}
]

@app.get("/items/")
async def read_items(skip: int = 0, limit: int = 10):
    # /items/?skip=0&limit=10
    return fake_items_db[skip : skip + limit]

访问 /items/?skip=0&limit=10，返回前 10 条数据。访问 /items/?skip=20，limit 使用默认值 10。

可选参数

使用 None 作为默认值声明可选参数：

from typing import Annotated

@app.get("/items/{item_id}")
async def read_item(
    item_id: str,
    q: Annotated[str | None, "搜索关键词"] = None,
    short: bool = False
):
    item = {"item_id": item_id}
    if q:
        item.update({"q": q})
    if not short:
        item.update({
            "description": "这是一个很棒的商品，有很长的描述"
        })
    return item

访问 /items/123 → q 为 None，short 为 False
访问 /items/123?q=search → q 为 "search"
访问 /items/123?short=true → short 为 True

必需参数

不设置默认值，参数即为必需：

@app.get("/items/{item_id}")
async def read_item(item_id: str, needy: str):
    # needy 是必需的查询参数
    return {"item_id": item_id, "needy": needy}

访问 /items/foo?needy=sooooneedy 才能成功，否则返回 422 错误。

查询参数验证

使用 Query 函数添加验证：

from typing import Annotated
from fastapi import FastAPI, Query

app = FastAPI()

@app.get("/items/")
async def read_items(
    q: Annotated[str | None, Query(
        title="搜索关键词",
        description="用于搜索商品的关键词，长度 3-50",
        min_length=3,
        max_length=50,
        pattern=r'^[a-zA-Z0-9\s]+$'
    )] = None
):
    return {"q": q}

列表查询参数

同一个参数名可以出现多次，接收为列表：

@app.get("/items/")
async def read_items(
    q: Annotated[list[str] | None, Query(
        title="搜索关键词列表",
        description="可以传递多个搜索关键词"
    )] = None
):
    # /items/?q=a&q=b&q=c
    # q = ["a", "b", "c"]
    return {"q": q}

也可以使用默认值：

@app.get("/items/")
async def read_items(
    q: Annotated[list[str], Query()] = ["foo", "bar"]
):
    return {"q": q}

布尔类型转换

布尔类型参数支持多种输入形式：

@app.get("/items/{item_id}")
async def read_item(item_id: str, short: bool = False):
    return {"item_id": item_id, "short": short}

以下值都会被转换为 True：

short=1
short=True
short=true
short=on
short=yes

以下值会被转换为 False：

short=0
short=False
short=false
short=off
short=no

请求体

请求体是客户端发送给服务器的数据，通常用于创建或更新资源。

Pydantic 模型

使用 Pydantic 模型声明请求体：

from fastapi import FastAPI
from pydantic import BaseModel

app = FastAPI()

class Item(BaseModel):
    name: str
    description: str | None = None
    price: float
    tax: float | None = None

@app.post("/items/")
async def create_item(item: Item):
    # FastAPI 自动将 JSON 转换为 Item 对象
    return item

请求示例：

curl -X POST "http://localhost:8000/items/" \
  -H "Content-Type: application/json" \
  -d '{"name": "商品A", "price": 99.9}'

模型属性访问

在函数内部可以直接访问模型属性：

@app.post("/items/")
async def create_item(item: Item):
    item_dict = item.model_dump()
    
    # 根据条件添加额外字段
    if item.tax is not None:
        price_with_tax = item.price + item.tax
        item_dict.update({"price_with_tax": price_with_tax})
    
    return item_dict

混合参数

路径参数、查询参数、请求体可以同时使用：

@app.put("/items/{item_id}")
async def update_item(
    item_id: int,           # 路径参数
    item: Item,             # 请求体
    q: str | None = None    # 查询参数
):
    result = {"item_id": item_id, **item.model_dump()}
    if q:
        result.update({"q": q})
    return result

FastAPI 会自动识别：

item_id 出现在路径中 → 路径参数
item 是 Pydantic 模型 → 请求体
q 是单一类型且有默认值 → 查询参数

多个请求体

可以同时接收多个请求体对象：

from fastapi import FastAPI
from pydantic import BaseModel

app = FastAPI()

class Item(BaseModel):
    name: str
    description: str | None = None
    price: float

class User(BaseModel):
    username: str
    full_name: str | None = None

@app.put("/items/{item_id}")
async def update_item(item_id: int, item: Item, user: User):
    return {"item_id": item_id, "item": item, "user": user}

请求格式：

{
    "item": {
        "name": "Foo",
        "description": "A description",
        "price": 45.2
    },
    "user": {
        "username": "johndoe",
        "full_name": "John Doe"
    }
}

单值请求体

使用 Body 函数将单个值作为请求体：

from typing import Annotated
from fastapi import FastAPI, Body

app = FastAPI()

@app.put("/items/{item_id}")
async def update_item(
    item_id: int,
    importance: Annotated[int, Body(gt=0)]
):
    return {"item_id": item_id, "importance": importance}

请求格式：

{
    "importance": 5
}

嵌入单个请求体

默认情况下，单个请求体直接作为 JSON 体。如果需要包裹在键中：

from typing import Annotated
from fastapi import FastAPI, Body

app = FastAPI()

class Item(BaseModel):
    name: str
    description: str | None = None
    price: float

@app.put("/items/{item_id}")
async def update_item(
    item_id: int,
    item: Annotated[Item, Body(embed=True)]
):
    return {"item_id": item_id, "item": item}

请求格式变为：

{
    "item": {
        "name": "Foo",
        "price": 45.2
    }
}

表单数据

处理 HTML 表单提交的数据：

from typing import Annotated
from fastapi import FastAPI, Form

app = FastAPI()

@app.post("/login/")
async def login(
    username: Annotated[str, Form()],
    password: Annotated[str, Form()]
):
    return {"username": username}

表单数据使用 application/x-www-form-urlencoded 编码，与 JSON 不同。

文件上传

处理文件上传需要使用 File：

from typing import Annotated
from fastapi import FastAPI, File, UploadFile

app = FastAPI()

# 简单的文件上传（字节）
@app.post("/files/")
async def create_file(
    file: Annotated[bytes, File()]
):
    return {"file_size": len(file)}

# 使用 UploadFile（推荐）
@app.post("/uploadfile/")
async def create_upload_file(
    file: UploadFile
):
    content = await file.read()
    return {
        "filename": file.filename,
        "content_type": file.content_type,
        "size": len(content)
    }

bytes vs UploadFile

特性	`bytes`	`UploadFile`
存储位置	内存	临时文件（适合大文件）
异步支持	否	是
文件信息	无	filename, content_type
适用场景	小文件	大文件、需要文件信息

多文件上传

from typing import Annotated
from fastapi import FastAPI, File, UploadFile

app = FastAPI()

@app.post("/uploadfiles/")
async def create_upload_files(
    files: Annotated[list[UploadFile], File(description="多个文件")]
):
    return {
        "filenames": [file.filename for file in files]
    }

文件与表单混合

from typing import Annotated
from fastapi import FastAPI, File, Form, UploadFile

app = FastAPI()

@app.post("/files/")
async def create_file(
    file: Annotated[bytes, File()],
    fileb: Annotated[UploadFile, File()],
    token: Annotated[str, Form()]
):
    return {
        "file_size": len(file),
        "fileb_content_type": fileb.content_type,
        "token": token
    }

请求对象

直接访问原始请求对象：

from fastapi import FastAPI, Request

app = FastAPI()

@app.get("/items/")
async def read_items(request: Request):
    # 客户端信息
    client_host = request.client.host
    
    # 请求头
    user_agent = request.headers.get("user-agent")
    
    # 查询参数
    query_params = dict(request.query_params)
    
    # 路径参数
    path_params = request.path_params
    
    return {
        "client_host": client_host,
        "user_agent": user_agent,
        "query_params": query_params,
        "path_params": path_params
    }

获取原始请求体

from fastapi import FastAPI, Request

app = FastAPI()

@app.post("/items/")
async def read_item(request: Request):
    # 获取原始 JSON
    json_data = await request.json()
    
    # 获取原始字节
    body = await request.body()
    
    # 获取表单数据
    form_data = await request.form()
    
    return {"json": json_data, "body_size": len(body)}

from typing import Annotated
from fastapi import FastAPI, Cookie

app = FastAPI()

@app.get("/items/")
async def read_items(
    session_id: Annotated[str | None, Cookie()] = None
):
    return {"session_id": session_id}

from typing import Annotated
from fastapi import FastAPI, Header

app = FastAPI()

@app.get("/items/")
async def read_items(
    user_agent: Annotated[str | None, Header()] = None,
    x_token: Annotated[str | None, Header(alias="X-Token")] = None
):
    return {
        "User-Agent": user_agent,
        "X-Token": x_token
    }

注意：HTTP 头通常使用连字符（如 User-Agent），但在 Python 中使用下划线（user_agent）。FastAPI 会自动转换。

参数元数据

为参数添加文档元数据：

from typing import Annotated
from fastapi import FastAPI, Query

app = FastAPI()

@app.get("/items/")
async def read_items(
    q: Annotated[str | None, Query(
        title="搜索关键词",
        description="用于在商品名称和描述中搜索的关键词",
        min_length=3,
        max_length=50,
        deprecated=True,  # 标记为已弃用
        examples=["phone", "laptop", "book"]  # 示例值
    )] = None
):
    return {"q": q}

完整示例：商品 API

from typing import Annotated
from fastapi import FastAPI, Path, Query, Body, HTTPException, status
from pydantic import BaseModel, Field

app = FastAPI()

# 模型定义
class Item(BaseModel):
    name: str = Field(min_length=1, max_length=100)
    description: str | None = Field(None, max_length=500)
    price: float = Field(gt=0, description="价格必须大于 0")
    tax: float | None = Field(None, ge=0)
    
    model_config = {
        "json_schema_extra": {
            "examples": [{
                "name": "商品名称",
                "description": "商品描述",
                "price": 99.9,
                "tax": 9.9
            }]
        }
    }

class ItemUpdate(BaseModel):
    name: str | None = Field(None, min_length=1, max_length=100)
    description: str | None = None
    price: float | None = Field(None, gt=0)
    tax: float | None = Field(None, ge=0)

# 模拟数据库
items_db: dict[int, Item] = {}

# 创建商品
@app.post("/items/", status_code=status.HTTP_201_CREATED)
async def create_item(item: Annotated[Item, Body(embed=True)]):
    item_id = len(items_db) + 1
    items_db[item_id] = item
    return {"id": item_id, "item": item}

# 获取商品列表
@app.get("/items/")
async def read_items(
    skip: Annotated[int, Query(ge=0)] = 0,
    limit: Annotated[int, Query(ge=1, le=100)] = 10,
    q: Annotated[str | None, Query(min_length=1, max_length=50)] = None
):
    items = list(items_db.values())[skip:skip + limit]
    
    if q:
        items = [item for item in items if q.lower() in item.name.lower()]
    
    return {"items": items, "total": len(items_db)}

# 获取单个商品
@app.get("/items/{item_id}")
async def read_item(
    item_id: Annotated[int, Path(gt=0, description="商品 ID")]
):
    if item_id not in items_db:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND,
            detail=f"商品 {item_id} 不存在"
        )
    return items_db[item_id]

# 更新商品
@app.put("/items/{item_id}")
async def update_item(
    item_id: Annotated[int, Path(gt=0)],
    item: Item,
    partial: Annotated[bool, Query(description="是否部分更新")] = False
):
    if item_id not in items_db:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND,
            detail=f"商品 {item_id} 不存在"
        )
    
    if partial:
        # 部分更新
        existing = items_db[item_id]
        update_data = item.model_dump(exclude_unset=True)
        updated_item = existing.model_copy(update=update_data)
        items_db[item_id] = updated_item
    else:
        # 完整更新
        items_db[item_id] = item
    
    return items_db[item_id]

# 删除商品
@app.delete("/items/{item_id}", status_code=status.HTTP_204_NO_CONTENT)
async def delete_item(
    item_id: Annotated[int, Path(gt=0)]
):
    if item_id not in items_db:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND,
            detail=f"商品 {item_id} 不存在"
        )
    
    del items_db[item_id]

小结

本章我们学习了：

参数识别规则：FastAPI 如何自动识别参数来源
路径参数：URL 路径中的变量，支持类型转换和验证
查询参数：URL 中的可选参数，用于过滤和分页
请求体：Pydantic 模型处理 JSON 数据
表单和文件：处理表单提交和文件上传
请求对象：直接访问原始请求信息
Cookie 和 Header：处理 Cookie 和 HTTP 头

FastAPI 参数处理的优势：

声明式语法，代码简洁
自动类型转换和验证
清晰的错误信息
自动生成 API 文档

练习

创建一个用户 API，支持路径参数（用户 ID）、查询参数（搜索关键词）和请求体（用户数据）
实现一个文件上传接口，支持同时上传文件和表单数据
创建一个分页查询接口，包含 page、page_size 和排序参数
使用枚举定义一个状态参数，限制可选值

参数识别规则​

路径参数​

基本使用​

类型转换与验证​

路径参数类型​

路径参数验证​

数值验证参数​

字符串验证参数​

预定义值（枚举）​

路由顺序很重要​

查询参数​

基本使用​

可选参数​

必需参数​

查询参数验证​

列表查询参数​

布尔类型转换​

请求体​

Pydantic 模型​

模型属性访问​

混合参数​

多个请求体​

单值请求体​

嵌入单个请求体​

表单数据​

文件上传​

bytes vs UploadFile​

多文件上传​

文件与表单混合​

请求对象​

获取原始请求体​

Cookie 和 Header​

Cookie​

Header​

参数元数据​

完整示例：商品 API​

小结​

练习​

参数识别规则

路径参数

基本使用

类型转换与验证

路径参数类型

路径参数验证

数值验证参数

字符串验证参数

预定义值（枚举）

路由顺序很重要

查询参数

基本使用

可选参数

必需参数

查询参数验证

列表查询参数

布尔类型转换

请求体

Pydantic 模型

模型属性访问

混合参数

多个请求体

单值请求体

嵌入单个请求体

表单数据

文件上传

bytes vs UploadFile

多文件上传

文件与表单混合

请求对象

获取原始请求体

Cookie 和 Header

Cookie

Header

参数元数据

完整示例：商品 API

小结

练习