接：一文掌握异步web框架FastAPI（二）-CSDN博客
目录
六、依赖注入
1、基本用法
2、使用依赖注入传递配置参数
3、使用依赖注入进行权限验证
4、使用依赖注入管理资源
5、使用依赖注入进行依赖项缓存
6、使用依赖注入进行跨模块依赖管理
7、使用依赖注入进行请求上下文管理
七、中间件
1、日志记录中间件
2、错误处理中间件
3、身份验证中间件（JWT认证）
4、压缩响应中间件
5、添加响应头中间件
六、依赖注入
FastAPI 提供了一个强大的依赖注入系统，使得管理和组织代码中的依赖关系变得更加简单和高效。依赖注入（Dependency Injection, DI）是一种设计模式，它允许将组件（如函数、类或服务）之间的依赖关系明确地表达出来，而不是硬编码在组件内部。这使得代码更加模块化、可测试和易于维护。
在 FastAPI 中，依赖注入主要通过 Depends 类实现。Depends 用于声明一个函数的依赖项，FastAPI 会自动处理这些依赖项的创建和传递。
- 提高代码的可维护性：依赖关系明确，易于理解和修改。
- 增强代码的可测试性：可以轻松地替换依赖项，进行单元测试。
- 减少代码耦合：组件之间的耦合度降低，系统更加灵活。
1、基本用法
定义依赖项、使用依赖项以及在类中使用依赖项
from fastapi import FastAPI, Depends, HTTPExceptionapp = FastAPI()def get_db():db = "数据库连接"return dbdef get_user(db: str = Depends(get_db)):# 依赖项的异常处理if db == "数据库连接":raise HTTPException(status_code=500, detail="数据库连接失败")user = {"name": "张三", "db": db}return userclass UserService:"""类作为依赖项"""def __init__(self, user: dict = Depends(get_user)):self.user = user# 多级依赖, FastAPI 会自动解析和注入这些依赖项。
@app.get("/users/")
async def read_users(user_service: UserService = Depends()):return user_service.user
2、使用依赖注入传递配置参数
使用依赖注入来传递配置参数，使得配置管理更加集中和灵活。
from fastapi import FastAPI, Dependsapp = FastAPI()# 设置类，用于存储数据库网址和密钥
class Settings:def __init__(self, db_url: str, secret_key: str):self.db_url = db_urlself.secret_key = secret_key# 获取设置的函数，返回包含数据库网址和密钥的 Settings 实例
def get_settings():return Settings(db_url="sqlite:///./test.db", secret_key="mysecret")# 获取数据库连接的函数，依赖于 get_settings 函数
def get_db(settings: Settings = Depends(get_settings)):db = f"连接到数据库: {settings.db_url}"return db# 异步读取数据库的路由，依赖于 get_db 函数
@app.get("/db/")
async def read_db(db: str = Depends(get_db)):return {"db": db}
3、使用依赖注入进行权限验证
用于实现权限验证，确保只有符合条件的用户才能访问特定资源。
from fastapi import FastAPI, Depends, HTTPException, status
from typing import Optionalapp = FastAPI()def get_current_user(token: str) -> Optional[str]:if token == "valid-token":return "张三"return Nonedef get_current_active_user(current_user: str = Depends(get_current_user)):if current_user is None:raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="未认证")return current_user@app.get("/users/me")
async def read_users_me(current_user: str = Depends(get_current_active_user)):return {"user": current_user}
请求：
import requests# 定义请求的URL
url = 'http://127.0.0.1:8000/users/me'
# 发送POST请求
response = requests.get(url, params={'token': 'valid-token'})
# 打印响应状态码和内容
print(f"Status Code: {response.status_code}")
print(f"Response Content: {response.text}")
4、使用依赖注入管理资源
用于管理资源，如数据库连接、文件句柄等，确保资源的正确创建和释放。
from fastapi import FastAPI, Depends, HTTPException
import sqlite3app = FastAPI()def get_db_connection():try:conn = sqlite3.connect("test.db")return connexcept sqlite3.Error as e:raise HTTPException(status_code=500, detail=f"Database connection error: {e}")@app.get("/items/")
async def read_items(conn: sqlite3.Connection =