python - 与测试无关的函数中的 SQLAlchemy 会话管理

Question

我正在开发一个 Python 应用程序，该应用程序使用以下 SQLAlchemy 模式将单元测试分离为独立的事务，并使用 pytest 固定装置：

@fixture(scope='function')
def session(engine):
    connection = engine.connect()
    transaction = connection.begin()
    Session = get_session(bind=connection)

    session = Session()
    session.begin_nested()

    @listens_for(session, 'after_transaction_end')
    def resetart_savepoint(sess, trans):
        if trans.nested and not trans._parent.nested:
            session.expire_all()
            session.begin_nested()

    yield session

    session.close()
    transaction.rollback()
    connection.close()

我正在定义get_session()实用程序函数，以便我也可以在我的应用程序代码中使用它：

def get_session(bind=None):
    if bind is None:
        bind = create_engine(DATABASE_URL)
    return scoped_session(sessionmaker(bind=bind))

这种模式在session直接使用fixture的单元测试代码中完美地工作：

def test_foo(session):
    assert session.query(Foo).count() == 0

但是，每当我想测试应该自己访问会话的应用程序代码时，如果它没有从单元测试中显式传递，我就会失败：

def create_foo_obj():
    Session = get_session()
    session = Session()
    session.add(Foo())
    session.commit()

def test_foo(session):
    create_foo_obj()
    assert session.query(Foo).count() == 1  # FAILS

由于我使用的是 ascoped_session我希望任何调用的应用程序代码都get_session()将获得与测试夹具绑定的相同会话，但这不会发生。

我知道我可以通过将session直接传递给函数来解决这个问题，但这似乎是错误的。

有没有更好的方法来实现这一点？

标签： pythonpython-3.xsqlalchemypytest