FastAPI+Pytest+SQLAlchemy单测试会话下工厂类数据库会话问题

核心问题是factory-boy使用的数据库会话和FastAPI接口依赖注入的会话未绑定到同一个临时测试数据库，且未共享同一实例。以下是具体解决方案：

1. 改造TestDatabase类，确保单测试用例内会话唯一 #

修改TestDatabase，让它在单个测试用例生命周期内复用同一个SQLAlchemy会话，避免每次调用生成新会话：

from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker, Session
from app.models import Base  # 导入你的SQLAlchemy Base模型

class TestDatabase:
    def __init__(self):
        # 生成唯一测试库名，避免测试间冲突
        self.db_name = f"test_db_{hash(self)}"
        self.engine = create_engine(f"postgresql://username:password@localhost/{self.db_name}")
        self._session: Session | None = None
        self._session_factory = sessionmaker(autocommit=False, autoflush=False, bind=self.engine)

    def setup(self):
        # 创建测试用表
        Base.metadata.create_all(bind=self.engine)

    def teardown(self):
        # 销毁测试用表和数据库
        Base.metadata.drop_all(bind=self.engine)
        # 可选：删除数据库本身（根据数据库类型调整）
        with self.engine.connect() as conn:
            conn.execute(f"DROP DATABASE IF EXISTS {self.db_name}")
            conn.commit()

    def get_session(self) -> Session:
        # 单例模式返回会话，确保测试用例内唯一
        if self._session is None:
            self._session = self._session_factory()
        return self._session

2. 让factory-boy容器化获取会话 #

修改工厂类，通过dependency-injector从容器中获取TestDatabase的会话，确保和接口用的是同一个：

import factory
from sqlalchemy.orm import Session
from app.models import User  # 你的模型类
from dependency_injector.wiring import inject, Provide
from app.containers import Container

class UserFactory(factory.alchemy.SQLAlchemyModelFactory):
    class Meta:
        model = User
        sqlalchemy_session_persistence = "commit"

    id = factory.Sequence(lambda n: n + 1)
    email = factory.LazyAttribute(lambda obj: f"test_user_{obj.id}@example.com")
    password = factory.PostGenerationMethodCall("set_password", "test123")

    @classmethod
    @inject
    def _create(cls, model_class, *args, session: Session = Provide[Container.database.get_session], **kwargs):
        # 从容器获取会话，替代工厂默认的会话创建逻辑
        obj = model_class(*args, **kwargs)
        session.add(obj)
        session.commit()
        return obj

3. 配置Pytest Fixture绑定容器与测试数据库 #

在conftest.py中定义fixture，确保每个测试用例使用独立的TestDatabase，并替换容器中的依赖：

import pytest
from app.containers import Container
from app.db import TestDatabase

@pytest.fixture(scope="function")
def test_database():
    db = TestDatabase()
    db.setup()
    yield db
    db.teardown()
    # 关闭会话
    if db._session:
        db._session.close()

@pytest.fixture(scope="function")
def container(test_database):
    container = Container()
    # 替换容器中的Database实现为TestDatabase实例
    container.database.override(test_database)
    # 注入到需要的模块（接口、工厂等）
    container.wire(modules=["app.api", "app.factories"])
    yield container
    # 测试结束后解除注入
    container.unwire()

关键注意事项 #

• 确保fixture的scope为function，每个测试用例都有独立的数据库和会话
• TestDatabase的get_session方法必须保证单例，避免同一测试内生成多个会话
• 工厂类的_create方法通过@inject获取容器会话，彻底替代factory-boy默认的会话逻辑

内容的提问来源于stack exchange，提问作者Daniel