FastAPI 依赖注入
依赖注入常用于以下场景：
-
共享业务逻辑（复用相同的代码逻辑）
-
共享数据库连接
-
实现安全、验证、角色权限
-
等……
上述场景均可以使用依赖注入，将代码重复最小化。
创建依赖项
依赖项就是一个函数，且可以使用与路径操作函数相同的参数，依赖项函数的形式和结构与路径操作函数一样。因此，可以把依赖项当作没有「装饰器」（即，没有 @app.get("/some-path") ）的路径操作函数。
from typing import Unionimport uvicorn
from fastapi import FastAPI, Dependsapp = FastAPI()# 依赖的组件函数
async def common_parameters(q: Union[str, None] = None,skip: int = 0,limit: int = 100,
):return {"skip": skip, "limit": limit, "q": q}@app.get("/items/")
async def read_items(commons: dict = Depends(common_parameters)):return commons@app.get("/users/")
async def read_users(commons: dict = Depends(common_parameters)):return commonsif __name__ == '__main__':uvicorn.run(app)实现过程：
1、导入 Depends
from fastapi import Depends2、定义依赖的组件函数
# 依赖的组件函数
async def common_parameters(q: Union[str, None] = None,skip: int = 0,limit: int = 100,
):return {"skip": skip, "limit": limit, "q": q}3、声明依赖项
声明依赖项需要使用 Depends 和一个新的参数：
async def read_items(commons: dict = Depends(common_parameters)):其他与「依赖注入」概念相同的术语为：
-
资源（Resource）
-
提供方（Provider）
-
服务（Service）
-
可注入（Injectable）
-
组件（Component）
依赖注入系统如此简洁的特性，让 FastAPI 可以与下列系统兼容：
-
关系型数据库
-
NoSQL 数据库
-
外部支持库
-
外部 API
-
认证和鉴权系统
-
API 使用监控系统
-
响应数据注入系统
-
等等……
类作为依赖项
依赖项时一个可调用的对象，Python 中的 "可调用对象" 是指任何 Python 可以像函数一样 "调用" 的对象。Python 类也是 可调用对象。因此，在 FastAPI 中，你可以使用一个 Python 类作为一个依赖项。
实际上 FastAPI 检查的是它是一个 "可调用对象"（函数，类或其他任何类型）以及定义的参数。
import uvicorn
from fastapi import FastAPI, Dependsapp = FastAPI()fake_items_db = [{"item_name": "Foo"}, {"item_name": "Bar"}, {"item_name": "Baz"}]class CommonQueryParams:def __init__(self, q: str | None = None, skip: int = 0, limit: int = 100):self.q = qself.skip = skipself.limit = limit# FastAPI 调用 CommonQueryParams 类。这将创建该类的一个 "实例"，该实例将作为参数 commons 被传递给你的函数。
@app.get("/items/")
async def read_items(commons: CommonQueryParams = Depends(CommonQueryParams)):response = {}if commons.q:response.update({"q": commons.q})items = fake_items_db[commons.skip: commons.skip + commons.limit]response.update({"items": items})return responseif __name__ == '__main__':uvicorn.run(app, host="127.0.0.1", port=8000)FastAPI 为这些情况提供了一个快捷方式，在这些情况下，依赖项 明确地 是一个类，FastAPI 将 "调用" 它来创建类本身的一个实例。
快捷写法
commons: CommonQueryParams = Depends()分层依赖注入
假设是用户管理模块，创建如下的目录结构以及python文件：
schemas/user_schema.py
from pydantic import BaseModel, Field
from datetime import datetimeclass User(BaseModel):id: int | None = Field(description="主键ID")username: str | None = Field(description="用户名")age: int | None = Field(description="年龄")is_vip: int | None = Field(description="是否会员 0-不是，1-是")create_time: datetime = Field(description="创建时间")dao/user_dao.py
from datetime import datetime
from user.schemas.user_schema import Userclass UserDAO:"""数据层"""def __init__(self):passdef get_user_by_username(self, username)-> User:"""根据用户名查询用户信息:param username::return:"""user = User(id=1, username="alice", age=25, is_vip=0, create_time=datetime.now())return userservice/user_service.py
from fastapi import Dependsfrom user.dao.user_dao import UserDAO
from user.schemas.user_schema import Userclass UserService:"""业务逻辑处理层"""def __init__(self, user_dao: UserDAO = Depends()):self.user_dao = user_daodef find_by_username(self, username: str)-> User:"""根据用户名查询用户信息:param username::return:"""return self.user_dao.get_user_by_username(username)routers/user_routers.py
from fastapi import APIRouter, Dependsfrom commons.result import Result
from user.exceptions.user_exception import UserNotFoundException
from user.service.user_serivce import UserServiceusers = APIRouter()@users.get(path="/find/{username}", summary="根据用户名查询用户", response_model=Result)
async def find_user_by_username(username: str, user_service: UserService = Depends()):user = user_service.find_by_username(username)# 用户不存在if user is None:raise UserNotFoundExceptionresult = Result()result.data = userreturn resultcommoms/result.py
from typing import Anyfrom pydantic import BaseModel, Fieldclass Result(BaseModel):code: int = Field(description="响应状态码", default=200)message: str = Field(description="响应结果信息", default="成功")data: Any = Field(description="响应结果数据", default=None)