You need to enable JavaScript to run this app.
优惠活动
大模型
产品
解决方案
定价
更多

FastAPI数据库会话传递疑问:详解第三种依赖注入策略

数据库会话传递方案解析

一、你当前的方案:DAO类直接导入sessionmaker

  • 实现方式:在DAO类中直接引入sessionmaker,初始化后生成数据库会话使用
  • 存在问题:DAO与具体的会话创建逻辑深度绑定,单元测试时要切换到测试数据库,要么修改DAO代码,要么依赖环境配置硬适配,灵活性极差

二、端点通过Depends(get_db)传递会话

  • 实现方式:先定义get_db函数生成会话,在FastAPI端点中通过db: Session = Depends(get_db)获取会话,再传入DAO方法
  • 优势:端点与会话创建逻辑解耦,测试时可通过覆盖get_db来使用测试会话
  • 局限:DAO仍依赖具体的Session类型,且每次调用DAO都需手动传递会话,代码冗余度高

三、依赖覆盖策略:抽象基类+存根函数实现

核心原理

这个方案的本质是通过抽象层解耦会话获取逻辑与业务代码,彻底打破依赖绑定,具体步骤如下:

  1. 定义抽象基类:先定下会话提供者的接口规则,不涉及具体实现:
from abc import ABC, abstractmethod
from sqlalchemy.orm import Session

class DatabaseSessionProvider(ABC):
    @abstractmethod
    def get_session(self) -> Session:
        pass
  1. 编写生产环境实现:基于真实数据库的sessionmaker实现上述抽象类:
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker

class ProductionSessionProvider(DatabaseSessionProvider):
    def __init__(self):
        engine = create_engine("sqlite:///./test.db")
        self.SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
    
    def get_session(self) -> Session:
        return self.SessionLocal()
  1. 创建存根依赖函数:作为FastAPI的依赖入口,默认返回生产环境的会话提供者:
from fastapi import Depends

def get_db_provider() -> DatabaseSessionProvider:
    return ProductionSessionProvider()
  1. DAO依赖抽象而非具体实现:DAO不再自行创建会话,而是依赖抽象的会话提供者:
class UserDAO:
    def __init__(self, db_provider: DatabaseSessionProvider = Depends(get_db_provider)):
        self.db_provider = db_provider
    
    def get_user(self, user_id: int):
        db = self.db_provider.get_session()
        # 执行具体数据库操作
        db.close()
  1. 测试时覆盖依赖:编写测试用的会话提供者(比如用内存数据库),直接替换原有的依赖函数即可,无需修改DAO代码:
class TestSessionProvider(DatabaseSessionProvider):
    def __init__(self):
        engine = create_engine("sqlite:///:memory:")
        self.SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
    
    def get_session(self) -> Session:
        return self.SessionLocal()

# 在测试中覆盖依赖
def override_get_db_provider():
    return TestSessionProvider()

app.dependency_overrides[get_db_provider] = override_get_db_provider

适用场景

  • 项目需要多环境切换(生产、测试、预发布),不同环境的会话逻辑不同(比如测试用内存库、生产用真实数据库),无需修改业务代码,仅替换依赖实现即可
  • 需要单元测试隔离,希望用内存数据库或模拟会话测试DAO逻辑,避免依赖真实数据库
  • 项目规模较大、DAO类较多,需要统一管理会话创建逻辑,避免重复代码

不适用场景

  • 小型项目或快速原型开发:该方案会增加代码复杂度,直接使用Depends(get_db)已足够
  • 简单服务,无需多环境切换也不需要编写单元测试:抽象层反而会让代码变得繁琐

内容的提问来源于stack exchange,提问作者Ch1irxo 17

相关产品推荐
方舟 Agent Plan

超全模态模型 × Harness 升级,最新支持 Deepseek-V4.1-Flash、GLM-5.3 系列、Doubao-Seedream-5.0-pro、Kimi-K3 (部分), 限时 9.9 元起

最近更新时间:2026.06.23 02:40:06