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 Depends

2、定义依赖的组件函数

# 依赖的组件函数
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 user

service/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 result

commoms/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)