依赖注入

• 依赖注入
  • 创建, 导入和声明依赖
  • 类作为依赖项
  • 子依赖的创建和调用
  • 路径操作装饰器中导入依赖
  • FastAPI 框架中全局依赖的使用
  • 使用 yield 的依赖和子依赖

---

"依赖注入" 是指在编程中, 为保证代码成功运行, 先导入或声明其所需要的 "依赖", 如子函数, 数据库连接等

• 提高代码的复用率
• 共享数据库的链接
• 增强安全, 认证和角色管理

FastAPI 的兼容性

• 所有的关系型数据库，支撑NoSQL数据库
• 第三方的包和API
• 认证和授权系统
• 响应数据注入系统

---

创建, 导入和声明依赖

将函数作为依赖进行注入操作(query)

from fastapi import (
    Depends,    # 引入依赖
)

##### Dependencies 创建、导入和声明依赖 #####


async def common_parameters(q: Optional[str] = None, page: int = 1, limit: int = 100):
    """公共函数测试"""
    return {"q": q, "page": page, "limit": limit}


@app05.get("/dependency01")
async def dependency01(commons: dict = Depends(common_parameters)):
    """使用 Depends 进行依赖注入
    """
    return commons


@app05.get("/dependency02")
def dependency02(commons: dict = Depends(common_parameters)):
    """可以在async def中调用def依赖  
    也可以在def中导入async def依赖
    """
    return commons

---

类作为依赖项

# 假设这是一个从数据库中获取的数据
fake_items_db = [{"item_name": "Foo"}, {"item_name": "Bar"}, {"item_name": "Baz"}]


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


@app05.get("/classes_as_dependencies")
# async def classes_as_dependencies(commons: CommonQueryParams = Depends(CommonQueryParams)):
# async def classes_as_dependencies(commons: CommonQueryParams = Depends()):
async def classes_as_dependencies(commons=Depends(CommonQueryParams)):
    """
    使用 Depends 创建类作为依赖项
    """
    response = {}
    if commons.q:
        response.update({"q": commons.q})
    # 切片操作
    items = fake_items_db[commons.page - 1 : commons.page + commons.limit]
    response.update({"items": items})
    return response

需要注意的是, 要与 Pydantic 派生类型作为参数相区分, 使用 pydantic.BaseModel 子类作为参数在函数请求体中, 而类作为依赖项进行注入作为 query 参数

---

子依赖的创建和调用

##### Sub-dependencies 子依赖 #####


def query(q: Optional[str] = None):
    return q


def sub_query(q: str = Depends(query), last_query: Optional[str] = None):
    if not q:
        return last_query
    return q


@app05.get("/sub_dependency")
async def sub_dependency(final_query: str = Depends(sub_query, use_cache=True)):
    """use_cache默认是True,
    表示当多个依赖有一个共同的子依赖时,
    每次request请求只会调用子依赖一次,
    多次调用将从缓存中获取
    """
    return {"sub_dependency": final_query}

query 是子依赖

---

路径操作装饰器中导入依赖

##### Dependencies in path operation decorators 路径操作装饰器中的多依赖 #####


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


@app05.get("/dependency_in_path_operation", 
        dependencies=[Depends(verify_token), Depends(verify_key)]
)   # 这时候不是在函数参数中调用依赖，而是在路径操作中调用依赖
async def dependency_in_path_operation():
    return [
        {"user": "user01"}, 
        {"user": "user02"}
    ]

可以用于校验 key 之类的, 在 Header 中包含 key, 后端路径操作装饰器中导入依赖

---

FastAPI 框架中全局依赖的使用

假设现在有一个子依赖需要在应用的任何地方使用(或者某个组件内部的所有地方), 那么可以使用全局依赖

在 APIRouter 中使用:

# 直接在 APIRouter 定义文件中使用:
app05 = APIRouter(dependencies=[Depends(verify_token), Depends(verify_key)])

在 main App 中使用:

# 引入 chapter05 中的全剧依赖 verify_token 和 verify_key
from .chapter05 import verify_token, verify_key
from fastapi import (
    FastAPI, 
    Request,
    Depends # 引入依赖  
)

# FastAPI 配置项
app = FastAPI(
    # 标题
    title='FastAPI Tutorial and Coronavirus Tracker API Docs',
    # 描述
    description='FastAPI教程 \
        新冠病毒疫情跟踪器API接口文档, \
        项目代码:https://github.com/liaogx/fastapi-tutorial',
    # 版本
    version='1.0.0',
    # Swagger UI 文档地址
    docs_url='/docs',
    # ReDoc 文档地址
    redoc_url='/redocs',
    dependencies = [Depends(verify_token), Depends(verify_key)]
)

---

使用 yield 的依赖和子依赖

yield 关键字在依赖中的使用

Python3.6

Python3.6需要pip install async-exit-stack async-generator

Python3.7

##### Dependencies with yield 带yield的依赖 #####


# 这个需要Python3.7才支持，Python3.6需要pip install async-exit-stack async-generator
# 以下都是伪代码

async def get_db():
    db = "db_connection"
    try:
        yield db
    finally:
        db.endswith("db_close")


async def dependency_a():
    dep_a = "generate_dep_a()"
    try:
        yield dep_a
    finally:
        dep_a.endswith("db_close")


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


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

实际上使用最多的就是 get_db:

# 引入数据库
from ..database import SessionLocal, engine

# Dependency (关键字 yield 可用于共享连接)
def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

---