接：一文掌握异步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, HTTPException
app = FastAPI()
def get_db():
db = "数据库连接"
return db
def get_user(db: str = Depends(get_db)):
# 依赖项的异常处理
if db == "数据库连接":
raise HTTPException(status_code=500, detail="数据库连接失败")
user = {"name": "张三", "db": db}
return user
class 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, Depends
app = FastAPI()
# 设置类，用于存储数据库网址和密钥
class Settings:
def __init__(self, db_url: str, secret_key: str):
self.db_url = db_url
self.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 Optional
app = FastAPI()
def get_current_user(token: str) -> Optional[str]:
if token == "valid-token":
return "张三"
return None
def 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 sqlite3
app = FastAPI()
def get_db_connection():
try:
conn = sqlite3.connect("test.db")
return conn
except 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 =
