配置API

多个装饰器装饰一个视图函数

如果一个视图函数需要同时支持多个请求地址的访问，可以使用多个装饰器装饰同一个视图函数。示例：

@app.get("/")
@app.get("/index")
def root():
    return {"msg": "hello world"}

一个URL配置多个HTTP请求方法

通过@app.api_route支持配置视图函数使用不同的HTTP请求方法

@app.get("/", 
         tags=["test"],
         summary="root url",
         description="fastapi quickstart",
         response_class=JSONResponse)
@app.api_route("/index", methods=["GET", "POST"])
def index():
    return {"msg": "Hello World"}

处理路径参数

app = FastAPI()

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

• 声明路径参数的类型

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

如果存在@app.get("/items/item_id")和@app.get("/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": "中国"}
    elif cityname == CityEnum.shanghai:
        return {"city": "上海", "country": "中国"}
    elif cityname == CityEnum.guangzhou:
        return {"city": "广州", "country": "中国"}
    elif cityname == CityEnum.shenzhen:
        return {"city": "深圳", "country": "中国"}
    else:
        return {"city": "未知", "country": "未知"}

处理查询参数

以下示例中为api设置两个查询参数，并指定默认值。如果不指定默认值，参数将为必需。

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

除了设置默认值，还可以设置参数为可选。

from typing import Union
from fastapi import FastAPI

app = FastAPI()

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

处理JSON请求体参数

这里使用路由组来组织代码，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)

if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app="main:app", host="127.0.0.1", port=8000)

然后创建myfastapi/models和myfastapi/routes文件夹，在两个文件下创建内容空白的__init__.py文件。

简单键值请求体

请求体示例：

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

编辑myfastapi/models/reqresp.py

from pydantic import BaseModel
from typing import Optional

class ReqBody1(BaseModel):
    username: str
    password: str
    age: int
    address: Optional[str] = None

编辑myfastapi/routes/jsonbody.py

from fastapi import APIRouter
from models import reqresp
from starlette.responses import JSONResponse

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,
        }
    })

测试响应结果

{
  "code": 200,
  "data": {
    "username": "string",
    "age": 0
  }
}

也可以使用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 = Body(default=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

包含列表的多层嵌套

请求体

{
  "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] # 列表类型. from typing import List
    groupLabels: Req4_grouplabels
    commonLabels: Req4_commonlabels
    externalURL: str
    version: str
    groupKey: str
    truncatedAlerts: int
    tags: Set[str] = [] # 集合类型, 默认为空 from typing import Set

编辑routes/jsonbody.py

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

处理表单请求参数

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 类接收到的内容是文件字节，无法读取到文件名等信息，一般用于上传小文件。

在异步视图中，通过引入aiofiles来异步写入文件。

python -m pip install aiofiles

示例代码

from fastapi import File, UploadFile
from typing import List

@router.post("/file1", summary="处理文件上传请求, File形式")
def filetest1(file: bytes = File(...)):
    """
    使用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类, bytes方式同步接收多文件")
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(...)):
    rst = {
        "filename": file.filename,
        "content_type": file.content_type,
    }
    content = await file.read()
    with open(f"{file.filename}", "wb") as f:
        f.write(content)
    return rst

@router.post("/file5", summary="基于UploadFile类接收多文件上传")
async def filetest5(files: List[UploadFile] = File(...)):
    return {filename: filesize for filename,filesize in zip([f1.filename for f1 in files],[len(await f2.read()) for f2 in files])}

处理请求头参数

fastapi提供了Header类用于请求头参数解析读取

from fastapi import Header
from typing import Optional, List

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

python的变量名中不能用短横杠，而浏览器传递参数用的都是短横杠，因此需要用到Header类的convert_underscores参数，其意为：如果在视图函数中声明的请求头参数使用了下划线命名法，那么是否对下划线进行转换。True表示转换，False表示不转换。

处理cookie参数

from fastapi import Cookie
from starlette.responses import Response
from typing import Optional

@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: Optional[str] = Cookie(None)):
    return {"cookie1": cookie1}