测试中跨SQLAlchemy会话复用分离ORM对象的问题排查

我想搞懂SQLAlchemy对ORM对象的处理逻辑，这些对象肯定带有我尚未理解的状态。在以下示例中，test_one执行成功，但test_two会失败——因为Employee对象未被添加到test_two的会话中。尽管模块作用域的会话已关闭，test_two理应使用全新会话，但SQLAlchemy内部似乎能识别到该对象曾被加入会话且已被移除。我临时的解决方案是将默认对象包装在函数中，每次返回新的ORM实例，但仍好奇背后的原因。

已尝试的操作 #

• 关闭会话时调用expunge_all、expire_all、expunge对象等方法
• 在test_two中第二次调用add_all时，对象已存在于会话的identity_map中，但commit()后从未写入数据库，导致第二个断言失败

这是一个基于Flask-SQLAlchemy和SQLAlchemy 1.4的Flask应用。

defaults.py #

from myapp.models import Employee
EMPLOYEES = [
        Employee(employee_id=8, given_name='first', family_name='last', email='test@example.com')
]

conftest.py #

mysql_proc是pytest-mysql提供的fixture

import pytest
from sqlalchemy import create_engine, delete
from sqlalchemy.orm import scoped_session, sessionmaker
from myapp import create_app, db
from myapp.config import TestingConfig

@pytest.fixture(scope="session")
def app(mysql_proc):
    url = f'mysql+mysqldb://root:@127.0.0.1:3307'
    # 创建测试数据库，之后使用TestingConfig连接
    engine = create_engine(url, echo=False, poolclass=NullPool)
    with engine.connect() as conn:
        conn.execute('CREATE DATABASE IF NOT EXISTS mydb')
        conn.execute('USE mydb')

    url = f'mysql+mysqldb://root:@127.0.0.1:3307/mydb'
    engine = create_engine(url, echo=True, poolclass=NullPool)
    with engine.connect() as conn:
        db.metadata.create_all(bind=conn)

    # 从此开始使用Flask-SQLAlchemy的会话
    app = create_app(TestingConfig())
    yield app

@pytest.fixture(scope="session")
def db_engine(mysql_proc):
    url = f'mysql+mysqldb://root:@127.0.0.1:3307/mydb'
    engine = create_engine(url, echo=False, poolclass=NullPool)
    yield engine
    engine.dispose()

@pytest.fixture(scope="session")
def db_session_maker(db_engine):
    return scoped_session(sessionmaker(bind=db_engine))

@pytest.fixture(scope="module")
def core_session(app, db_session_maker):
    session = db_session_maker()
    yield session
    session.rollback()
    session.close()

@pytest.fixture(scope="module")
def employees_fixture(core_session):
    core_session.add_all(EMPLOYEES)
    core_session.commit()
    yield core_session
    core_session.execute(delete(Employee))
    core_session.commit()

test_one.py #

import pytest
from myapp.models import Employee

def test_one(core_session, employees_fixture):
    assert core_session.query(Employee).count() == 1

test_two.py #

import pytest
from myapp.models import Employee

def test_two(core_session, employees_fixture):
    assert core_session.query(Employee).count() == 1

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