依赖注入
函数依赖、类依赖、子依赖
路径操作装饰器依赖、全局依赖、使用yield的依赖

依赖注入

为代码注入依赖项目，从而提高代码的复用率

依赖注入常用于以下场景：
- 共享业务逻辑（复用相同的代码逻辑）
- 共享数据库连接
- 实现安全、验证、角色权限等

依赖项Depends

依赖项必须是可调用对象(callable)
需要在路径操作函数的参数中使用 Depends
- Depends接收一个参数：可调用对象，如函数：该函数接收的参数和路径操作函数的参数一样

函数作为依赖项

from typing import Optional

import uvicorn
from fastapi import FastAPI, Depends

app = FastAPI()

# 也可以不使用异步
async def common_parameters(q: Optional[str] = None, skip: int = 0, limit: int = 100):
    print("执行依赖函数...")
    return {"q": q, "skip": skip, "limit": limit}

@app.get("/items/")
async def read_items(commons: dict = Depends(common_parameters)):
    print("执行请求...")
    return commons

@app.get("/users/")
async def read_users(commons: dict = Depends(common_parameters)):
    return commons

if __name__ == '__main__':
    uvicorn.run("main:app", reload=True)

浏览器访问：http://127.0.0.1:8000/items/?q=jerry&skip=10&limit=99在控制台看看函数执行的先后顺序
参数传递给依赖项函数common_parameters，函数先执行返回的结果会先赋值给路径操作函数的参数commons
显然，这样的好处是，只编写一次代码，FastAPI 就可以为多个路径操作（如上例子的/items和/users下）共享这段代码

类作为依赖项

在python中类也是可调用对象（创建类实例）

from typing import Optional

import uvicorn
from fastapi import FastAPI, Depends

app = FastAPI()

class CommonQueryParams:
    def __init__(self, q: Optional[str] = None, skip: int = 0, limit: int = 100):
        self.q = q
        self.skip = skip
        self.limit = limit

@app.get("/items/")
async def read_items(commons: CommonQueryParams = Depends(CommonQueryParams)):  # 注意写法（写了两次CommonQueryParams）
    response = {"skip": commons.skip, "limit": commons.limit}
    if commons.q:
        response.update({"q": commons.q})
    return response

if __name__ == '__main__':
    uvicorn.run("main:app", reload=True)

同样访问http://127.0.0.1:8000/items/?q=jerry&skip=10&limit=99时，会先实例化CommonQueryParams，该实例会作为参数commons传递给函数
注意：其中commons: CommonQueryParams = Depends(CommonQueryParams) CommonQueryParams该类型注释，在fastapi中没有任何作用，不会起到类型转换或参数校验的作用

一种简化写法：Depends()不需要写任何参数
- 仅针对类作为依赖项，且fastapi将会通过实例化来调用的情形
  如上例子则可写为

...
...
@app.get("/items/")
async def read_items(commons: CommonQueryParams = Depends()):
...
...

Depends 的参数 use_cache
- use_cache 默认值是 True
- FastAPI 不会为同一个请求多次调用同一个依赖项，而是把依赖项的返回值进行缓存，并把它传递给同一请求中所有需要使用该返回值的依赖项
- 如果不想使用缓存值，需要在同一请求的每一步操作（多次）中都实际调用依赖项，可以把 Depends 的参数 use_cache 的值设置为 False

子依赖项

可以按需声明任意深度的子依赖项嵌套层级

from typing import Optional

import uvicorn
from fastapi import FastAPI, Depends, Cookie

app = FastAPI()


def query_extractor(q: Optional[str] = None):
    print(1, "执行第一层依赖项query_extractor...")
    return q


def query_or_cookie_extractor(q: str = Depends(query_extractor), last_query: Optional[str] = Cookie(None)):
    print(2, "执行第二层依赖项query_or_cookie_extractor...")
    if not q:
        return last_query
    return q


@app.get("/items/")
async def read_query(query_or_default: str = Depends(query_or_cookie_extractor)):
    return {"q_or_cookie": query_or_default}


if __name__ == '__main__':
    uvicorn.run("main:app", reload=True)

在postman中访问http://127.0.0.1:8000/items/?q=123或者访问http://127.0.0.1:8000/items/且在请求头中带上Cookie，值为last_query=222

路径操作装饰器依赖项

有些依赖项不返回值，但仍要执行或解析该依赖项，此时我们可以在路径操作装饰器中添加一个由 dependencies 组成的 list


import uvicorn
from fastapi import FastAPI, Depends, Header, HTTPException

app = FastAPI()


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")


async def verify_key(x_key: str = Header(...)):
    if x_key != "fake-super-secret-key":
        raise HTTPException(status_code=400, detail="X-Key header invalid")
    return x_key   # 无论路径装饰器依赖项是否返回值，路径操作都不会使用这些值


@app.get("/items/", dependencies=[Depends(verify_token), Depends(verify_key)])
async def read_items():
    return [{"item": "Foo"}, {"item": "Bar"}]


if __name__ == '__main__':
    uvicorn.run("main:app", reload=True)

在postman访问http://127.0.0.1:8000/items并添加headers：x-token:fake-super-secret-token 和 x-key: fake-super-secret-key

全局依赖项

通过把依赖项list添加到app（fastapi.FastAPI的参数dependencies ）

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")


async def verify_key(x_key: str = Header(...)):
    if x_key != "fake-super-secret-key":
        raise HTTPException(status_code=400, detail="X-Key header invalid")
    return x_key

app = FastAPI(dependencies=[Depends(verify_token), Depends(verify_key)])  # 

@app.get("/items/")
async def read_items():
    return [{"item": "Portal Gun"}, {"item": "Plumbus"}]


@app.get("/users/")
async def read_users():
    return [{"username": "Rick"}, {"username": "Morty"}]

这样，依赖项可以用于应用中的所有路径操作

使用yield的依赖项

在依赖函数调用后，希望做一些额外的操作，如创建一个数据库session依赖，希望在响应发送后关闭这个session，则需要使用生成器函数作为依赖项
如下创建数据库session的伪代码：

async def get_db():
    db = DBSession()
    try:
        yield db
    finally:
        db.close()

yield的值db将作为依赖项参数注入到路径操作中
yield语句后面的语句将在响应发送之后执行

使用yield的子依赖
如以下伪代码：

from fastapi import Depends

async def dependency_a():
    dep_a = generate_dep_a()
    try:
        yield dep_a
    finally:
        dep_a.close()

async def dependency_b(dep_a=Depends(dependency_a)):
    dep_b = generate_dep_b()
    try:
        yield dep_b
    finally:
        dep_b.close(dep_a)

async def dependency_c(dep_b=Depends(dependency_b)):
    dep_c = generate_dep_c()
    try:
        yield dep_c
    finally:
        dep_c.close(dep_b)

如果在yield语句之后的代码抛出异常，通过try...except将无法捕获异常
可以在响应返回后执行后台任务，如后台任务触发了一个db异常，我们在finally中回滚事务或关闭会话，甚至可以记录日志等
如果想在响应返回前，修改响应或处理异常，可使用自定义异常
Client、Exception handler、Dependencies with yield、Path Operation、Background tasks交互流程图

上下文管理器

在类中实现__enter__、__exit__魔术方法，可进一步参考python学习笔记：上下文管理
如以下伪代码：

class MySuperContextManager:
    def __init__(self):
        self.db = DBSession()

    def __enter__(self):
        return self.db

    def __exit__(self, exc_type, exc_value, traceback):
        self.db.close()


async def get_db():
    with MySuperContextManager() as db:
        yield db

小猴子jerry

fastapi 官方文档学习3：依赖注入

依赖注入

依赖项Depends

函数作为依赖项

类作为依赖项

子依赖项

路径操作装饰器依赖项

全局依赖项

使用yield的依赖项

上下文管理器

扩展阅读与学习

fastapi 官方文档学习2：响应