解析请求

前言

FastAPI 对各类请求参数的解析：路径参数、查询参数、请求体（JSON）、表单、文件上传、请求头、Cookie，以及直接使用 Request 对象。各参数类型的校验器（Path / Query / Form / Body / Header / Cookie）详细参数见 参数校验。

Request 对象

当需要直接解析原始请求报文时，可以把 Request 对象显式声明到视图函数参数上：

from fastapi import Request


@router.get("/reqobj", summary="获取request对象")
async def get_req_obj(request: Request):
    form_data = await request.form()   # form() 是协程函数
    body_data = await request.body()   # body() 是协程函数
    return {
        "url": request.url,
        "base_url": request.base_url,
        "method": request.method,
        "headers": dict(request.headers),
        "cookies": dict(request.cookies),
        "form_data": dict(form_data),
        "body_data": body_data,
        "client_host": request.client.host if request.client else None,
        "json_data": await request.json() if body_data else None,  # json() 是协程函数
    }

注意

await request.form()、await request.body()、await request.json() 都是协程方法，且请求体只能读取一次，读取后再次读取为空。

解析路径参数

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

声明路径参数的类型后，FastAPI 会自动校验并转换：

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

如果同时存在静态路由 /items/item_id 和动态路由 /items/{item_id}，访问 /items/item_id 时按照先定义者优先的顺序匹配，所以一般把静态路由定义在前面。

含有文件路径的路径参数

如果路径参数是一段文件路径，需要将参数类型声明为 path：

@app.get("/{filepath:path}")
async def filepath_param(filepath: str):
    return {"filepath": filepath}

枚举路径参数

路径参数有几个预设值时，可以用枚举类限制：

from enum import Enum


class CityEnum(str, Enum):
    beijing = "beijing"
    shanghai = "shanghai"
    guangzhou = "guangzhou"
    shenzhen = "shenzhen"


@router.get("/city/{cityname}", summary="枚举城市名称")
async def get_city_info(cityname: CityEnum):
    if cityname == CityEnum.beijing:
        return {"city": "北京", "country": "中国"}
    if cityname == CityEnum.shanghai:
        return {"city": "上海", "country": "中国"}
    if cityname == CityEnum.guangzhou:
        return {"city": "广州", "country": "中国"}
    return {"city": "深圳", "country": "中国"}

解析查询参数

声明查询参数并指定默认值；不指定默认值即为必填：

from fastapi import FastAPI

app = FastAPI()

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


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

请求示例：

curl 'http://127.0.0.1:8000/items/?skip=0&limit=10'

可选参数：

@app.get("/items/{item_id}")
async def read_item(item_id: str, q: str | None = None):
    if q:
        return {"item_id": item_id, "q": q}
    return {"item_id": item_id}

解析 JSON 请求体参数

这里用路由组组织代码。目录结构：

myfastapi/
├── main.py
├── models/
│   ├── __init__.py
│   └── reqresp.py
└── routes/
    ├── __init__.py
    └── jsonbody.py

myfastapi/main.py：

from fastapi import FastAPI

app = FastAPI(
    title="Hello FastAPI",
    description="Quickstart of FastAPI",
    version="0.1.0",
)

from routes import jsonbody

app.include_router(jsonbody.router)

简单键值请求体

请求体示例：

{
  "username": "string",
  "password": "string",
  "age": 0,
  "address": "string"
}

models/reqresp.py：

from pydantic import BaseModel


class ReqBody1(BaseModel):
    username: str
    password: str
    age: int
    address: str | None = None

routes/jsonbody.py：

from fastapi import APIRouter
from starlette.responses import JSONResponse

from models import reqresp

router = APIRouter(prefix="/jsonbody", tags=["解析json请求体"])


@router.post("/test1", summary="测试解析json请求1")
def jtest1(req: reqresp.ReqBody1):
    return JSONResponse(
        {
            "code": 200,
            "data": {
                "username": req.username,
                "age": req.age,
            },
        }
    )

也可以使用 fastapi.Body 类（单字段级别的请求体声明）：

from fastapi import Body


@router.post("/test2", summary="测试解析json请求2")
def jtest2(
    username: str = Body(...),       # 必填
    password: str = Body(...),       # 必填
    age: int = Body(..., ge=0),      # 大于等于 0
    address: str | None = Body(None),  # 可选字段
):
    return {"username": username, "age": age}

二层嵌套

请求体示例：

{
  "user": {
    "username": "string",
    "sex": "string",
    "age": 0
  },
  "job": {
    "name": "string",
    "job_no": 0,
    "dept_no": 0
  },
  "company": "string"
}

models/reqresp.py：

class ReqBody3_User(BaseModel):
    username: str
    sex: str
    age: int


class ReqBody3_Job(BaseModel):
    name: str
    job_no: int
    dept_no: int


class ReqBody3(BaseModel):
    user: ReqBody3_User
    job: ReqBody3_Job
    company: str

routes/jsonbody.py：

@router.post("/test3", summary="测试解析json请求3")
def jtest3(req: reqresp.ReqBody3):
    return req

包含列表的多层嵌套

请求体示例（Prometheus Alertmanager Webhook 结构）：

{
  "receiver": "string",
  "status": "string",
  "alerts": [
    {
      "status": "string",
      "labels": {
        "alertname": "string",
        "instance": "string",
        "job": "string",
        "severity": "string"
      },
      "annotations": {
        "description": "string",
        "summary": "string"
      },
      "startsAt": "string",
      "endsAt": "string",
      "generatorURL": "string",
      "fingerprint": "string"
    }
  ],
  "groupLabels": {
    "alertname": "string"
  },
  "commonLabels": {
    "alertname": "string",
    "instance": "string",
    "job": "string",
    "severity": "string"
  },
  "externalURL": "string",
  "version": "string",
  "groupKey": "string",
  "truncatedAlerts": 0,
  "tags": []
}

models/reqresp.py：

class Req4_alerts_labels(BaseModel):
    alertname: str
    instance: str
    job: str
    severity: str


class Req4_alerts_annotations(BaseModel):
    description: str
    summary: str


class Req4_alerts(BaseModel):
    status: str
    labels: Req4_alerts_labels
    annotations: Req4_alerts_annotations
    startsAt: str
    endsAt: str
    generatorURL: str
    fingerprint: str


class Req4_grouplabels(BaseModel):
    alertname: str


class Req4_commonlabels(BaseModel):
    alertname: str
    instance: str
    job: str
    severity: str


class Req4_commonAnnotations(BaseModel):
    description: str
    summary: str


class ReqBody4(BaseModel):
    receiver: str
    status: str
    alerts: list[Req4_alerts]        # 列表类型
    groupLabels: Req4_grouplabels
    commonLabels: Req4_commonlabels
    externalURL: str
    version: str
    groupKey: str
    truncatedAlerts: int
    tags: list[str] = []             # 列表类型，默认为空

routes/jsonbody.py：

@router.post("/test4", summary="测试解析json请求4")
def jtest4(req: reqresp.ReqBody4):
    return req

解析未知结构的 JSON 请求体

不确定请求体结构时，可以直接用 Request 读取原始 JSON：

from fastapi import FastAPI, Request
from fastapi.responses import PlainTextResponse
import uvicorn

app = FastAPI()


@app.post("/alert")
async def grafana_alert(request: Request):
    data = await request.json()  # 返回的是 python dict
    print(data)
    return PlainTextResponse(status_code=200, content="ok")

解析表单请求参数

FastAPI 处理表单需要安装第三方库：

python -m pip install python-multipart

from fastapi import Form


@router.post("/form", summary="处理表单请求参数")
def formtest1(
    username: str = Form(..., min_length=6, max_length=12),
    password: str = Form(..., min_length=6, max_length=12),
):
    return {"username": username, "password": password}

接收文件上传

FastAPI 提供了 File 和 UploadFile 两个类处理文件上传。File 接收到的内容是文件字节，拿不到文件名等信息，一般用于上传小文件；UploadFile 以文件对象的形式提供文件名、Content-Type，并支持异步读取，推荐使用。

python -m pip install aiofiles

from fastapi import File, UploadFile


@router.post("/file1", summary="File形式, bytes接收")
def filetest1(file: bytes = File(...)):
    """文件内容以 bytes 写入内存，适合小文件"""
    with open("test.jpg", "wb") as f:
        f.write(file)
    return {"msg": "上传成功", "filesize": len(file)}


import aiofiles


@router.post("/file2", summary="File形式, 异步接收")
async def filetest2(file: bytes = File(...)):
    async with aiofiles.open("test.jpg", "wb") as f:
        await f.write(file)
    return {"msg": "上传成功", "filesize": len(file)}


@router.post("/file3", summary="File形式, 多文件")
def filetest3(files: list[bytes] = File(...)):
    return {"file_sizes": [len(file) for file in files]}


@router.post("/file4", summary="UploadFile形式")
async def filetest4(file: UploadFile = File(...)):
    content = await file.read()
    with open(file.filename, "wb") as f:
        f.write(content)
    return {
        "filename": file.filename,
        "content_type": file.content_type,
        "filesize": len(content),
    }


@router.post("/file5", summary="UploadFile形式, 多文件")
async def filetest5(files: list[UploadFile] = File(...)):
    return {f.filename: len(await f.read()) for f in files}

提示

UploadFile 默认会把内容写入临时文件，适合大文件；File 直接读入内存，适合小文件。

处理请求头参数

from fastapi import Header


@router.get("/header", summary="解析请求头参数")
async def headertest(
    user_agent: str | None = Header(None, convert_underscores=True),
    accept_encoding: str | None = Header(None, convert_underscores=True),
    accept: str | None = Header(None),
    accept_token: str = Header(..., convert_underscores=False),  # 自定义请求头
    x_token: list[str] | None = Header(None),                    # 多值的重名请求头
):
    return {
        "user_agent": user_agent,
        "accept_encoding": accept_encoding,
        "accept": accept,
        "accept_token": accept_token,
        "x_token": x_token,
    }

备注

Python 变量名中不能用短横杠，而 HTTP 头通常使用短横杠（如 X-Token）。convert_underscores 表示是否把参数名中的下划线转换为短横杠去匹配请求头：True 时 accept_encoding 匹配 accept-encoding；False 时按原名匹配（自定义头常需显式传 alias，如 Header(..., alias="X-Token")，更清晰）。

处理 Cookie 参数

from fastapi import Cookie
from starlette.responses import Response


@router.post("/cookie", summary="设置cookie")
def set_cookie(response: Response, username: str):
    response.set_cookie(key="cookie1", value=f"{username}.2023")
    return {"msg": "set cookie success"}


@router.get("/cookie", summary="获取cookie")
async def get_cookie(cookie1: str | None = Cookie(None)):
    return {"cookie1": cookie1}