FastAPI 快速入门

基于 官方文档 快速学习一波，我想吐槽这官方文档的组织形式就是我想写的形式，问题驱动，类似 cookbook……这让我感觉做笔记的必要性降低了。

FastAPI 是 Python 上的一个现代的，大量利用类型标注的（基于 pydantic），支持异步的，支持依赖注入的，支持插件的 Web 后端框架。因为基于 pydantic，所以它能利用上你的接口的类型标注，自动生成 API 文档，做类型校验……

我只关注我关心的部分，以 Cookbook 的形式明确常见需求和解决方案。

Hello World

https://fastapi.tiangolo.com/zh/tutorial/

安装 FastAPI：

1
pip install fastapi[standard]
1
2
3
4
5
6
7
8
9
10
11
12
13
from fastapi import FastAPI
import uvicorn

app = FastAPI()

@app.get("/")
# 既可以是 async 函数，也可以是普通函数
async def hello_world():
    return {"message": "Hello World"}

# as always，这样运行是出于开发，性能不好（无法利用多进程），生产环境应当使用外界的命令，这里不表
if __name__ == '__main__':
    uvicorn.run(app, port=8088)

这里的@app.get('/')，称为路径操作（Path Operation）装饰器，hello_world函数称为路径操作函数，这里的路径即/，操作即GET。就像 flask 一样，FastAPI 直接使用 app 的方法来定义路由（FastAPI 称为路径但我还是喜欢叫路由 Route）。

启动，访问localhost:8088看到 json 响应，访问localhost:8088/docs看到自动生成的 swagger 文档（显然，这里的响应体的类型无法产生文档，因为我没有显式地标注，即使 IDE 能推断，FastAPI 无法推断这个）。

查询参数，路径变量

路径变量直接在 path 中通过{path_variable_name}的形式去定义，然后只需在函数参数中定义同名参数即可，FastAPI 知晓如何进行绑定，参数可以标注类型让 FastAPI 进行校验和转换（不然总是str）。

1
2
3
@app.get("/items/{item_id}/{msg}")
async def read_item(item_id: int, msg):
    return { "item_id": item_id, 'msg': msg } # type(item_id) == int, type(msg) == str

查询参数是可选的如果给定默认值或者类型包含| None。

1
2
3
4
@app.get("/items")
async def read_item(query: str, asc: bool = False, order_by: str | None = None):
    # query is required
    return {'query': query, 'order_by': order_by, 'asc': asc}

查询参数可以使用 Query 进行更复杂的定义，包括指定校验，文档描述……Query 通常以默认值的方式提供，但也可以通过元信息的方式提供。

同理，路径变量可以用 Path 进行更复杂的定义，这里不表。

1
2
3
4
5
6
7
8
9
10
@app.get("/items")
async def read_item(query: str | None = Query(default=None, min_length=3, max_length=50, description='item 关键字')):
    # 要么不给定 query，要么 query 给定后长度必须在 3-50 之间，这里的 description 会显示在 swagger 上
    return {'query': query}

# 或者用 Annotated 形式，这允许做更多骚操作
# 注意使用 Annotated 时，default 参数不可用，必须使用默认值语法
@app.get("/items_1")
async def read_item_1(query: Annotated[str | None, Query(min_length=3, max_length=50, description='item 关键字')] = None):
    return {'query': query}

FastAPI 能自动识别查询参数，路径变量和请求体：

  1. 路径中声明了相同参数的参数，是路径变量
  2. 类型是未给定或者（int、float、str、bool 等）单类型的参数，是查询参数
  3. 类型是 Pydantic 模型的参数，是请求体
    1. 实践证明，dict，list 类型也是请求体

这里不学习什么用一个 dict 获取所有查询参数等操作，只有最操蛋的业务才需要关心这个（恼）。但确实存在用 pydantic 模型来定义查询参数模型的操作，只需标注它的类型为Annotated[ParamModel, Query()]即可。

JSON 请求体

https://fastapi.tiangolo.com/zh/tutorial/body/

dict，list 或 Pydantic 模型参数被视为请求体。

1
2
3
4
5
6
7
class Idol(BaseModel):
    name: str
    age: int

@app.put("/idol")
async def add_idol(idol: Idol):
    return idol

如果有多个复杂类型，它们会被视为请求体 json 中的各个 key。

1
2
3
4
5
6
7
class Idol(BaseModel):
    name: str
    age: int

@app.put("/idol")
async def add_idol(idol: Idol, friends: list[Idol]):
    return {'idol': idol, 'friends': friends} # 请求体会是这个形式

同上面的 Query 和 Path，FastAPI 提供了 Body 类型，允许添加元信息云云，如果要使用单类型作为请求体或者请求体中特定 key 的类型（它默认会被视为查询参数），必须使用 Body。

x-www-urlencoded 和 form-data 请求体

要使用表单，需要安装下面的依赖（我猜测fastapi[standard]自带了它）：

1
pip install python-multipart

同上面的 Query，Path，Body，FastAPI 提供 Form 表示表单数据，File 表示文件数据，其中 Form 继承自 Body，File 继承自 Form（……为何？）。显然——只要使用了 Form，请求体类型就会是 x-www-urlencoded，只要使用了 File，请求体类型就会是 form-data。这里不详细学习，只给示例。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
class FormData(BaseModel):
    username: str
    password: str

@app.post("/login")
# 也可以逐字段去定义
async def login(form: FormData = Form()): # 文档中用的是 Annotated
    return form
    
@app.post("/upload")
# 最佳实践——文件用 File，其他用 Form（虽然最终总是 form-data）
# UploadFile 不能放到 BaseModel 中，因为 pydantic 只支持能 json 序列化的玩意儿 
# 这里就不研究怎么把 File 也放到一个模型中了
async def upload(override_name: str = Form(default=''), file: UploadFile = File()):
    return file.filename

Header

https://fastapi.tiangolo.com/zh/tutorial/header-params/

就如同上面的 Query，Path……（我不报菜名了），Header 也提供了一个类供你绑定到参数上，FastAPI 会处理名称的转换，将 header 中的中划线替换为下划线。

1
2
3
@app.get("/items/")
async def read_items(user_agent: str = Header(None), x_token: str = Header(None)):
    return {"User-Agent": user_agent, "X-Token": x_token}

自定义响应

这里说的是通过 return 返回的正常响应。

在路径操作装饰器中可以指定响应的状态码等。

1
2
3
@app.post("/items/", status_code=201)
async def create_item(name: str):
    return {"name": name}

这里主要关注如何提供非 json 的响应，如文件下载，图像。

FastAPI 提供了诸多类来自定义响应（显然会自动设置 header），如HTMLResponse，FileResponse，以及原始的Response供你做完全的自定义。当然，你也可以用Request（给参数标注这个类型，request: Request）来获取原始 request 来获取所有 header。

异常处理

https://fastapi.tiangolo.com/zh/tutorial/handling-errors/

如果代码出现异常要告知前端，应当 raise HTTPException，这里可以手动指定 status_code 和要返回的 detail，也可以指定响应的 header 等。可以说 raise HTTPException(...) 就等价于 return Fail(...)，只不过异常是传递性的。

1
2
3
4
5
@app.get("/items/{item_id}")
async def read_item(item_id: str):
    if item_id not in items:
        raise HTTPException(status_code=404, detail="Item not found") # detail 可以是任何可 json 序列化的类型
    return {"item": items[item_id]}

FastAPI 支持全局异常处理器，且支持复用和覆盖，但这里不表，用到的时候 现查 即可。

静态文件

https://fastapi.tiangolo.com/zh/tutorial/static-files/

这样：

1
app.mount("/static", StaticFiles(directory="static"), name="static")

它其实不止是处理静态文件，实际上它是一个更广泛的东西——去“挂载”一个“子应用”到特定的路径下，这里不表。

按文件划分接口

https://fastapi.tiangolo.com/zh/tutorial/bigger-applications/

不能把所有东西都塞到main.py吧？按文件划分接口是常见需求，使用APIRouter。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
from fastapi import FastAPI, APIRouter
import fastapi
import uvicorn

app = FastAPI()

# APIRouter 的形式和 app 几乎一致，APIRouter 允许嵌套
# 模块 A（模块就像 Controller，一般分到单独文件中）
router_user = APIRouter(prefix='/user')
@router_user.get('/') # 实际对应 /user
def hello():
    return 'user module'

# 模块 B
router_sys = APIRouter(prefix='/sys')
@router_sys.get('/') # 实际对应 /sys
def world():
    return 'sys module'

# 注意需要手动 include
app.include_router(router_user)
app.include_router(router_sys)

if __name__ == '__main__':
    uvicorn.run(app, port=8088)

依赖注入

https://fastapi.tiangolo.com/zh/tutorial/dependencies/

FastAPI 的依赖注入并非像 Spring 那样提前把所有 Bean 给你创建好放到一个全局的容器中，然后后面反复用，而是每次来新请求的时候，都会重新去创建依赖。可以说，依赖是请求作用域的（request-scoped）。以及，在引用其他依赖时，可以配置是否使用缓存（默认为 True），如果使用且这个依赖已经被创建过，则复用它，就是说同一个依赖的实例可能被多个子依赖同时使用。（简单理解这个use_cache，就是同一个请求中这一个依赖是否会被执行多次）

以 FastAPI 的角度出发，或许可以说就连上面的 Header，Body，Query 等 HTTP 请求中的东西也可以说是依赖了，有趣的心智模型。是时候把我心里的 Spring MVC 的那一套东西丢掉了。

显然——依赖的创建不能太昂贵，比如你不应当在依赖里创建数据库连接（而是从现成的连接池中取连接）……大概吧。

依赖本质来说有两种情况——你需要它的返回值，如数据库连接，或者鉴权并获取用户信息；你不需要它的返回值，而只做一些副作用，如鉴权，限流，在请求前后记录日志。后者会让你疑惑为何不直接使用中间件——中间件不能利用依赖，而副作用型依赖可以。

返回值型依赖

返回值型依赖通过函数参数注入，通过 Depends 去注入。

一个可能常见的需求就是获取当前用户信息，这里就适合使用依赖注入。这里同时演示了嵌套的依赖——get_me 依赖 get_current_user 依赖 UserService 依赖 foo。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
async def foo():
    return 'hello'

# 似乎 Python 不兴这个……？
class UserService:
    def __init__(self, foo: str = Depends(foo)): # 如果 UserSerivce 有自己的依赖，可以加到这里
        print(foo)

    def get_current_user(self):
        if ...: # not logined
            return None
        return { 'username': 'naer' } 

# 注意这里的 user_service: UserService = Depends() 的写法——它等价于 
# user_service: UserService = Depends(UserService)，
# FastAPI 允许避免这里的 UserService 出现两次
async def get_current_user(user_service: UserService = Depends()):
    if user := user_service.get_current_user():
        return user
    raise HTTPException(401, 'not logined')

# 但如果依赖不是一个具体的类（显然 FastAPI 无法通过类型标注定位这个依赖），Depends 的参数就必须传了，这个很蛋疼
@app.get("/me")
async def get_me(current_user: dict = Depends(get_current_user)):
    return current_user

副作用型依赖

副作用型依赖，典型情况就是验证用户是否登陆，以及记录请求日志等……

副作用型依赖就不再使用参数去注入了——而是在装饰器，或者特定 APIRouter，或者整个 app 去注入。

下面是各种依赖注入方式，这里的依赖提供一个计时功能，这里同时演示了生成器函数作为依赖——这允许提供“清理”功能。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
# Request 即原始请求，FastAPI 会帮忙注入
async def timer(request: Request):
    # 接口执行之前
    url = request.url
    start_time = time()
    yield # 对于返回值型依赖，这里可以返回一些东西
    # 接口执行之后
    end_time = time()
    print(f'接口 {url}: {end_time-start_time:.4f} s')
    
# 全局
app = FastAPI(dependencies=[Depends(timer)])

# 特定 router
some_router = APIRouter(dependencies=[Depends(timer)])

# 添加 router 时（尚不知这和上面的 router 自己的配置是覆盖还是并集，不研究）
app.include_router(some_router, dependencies=[Depends(timer)])

# 特定路径操作
@app.get("/", dependencies=[Depends(timer)])
async def hello():
    await asyncio.sleep(1)
    return 'Hello, World'

考虑到副作用型依赖在我的需求下大概能完全地替代中间件，中间件就不研究了。

Basic Auth 鉴权

直接贴代码，这里使用了 FastAPI 的 security 模块，同时也展示了有状态的依赖是怎么一回事——我给 Depends 的不是一个函数而是一个实例（明显需要覆盖__call__方法）。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
app = FastAPI()
security = HTTPBasic()

# 简单的 Basic Auth 验证函数
def get_current_username(credentials: HTTPBasicCredentials = Depends(security)):
    correct_username = "naer"
    correct_password = "aaaaaa"
    if credentials.username != correct_username or credentials.password != correct_password:
        raise HTTPException(status_code=401, detail="Incorrect credentials")
    return credentials.username

@app.get("/")
async def hello(user_name: str = Depends(get_current_username)):
    await asyncio.sleep(1)
    return f'Hello, {user_name}'

WebSocket

https://fastapi.tiangolo.com/zh/advanced/websockets/

WebSocket 的编写在协程编程下还是挺方便的。

安装下面依赖：

1
pip install websockets

直接看代码吧，来自官方的示例。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
from fastapi import FastAPI, WebSocket
from fastapi.responses import HTMLResponse

app = FastAPI()

html = """
<!DOCTYPE html>
<html>
    <head>
        <title>Chat</title>
    </head>
    <body>
        <h1>WebSocket Chat</h1>
        <form action="" onsubmit="sendMessage(event)">
            <input type="text" id="messageText" autocomplete="off"/>
            <button>Send</button>
        </form>
        <ul id='messages'>
        </ul>
        <script>
            var ws = new WebSocket("ws://localhost:8000/ws");
            ws.onmessage = function(event) {
                var messages = document.getElementById('messages')
                var message = document.createElement('li')
                var content = document.createTextNode(event.data)
                message.appendChild(content)
                messages.appendChild(message)
            };
            function sendMessage(event) {
                var input = document.getElementById("messageText")
                ws.send(input.value)
                input.value = ''
                event.preventDefault()
            }
        </script>
    </body>
</html>
"""

@app.get("/")
async def get():
    return HTMLResponse(html)

@app.websocket("/ws")
async def websocket_endpoint(websocket: WebSocket):
    await websocket.accept()
    while True:
        data = await websocket.receive_text()
        await websocket.send_text(f"Message text was: {data}")

生命周期

https://fastapi.tiangolo.com/zh/advanced/events/

在项目启动、关闭时执行特定操作是常见需求。FastAPI 倒也支持像回调一样去定义生命周期钩子，但更倾向使用 app 的 lifespan 参数去做操作（这会覆盖掉生命周期钩子）。lifespan 接受一个异步上下文处理器。

见代码：

1
2
3
4
5
6
7
8
9
@asynccontextmanager
async def lifespan(app: FastAPI):
    # 启动时
    print('hello')
    yield
    # 关闭时
    print('bye')

app = FastAPI(lifespan=lifespan)

and…That’s it.