跳到主要内容

依赖注入

前言

依赖注入(Dependency Injection, DI)是编程模式中的一种依赖倒置原则的范式应用实现。可以理解为,是一种能让一个对象接受来自其所依赖的其它对象的一种模式,是一种能通过某种依赖机制自动对依赖对象进行依赖处理,并对依赖对象执行具体实例化的操作。依赖注入中涉及4个比较关键的概念:

  • 服务:它可以是一个类,也可以是一个函数,主要负责提供具体功能的实现。
  • 客户:负责对服务进行使用的对象。
  • 接口:负责连接客户和对象。因为客户不需要对服务内部的实现细节进行了解,只需要指定具体的服务对象名称即可进行关联。
  • 注入器:主要负责在客户中引入具体的服务实例。

FastAPI中存在一种依赖树机制,依赖树在某种程度上扮演了整个IOC容器的角色,它可自动解析并处理每层中所有依赖项的注册和执行。依赖注入的实现其实是FastAPI框架的另一个特性,主要应用场景有:

  • 业务逻辑共享:使用依赖注入机制来统一定义业务逻辑处理,可以避免在每个函数中重复创建,快速共享一段业务逻辑处理。
  • 数据库连接:使用依赖注入机制管理数据库连接,可以避免在每个函数中重复创建连接,共享同一上下文中的连接对象。
  • 认证和授权:使用依赖注入机制来管理用户认证和授权,可以避免在每个函数中重复编写认证和授权代码。
  • 缓存管理:使用依赖注入机制来管理缓存,可以避免在每个函数中重复编写缓存管理代码。
  • 外部API调用:使用依赖注入机制来管理外部API调用,可以避免在每个函数中重复编写API调用代码。
  • 参数校验和转换:使用依赖注入机制来管理参数校验和转换,可以避免重复编码

在FastAPI中,依赖注入机制通过声明函数参数来实现。依赖项可以在路径操作函数中使用,通过Depends函数来注入。

相关模块

from fastapi import Depends

示例-函数式依赖项

from fastapi import Depends, Query
from fastapi.exceptions import HTTPException

# 定义函数式依赖项
def username_check(username: str = Query(...)):
if username != "zhangsan":
raise HTTPException(status_code=403, detail="Permission Denied")
return username

# 通过Depends将依赖项注入到视图函数中
@app.get("/user/login")
def login(username: str = Depends(username_check)):
return username

# 通过依赖注入实现复用同一段业务处理逻辑
@app.get("/user/info")
def userinfo(username: str = Depends(username_check)):
return username

示例-类方式依赖项

"""类作为依赖项"""

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

@app.get("/classes_as_dependency")
# async def classes_as_dependency(commons: CommonQueryParams = Depends(CommonQueryParams)):
# async def classes_as_dependency(commons: CommonQueryParams = Depends()):
async def classes_as_dependency(commons = Depends(CommonQueryParams)):
response = {}
if commons.q:
response.update({"q": commons.q})
items = fake_items_db[commons.page: commons.page + commons.limit]
response.update({"items": items})
return response

多个依赖项注入和依赖项传参

nil

示例-多层依赖项嵌套注入

在复杂的业务场景中,可能会需要处理更多的业务逻辑,比如A依赖项会依赖于B依赖项的返回值,而B依赖项有依赖于C依赖项的返回值。

"""子依赖注入"""
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

@app.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}

多个依赖项注入

# 定义函数式依赖项
def username_check(username: str = Query(...)):
if username != "zhangsan":
raise HTTPException(status_code=403, detail="Permission denied")
return username

def age_check(age: int = Query(...)):
if age < 18:
raise HTTPException(status_code=403, detail="用户未满18岁")
return age

# 通过Depends将依赖项注入到视图函数中
@router.get("/user/login/", summary="user login")
def user_login(username: str = Depends(username_check),
age: int = Depends(age_check)):
return {
"username": username,
"age": age,
}

全局依赖项

FastAPI类提供了一个dependencies参数,该参数可用于实现全局依赖项和路由组依赖项。

from fastapi import APIRouter, Query, Depends
from fastapi.exceptions import HTTPException


# 定义函数式依赖项
def username_check(username: str = Query(...)):
if username != "zhangsan":
raise HTTPException(status_code=403, detail="Permission denied")
return username

def age_check(age: int = Query(...)):
if age < 18:
raise HTTPException(status_code=403, detail="用户未满18岁")
return age

router = APIRouter(
prefix="/dependency",
tags=["FastAPI笔记 - 依赖注入"],
dependencies=[
Depends(username_check),
Depends(age_check),
])
# 通过Depends将依赖项注入到视图函数中
@router.get("/user/login/", summary="user login")
def user_login():
return {
"code": 200,
}

# 通过依赖注入实现复用同一段业务处理逻辑
@router.get("/user/info", summary="user info")
def user_info(username:str, age: int):
return {
"username": username,
"age": age,
}