FastAPI

在阅读这篇文章前建议先了解RESTful API相关概念，可查看本站的“RESTfulAPI”这篇文章。

FastAPI是什么？

是一种在Python语言中实现RESTfulAPI设计的API构建方式。
不想写API文档？它能自动生成API文档。
要实现IO比较频繁的场景？它支持异步编程，处理IO密集小能手。
对参数验证不厌其烦？它提供基于Python的输入验证和代码提示
担心它的稳定性？它能在生产环境中使用。

FastAPI能做什么？

FastAPI的特点决定了它的应用场景：

构建API后端：用于在Python中构建RESTful API，前后端分离的Web应用，俗称写接口
微服务架构：可作为微服务的后端框架，支持快速开发和部署
数据处理API：适用于处理数据，接受和返回JSON数据
实时通信：支持WebSocket，可用作实时通信

FastAPI要什么？

FastAPI依赖Python3.8及更高版本
使用FastAPI需要先安装fastapi包，使用pip命令：
1
pip install fastapi
需要一个ASGI服务器，生产环境可以使用 Uvicorn 或者 Hypercorn。
推荐FastAPI搭配Uvicorn，轻量支持异步并行，高效处理高并发，可以使用命令安装：
1
pip install "uvicorn[standard]"

必不可少的HelloWorld

创建一个main.py文件，写入下面内容：

from fastapi import FastAPI
import uvicorn

# 创建FastAPI应用实例
app = FastAPI()

# 根路由，返回简单响应
@app.get("/")
def read_root():
    return {"Hello": "World"}

在命令行使用命令启动：

1	uvicorn main:app

如果你是在开发环境，建议加上--reload参数：

1	uvicorn main:app --reload

--reload参数能帮你在修改代码并保存后自动重启服务。
要记住，在生产环境不要加–reload参数
main参数对应文件名main.py。
app参数对应代码中的app=FastAPI()这句生成的FastAPI对象。
本地启动的话默认服务在http://127.0.0.1:8000/,浏览器即可访问，得到返回结果{"Hello":"World"}
我们定义的处理函数read_root()返回一个字典，该字典将被 FastAPI 自动转换为 JSON 格式，并作为响应发送给客户端

复杂一点，带路径参数的

1
2
3

@app.get("/name/{my_name}")
def read_name(my_name: str):
    return {"my_name": my_name}

一般我们习惯于在参数前加上表示资源的uri前缀用于区分不同的服务，就像这里的name
{my_name}为路径参数，方法的参数名要与路径参数保持一致才能读到
例子：
请求url： http://127.0.0.1:8000/name/fjsi
响应体: {"my_name":"fjsi"}

丰富一点，参数类型加限制

from typing import Union

@app.get("/items/{item_id}")
def read_item(item_id: int, item_name: Union[str, None] = None):
    # item_id: 路径参数(整数)
    # item_name: 可选查询参数(字符串或None),不传就默认None
    return {"item_id": item_id, "item_name": item_name}

例子：
请求url： http://127.0.0.1:8000/items/1?item_name=fjsi
响应体： {"item_id":1,"item_name":"fjsi"}

item_id: int这种形式前面item_id是变量名与路径参数对应，int是固定的数据类型，fastapi会根据此规定对参数进行校验。
item_name: Union[str, None] = None表示item_name参数类型可为str或None，=None表示默认值为None，这样就实现了可选参数item_name。

想要发请求体

POST请求这种包含多个请求参数的使用请求体更好点，借助 Pydantic 来使用标准的 Python 类型声明请求体。

from pydantic import BaseModel

class Student(BaseModel):
    name: str
    age: int
    blog: Union[str, None] = None

@app.post("/student")
def create_student(student: Student):
    return {"name": student.name, "age": student.age, "blog": student.blog}

借助Postman发送POST请求，url：http://127.0.0.1:8000/student
请求体：

{
    "name":"fjsi",
    "age":18,
    "blog":"https://www.fjsi.top"
}

返回体：

{
    "name":"fjsi",
    "age":18,
    "blog":"https://www.fjsi.top"
}

借助 Pydantic 来使用标准的 Python 类型声明方式不仅可用于定义请求体，还可用于定义返回体。
关于Pydantic模型，详细内容见下面的 FastAPI Pydantic 模型 部分

不仅要请求体还要返回体

class StudentVo(BaseModel):
    student_name: str
    student_age: int

@app.post("/student")
def create_student(student: Student):
    return StudentVo(student_name=student.name, student_age=student.age)

这里定义了返回体的类型，限制了参数名，参数数量和参数类型。

使用 Header 和 Cookie 类型注解获取请求头和 Cookie 数据。

from fastapi import Header, Cookie

@app.get("/header-cookie/")
def read_item(content_type: str = Header("application/json"),user_agent: str = Header("PostmanRuntime/7.43.3"), session_token: str = Cookie(None)):
        return {"Content-Type": content_type,"User-Agent": user_agent, "Session-Token": session_token}

首先要接收请求头和Cookie就需要引入Header和Cookie
以content_type: str = Header("application/json")为例，content_type是请求头的名字，格式按这个标准来，只要参数名字正确就能从Header中读入。
Header()中可填入默认值，如果请求头中不包含此参数就会使用默认值，默认值也可以为None，但是不建议不填，如果不填，请求头不包含相应参数时会报错。

重定向路由

使用 RedirectResponse 实现重定向，将客户端请求重定向。

from fastapi.responses import RedirectResponse

@app.get("/header-cookie/")
def read_item(content_type: str = Header("application/json"),user_agent: str = Header("PostmanRuntime/7.43.3"), session_token: str = Cookie(None)):
        return {"Content-Type": content_type,"User-Agent": user_agent, "Session-Token": session_token}

@app.get("/redirect")
def redirect():
    return RedirectResponse(url="/header-cookie/")

访问/redirect uri就会转发到header-cookie上去，看上去好像有点多次一举，但其实在转发前可以在转发部分做很多预处理，例如在/redirect/中进行数据校验、数据预处理等，让/header-cookie/部分专心实现业务。

异常抛出和异常处理

使用 HTTPException 在你认为合适的地方抛出异常给请求方，返回自定义的状态码和详细信息

from fastapi import HTTPException

@app.get("/exception/{age}")
def read_item(age: int):
    if age < 0:
        raise HTTPException(status_code=400, detail="param 'age' is out of range")
    else:
        return {"age": age}

传入的age>=0时返回age，小于0时返回400报错，错误信息:{"detail": "param 'age' is out of range"}

FastAPI Pydantic 模型

我的请求有多个参数时，我应该如何将这多个参数管理并做类型验证？
Pydantic 就是一个解决此问题的工具，Pydantic是一个用于数据验证和序列化的 Python 模型库
用于定义请求体、响应体和其他数据模型，提供了强大的类型检查和自动文档生成功能

FastAPI 将自动验证传入的 JSON 数据是否符合模型的定义，并将其转换为 Item 类型的实例，这意味着你可以直接在方法中使用item.name的方式访问传入的name值，这个使用Pydantic模型来做数据验证和映射的方式有点类似java中定义一个实体专门对应参数，利用RESTfulAPI实现参数映射到实体上

要使用分为两个步骤：先定义Pydantic模型或者说是模板，然后使用定义好的Pydantic模型来做验证

from fastapi import FastAPI
from pydantic import BaseModel

app = FastAPI()

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

@app.post("/items/")
def create_item(item: Item):
    return item

更加严格的参数限制验证规则

from fastapi import FastAPI, Query

app = FastAPI()

@app.get("/items/")
def read_item(name: str = Query(..., max_length=10)):
    return {"name": name}

read_item()接受一个名为 name 的字符串查询参数。通过使用 Query 函数可以为查询参数指定更多的验证规则，如最大长度限制，当name的长度小于等于10就正常执行，大于10就给出错误提示。
同样的参数限制规则还有：

参数	作用	示例
`min_length`	最小长度	`Query(..., min_length=3)`
`regex`	正则表达式	`Query(..., regex="^[a-z]+$")`
`default`	默认值	`Query(default="hi")`
`alias`	参数别名	`Query(..., alias="item-query")`

示例：

from fastapi import Query

@app.get("/items/")
def read_item(
    name: str = Query(
        ...,
        min_length=3,
        max_length=10,
        regex="^[a-zA-Z]+$",  # 只允许字母
        example="Alice",       # OpenAPI 文档示例
    )
):
    return {"name": name}

使用 Pydantic 模型能够在服务启动后自动生成 FastAPI 生成交互式 API 文档。文档会包括模型的字段、类型、验证规则等信息，让开发者和 API 使用者能够清晰地了解如何正确使用 API。
API文档地址: http://host:port/docs

怎么生成交互文档？

FastAPI 提供了内置的交互式 API 文档，基于 OpenAPI 规范自动生成，支持 Swagger UI 和 ReDoc 两种交互式界面。
在运行 FastAPI 应用时，Uvicorn 同时启动了交互式 API 文档服务。默认可以通过访问 http://127.0.0.1:8000/docs 来打开 Swagger UI 风格的文档。
文档不仅用于浏览 API 的各个端点、查看请求和响应的结构，还支持直接在文档中进行 API 请求测试。
通过 Swagger UI，你可以轻松理解每个路由操作的输入参数、输出格式和请求示例。
或者通过 http://127.0.0.1:8000/redoc 来打开 ReDoc 风格的文档。ReDoc 的设计强调文档的可视化和用户体验

依赖项，前处理和后处理

依赖项是在路由操作函数执行前或后运行的可复用的函数或对象。

它们被用于执行一些通用的逻辑，如验证、身份验证、数据库连接等。在 FastAPI 中，依赖项通常用于两个方面：

预处理（Before）依赖项：在路由操作函数执行前运行，用于预处理输入数据，验证请求等。

from fastapi import Depends, FastAPI, HTTPException

app = FastAPI()
# 依赖项函数，用于参数检查
def common_parameters(q: str = None, skip: int = 0, limit: int = 100):
    return {"q": q, "skip": skip, "limit": limit}

# 路由操作函数
@app.get("/depend/")
async def read_depend(commons: dict = Depends(common_parameters)):
    return commons

请求处理完成后的操作，比如记录日志、修改响应等。在FastAPI中，这样的操作通常不会通过依赖注入（即使用Depends）来实现，因为Depends的主要目的是在请求处理之前提供所需的资源或数据

后处理逻辑，比如记录日志或修改响应，你更可能会使用FastAPI的@app.on_event(“after_request”)装饰器来注册一个请求后的事件处理函数。这个函数会在每个请求处理完成后被自动调用，更好的方式是使用Fastapi的中间件

Fastapi中间件的花样

单个中间件执行顺序

先看一个例子，用于在所有 HTTP 请求的响应中添加自定义头部 X-After-Request: Processed，并对处理进行计时

from fastapi import Request
import time

@app.middleware("http") # 注册 HTTP 中间件，作用是所有经过 FastAPI 的 HTTP 请求都会先经过此中间件处理。
async def add_process_time_header(request: Request, call_next):
    start_time = time.time()
    response = await call_next(request)
    end_time = time.time()
    response.headers["X-Process-Time"] = str(end_time - start_time) # 请求处理计时
    response.headers["X-After-Request"] = "Processed" # 添加自定义头 X-After-Request: Processed（常用于调试、日志或安全策略）
    return response # 将添加了自定义头的响应返回给客户端。

@app.middleware("http")为标准 FastAPI 使用中间件
request: Request：当前请求的上下文对象。
call_next：调用链中的下一个处理函数（可能是路由函数或其他中间件）。

这里最不好理解的就是请求在加入中间件后的处理流程和顺序，这里我详细介绍一下：
以上面的add_process_time_header中间件为例，当访问路由函数GET http://127.0.0.1:8000/name/fjsi 时，会发生以下步骤：

请求到达中间件
→ 先执行 start_time = time.time()
→ 调用 await call_next(request) ，由于没有别的中间件定义，所以直接进入路由函数，即/name/{my_name}对应的函数read_name()

请求进入路由函数
→ 执行路由函数read_name()
→ 生成响应 {"my_name": "fjsi"}

返回中间件继续处理
→ 执行 end_time = time.time()
→ 添加两个响应头
→ 返回最终响应

多个中间件执行顺序

搞明白这个执行顺序后，再来研究一下刚提到的多个中间件的情况，多个中间件的执行顺序和中间件定义的顺序有关，用下面的例子来说明：
这里我将上面的加processed和计时分成两个中间件了，并且加入了一些日志打印方便查看运行顺序：

@app.middleware("http")
async def add_process_header(request: Request, call_next):
    print("进入add_process_header")
    response = await call_next(request)
    print("退出add_process_header")
    response.headers["X-After-Request"] = "Processed"
    return response
@app.middleware("http")
async def add_spend_time_header(request: Request, call_next):
    start_time = time.time()
    print("进入add_spend_time_header")
    response = await call_next(request)
    print("退出add_spend_time_header")
    end_time = time.time()
    response.headers["X-Process-Time"] = str(end_time - start_time)
    return response

同样访问路由函数GET http://127.0.0.1:8000/name/fjsi 控制台有下面的输出：

进入add_spend_time_header
进入add_process_header
退出add_process_header
退出add_spend_time_header

可见执行顺序是：

请求到达计时中间件
→ 先执行 start_time = time.time()
→ 调用 await call_next(request)
请求到达processed中间件
→ 调用 await call_next(request)
请求进入路由函数
→ 执行路由函数read_name()
→ 生成响应 {"my_name": "fjsi"}
请求返回processed中间件
→ 添加processed响应头
返回计时中间件
→ 执行 end_time = time.time()
→ 添加计时响应头
→ 返回最终响应

形象一点地做个比喻：
有个孤零零地蛋黄(请求)来了，给它裹上蛋白(processed中间件)，再裹上蛋壳(计时中间件)，然后我们的程序想吃蛋黄开始剥鸡蛋(想要处理路由函数)，先剥蛋壳(执行计时中间件)，再剥蛋白(执行processed中间件)，最后吃蛋黄(执行路由函数)，吃完后将蛋白还原(拿到respond，执行processed中间件剩余部分)，最后将蛋壳还原(执行计时中间件剩余部分，返回最终respond)

要注意中间件顺序：多个中间件按声明顺序执行，执行顺序类似洋葱：从外到内进入，从内到外返回