依赖注入用于把一些可复用的逻辑抽离出来，减少代码重复。例如，返回列表的 API 中都会用到 page 和 page_size 这几个参数，那么可以创建一个依赖来包含这两个参数。

还可以使用依赖来做插件，如在依赖中做一些 Token 的验证等等。所以，在 FastAPI 中，不需要 插件系统，只要把要复用的逻辑实现为一个依赖就好了。

定义一个依赖

依赖的定义是一个 callable, 也就是说函数或者类都可以。依赖的参数和每一个 handler 的参数都 一样，所以 GET/POST/Cookie/Header 等参数都可以以相同的方式使用。

# 使用函数作为依赖
from fastapi import Depends

async def pagination(page: int, size: int):
    return {"page": page, "size": size}

@app.get("/users")
def get_users(pagination: dict=Depends(pagination)):
    users = user_model.get(**pagination)
    return users

@app.get("/items")
def get_items(pagination: dict=Depends(pagination)):
    items = items_model.get(**pagination)
    return items

使用类有一个好处，你可以把这个类作为类型注释，这样和其他的参数使用方式更加一致。但是也有 一个缺点，那就是函数的 __init__ 必须是 sync 的，可能会在线程池中执行。

class Pagination:
    def __init__(self, page: int, size: int):
        self.page = page
        self.size = size

@app.get("/users")
def get_users(pagination: Pagination=Depends(Pagination)):
    users = user_model.get(**pagination)
    return users

# 比较巧妙的一点，我们可以省掉 Depends 的参数
def get_users(pagination: Pagination=Depends()):
    ...

Pydantic 的 model 也可以作为依赖直接使用，这样就相当于把 GET 参数作为一个 model 了。

class Pagination(BaseModel):
    page: int
    page_size: int

@app.get("/items")
def list_items(pagination: Pagination = Depends()):
    ...

添加依赖

依赖可以在三个地方添加：handler 函数参数，路径装饰器，全局 app 实例。如果在 handler 函数的 参数中添加，那么依赖的返回值会作为参数传递进去，就像其他参数一样。其他两种方式返回值都会被丢弃。

from fastapi import FastAPI, APIRouter, Depends

class Dependable:
    def __init__(self, ...):
        ...

# 全局依赖
app = FastAPI(dependencies=[Depends(Dependable), ...])

# 在路径装饰器中添加一个依赖数组
@app.get("/", dependencies=[Depends(Dependable), ...])
def home(param: Dependable = Depends()):   # 在 handler 参数中使用
    ...

# 在 Router 中使用
router = APIRouter(dependencies=[Depends(Dependable)])

依赖依赖依赖

依赖还可以有依赖，也就是依赖的参数也可以是 Depends

def get_user(username: int, token: str = Depends(verify_token)):
    ...

例子

读取固定参数

见前边 pagination 的例子

验证 Token

from fastapi import Depends, FastAPI, Header, HTTPException

async def verify_token(x_token: str = Header(...)):
    if x_token != "fake-super-secret-token":
        raise HTTPException(status_code=400, detail="X-Token header invalid")

app = FastAPI(dependencies=[Depends(verify_token)])

这个例子可以进一步扩展成使用 session.

加载数据库连接

我们需要使用 yield 来返回创建的实例，当 handler 函数执行完毕之后，yield 后边的语句才会 继续执行（关闭数据库）。请特别注意，不要在这里执行 commit，应该由业务代码在 return 之前 执行，否则可能会造成给前端返回的数据不一致的问题。

async def get_db():
    db = DBSession()
    try:
        yield db
    except Exception:
        db.rollback()
        raise
    finally:
        db.close()

如果是 redis，可以这样：

REDIS_URL = "redis://localhost:6379/0"
redis_pool = ConnectionPool.from_url(REDIS_URL)

def get_redis():
    return Redis(connection_pool=redis_pool)

依赖去重

当依赖之间有依赖关系的时候，可能会出现有两个依赖同时调用同一个上级依赖的问题，有点类似 继承中的多继承关系。

参考

1. https://stackoverflow.com/questions/65243587/fastapi-async-class-dependencies
2. Session 相关 Issue
3. https://stackoverflow.com/questions/18022767/python-redis-connection-should-be-closed-on-every-request-flask