在测试期间跨 SQLAlchemy 会话重用分离的 ORM 对象

Question

我试图理解 SQL-Alchemy 对 orm 对象做了什么。他们一定有某种状态附加在他们身上，我不明白。在下面的示例中，

test_one

将成功，但

test_two

将失败，因为 Employee 对象不会添加到

test_two

中的会话中。在 sqla 内部的某个地方，它看到该对象已经添加到会话中并从会话中删除，即使会话在模块范围内关闭，因此

test_two

应该有一个全新的会话。我的解决方法是将默认值包装在一个函数中，并每次返回 orm 对象的新实例。仍然好奇发生了什么？

我尝试过的一些事情： 当我关闭会话时，调用

expunge_all

、

expire_all

、

expunge

对象等。 在

add_all

中第二次调用

test_two

时，对象被添加到会话

identity_map

但在

commit()

上它永远不会添加到数据库中。因此第二个断言失败了。

这是一个使用 Flask-SQLAlchemy 和 SQLAlchemy 1.4 的 Flask 应用程序

默认值.py

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

conftest.py

mysql_proc

是pytest-mysql提供的fixture

@pytest.fixture(scope="session")
def app(mysql_proc):
    url = f'mysql+mysqldb://root:@127.0.0.1:3307'
    # create the test database before we connect using 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)

    # this will use Flask-SQLAlchemy's session from now on
    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

Answer 1

看来除了打

make_transient

还得打

expunge

。如果打印对象的

__dict__

，您可以看到对象中仍然存在状态。

{'_sa_instance_state': <sqlalchemy.orm.state.InstanceState object at 0x7fb3957f2a40>, 'employee_id': 8, 'given_name': 'first', 'family_name': 'last', 'email': '[email protected]'}

尽管我认为返回新鲜对象的函数的原始解决方案可能是一个更好的主意。无论如何，员工应该只是固定资产。

sqlalchemy.orm.make_transient