FastAPI+SQLAlchemy测试会话死锁问题及事务回滚需求

我正在开发基于FastAPI 0.101和SQLAlchemy 2.0.20的服务，但编写测试时遇到问题。推测测试（Pytest）与路由代码因共用yield表达式中的同一个Session而发生死锁。以清空Users表的路由为例：测试中需要先清空表、添加测试用户，调用接口后验证表中数据为0；同时希望整个测试在事务中执行，即使接口调用了session.commit()，最终也能回滚事务不影响数据库原有数据。

当测试执行到test_router.py中标注# HERE的行时，程序会永久挂起，猜测是会话死锁导致。请问如何优化这类数据库测试，实现事务回滚且兼容被测试代码中的commit操作？

相关代码 #

database.py #

SQLALCHEMY_DATABASE_URL = "postgresql://postgres:password@0.0.0.0:5432/db"

engine = create_engine(
    SQLALCHEMY_DATABASE_URL
)
Session = sessionmaker(autocommit=False, autoflush=True, bind=engine, expire_on_commit=False)

Base = declarative_base()


# Dependency
def get_session():
    """Shares a single db session to be used throughout the service"""
    session = Session()
    print("Tried to get session")
    try:
        yield session
    finally:
        session.close()

router.py #

@router.post("/reset")
async def reset(
    session: Session = Depends(get_session),
):
    """
    Clears the users table
    Returns:
        msg: Todos los datos fueron eliminados
    """
    try:
        statement = delete(User)
        print(statement)
        with session:
            session.execute(statement)
            session.commit()
    except Exception as e:
        print("ERROR: /users/reset")
        print(e)
        return JSONResponse(
            status_code=500,
            content={
                "msg": "Un error desconocido ha ocurrido"
            })

    return {"msg": "Todos los datos fueron eliminados"}

conftest.py #

@pytest.fixture(scope="session")
def session() -> Generator:
    Base.metadata.create_all(bind=engine)
    connection = engine.connect()

    # begin a non-ORM transaction
    transaction = connection.begin()
    session = orm.Session(bind=connection)
    yield session
    session.close()
    transaction.rollback()
    connection.close()


@pytest.fixture(scope="session")
def client() -> Generator:
    with TestClient(app) as c:
        yield c

test_router.py #

def test_reset(
        client: TestClient, session: Session
):
    with session as ses:
        ses.execute(
            delete(User)
        )
        mock_user = User(
            id=uuid.UUID("c62147cf-2e63-4508-b1ff-98f805577f2c"),
            username="user",
            email="user@gmail.com",
            phoneNumber="+57 300 500 2000",
            dni="10100190",
            fullName="Usuario Perez Gómez",
            passwordHash="a68f7b00815178c7996ddc88208224198a584ce22faf75b19bfeb24ed6f90a59",
            salt="ES25GfW7i4Pp1BqXtASUFXJFe9PMb_7o-2v73v3svWc",
            token="eHBL1jbhBY6GfZ96DC03BlxM38SPF3npRBceefRgnkTpByFexOe7RPPDdLCh9gejD6Fe6Kdl_s5C3Gljqh3WM2xW1IGdlZQYg"
                  "V0_v55tw_NB19oMzH2t9AjKycEDdwmqPFJVR4sZuk9MFvSGoY_vQa4Y0pwCvxhBDT1VNsDnQio",
            status=UserStatusEnum.NO_VERIFICADO,
            expireAt=datetime.datetime.now(),
            createdAt=datetime.datetime.now(),
            updateAt=datetime.datetime.now()
        )
        session.add(mock_user)
        session.commit()
        assert session.scalar(select(func.count()).select_from(User)) == 1, "Table couldn't be set up successfully"

        response = client.post("/users/reset") # HERE
        assert response.status_code == 200, "The request failed for an unknown reason"
        assert "los datos fueron eliminados" in str(response.json()['msg'])
        assert session.scalar(select(func.count()).select_from(User)) == 0, "Table reset was unsuccessful"

核心问题分析 #

当前死锁的根源：

1. 测试用例持有已开启全局事务的Session，调用session.commit()会尝试提交底层连接事务，与全局测试事务冲突。
2. 路由的get_session()会创建新Session，底层数据库连接被测试的全局事务锁住，导致新Session无法获取可用连接，最终挂起。
3. 路由中with session:块会尝试重新绑定Session上下文，加剧事务嵌套冲突。

优化步骤 #

1. 修改测试fixture，实现Session共享与事务隔离

将conftest.py的fixture改为函数级作用域，并通过依赖覆盖让接口和测试用例共享同一个Session，避免多Session竞争连接：

import pytest
from sqlalchemy import orm
from fastapi.testclient import TestClient
from your_app_module import app, get_session, Base, engine

@pytest.fixture(scope="function")
def db_session():
    # 确保表结构存在
    Base.metadata.create_all(bind=engine)
    # 获取数据库连接并开启全局测试事务
    connection = engine.connect()
    transaction = connection.begin()
    # 创建绑定到该连接的测试Session
    session = orm.Session(
        bind=connection,
        autoflush=True,
        autocommit=False,
        expire_on_commit=False
    )
    
    # 替换FastAPI的get_session依赖，让接口使用测试Session
    def override_get_session():
        try:
            yield session
        finally:
            pass  # 由fixture统一管理Session生命周期
    
    app.dependency_overrides[get_session] = override_get_session
    
    yield session
    
    # 测试结束后回滚事务并清理资源
    session.close()
    transaction.rollback()
    connection.close()
    # 清除依赖覆盖，避免影响其他测试
    app.dependency_overrides.pop(get_session)

@pytest.fixture(scope="function")
def client(db_session):
    with TestClient(app) as c:
        yield c

2. 简化路由代码，移除不必要的Session上下文

路由中嵌套的with session:块会触发额外事务逻辑，直接使用传入的Session即可：

@router.post("/reset")
async def reset(
    session: Session = Depends(get_session),
):
    """
    Clears the users table
    Returns:
        msg: Todos los datos fueron eliminados
    """
    try:
        statement = delete(User)
        print(statement)
        session.execute(statement)
        session.commit()
    except Exception as e:
        print("ERROR: /users/reset")
        print(e)
        return JSONResponse(
            status_code=500,
            content={
                "msg": "Un error desconocido ha ocurrido"
            })

    return {"msg": "Todos los datos fueron eliminados"}

3. 调整测试用例，避免事务冲突

测试中无需真正提交事务，用session.flush()替代session.commit()，同步数据到数据库但不提交，保证验证准确性的同时避免与全局事务冲突：

def test_reset(
        client: TestClient, db_session: Session
):
    # 清空表并同步数据
    db_session.execute(delete(User))
    db_session.flush()
    
    mock_user = User(
        id=uuid.UUID("c62147cf-2e63-4508-b1ff-98f805577f2c"),
        username="user",
        email="user@gmail.com",
        phoneNumber="+57 300 500 2000",
        dni="10100190",
        fullName="Usuario Perez Gómez",
        passwordHash="a68f7b00815178c7996ddc88208224198a584ce22faf75b19bfeb24ed6f90a59",
        salt="ES25GfW7i4Pp1BqXtASUFXJFe9PMb_7o-2v73v3svWc",
        token="eHBL1jbhBY6GfZ96DC03BlxM38SPF3npRBceefRgnkTpByFexOe7RPPDdLCh9gejD6Fe6Kdl_s5C3Gljqh3WM2xW1IGdlZQYg"
              "V0_v55tw_NB19oMzH2t9AjKycEDdwmqPFJVR4sZuk9MFvSGoY_vQa4Y0pwCvxhBDT1VNsDnQio",
        status=UserStatusEnum.NO_VERIFICADO,
        expireAt=datetime.datetime.now(),
        createdAt=datetime.datetime.now(),
        updateAt=datetime.datetime.now()
    )
    db_session.add(mock_user)
    db_session.flush()  # 同步数据到数据库，不提交事务
    
    # 验证测试数据已存在
    assert db_session.scalar(select(func.count()).select_from(User)) == 1, "Table couldn't be set up successfully"

    # 调用重置接口
    response = client.post("/users/reset")
    assert response.status_code == 200, "The request failed for an unknown reason"
    assert "los datos fueron eliminados" in str(response.json()['msg'])
    
    # 验证表已清空
    assert db_session.scalar(select(func.count()).select_from(User)) == 0, "Table reset was unsuccessful"

关键优化点说明 #

• 共享测试Session：通过依赖覆盖让接口和测试用例共用同一个Session，避免多Session竞争连接导致死锁。
• 函数级事务隔离：每个测试用例独立执行，结束后自动回滚事务，完全不污染数据库。
• flush替代commit：测试中仅同步数据到数据库，不提交事务，既保证验证准确性，又避免与全局测试事务冲突。
• 简化Session使用：移除路由中不必要的Session上下文块，减少事务嵌套逻辑。

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