嵌套会话after_transaction_end需要 SQL Alchemy v2-解网

问：

在设置基于 SQLAlchemy 和 Pytest 的测试套件（2022 年 5 月随 SQLAlchemy v1 一起发布）中，它使用以下代码片段进行事务测试。注意：我在这篇文章中清理了一下：

# conftest.py

import pytest

# Import a session factory in our app code. Possibly created using
# `sessionmaker()`
from myapp.db import Session

@pytest.fixture(autouse=True)
def session(connection):
    transaction = connection.begin()
    session = Session(bind=connection)
    session.begin_nested()

    @event.listens_for(session, "after_transaction_end")
    def restart_savepoint(db_session, ended_transaction):
        if ended_transaction.nested and not ended_transaction._parent.nested:
            session.expire_all()
            session.begin_nested()

    yield session

    Session.remove()
    transaction.rollback()

我现在在 2023 年 11 月使用 SQLAlchemy v2，并对内部函数有三个问题：restart_savepoint

你能解释一下其中的逻辑吗？ended_transaction.nested and not ended_transaction._parent.nested
SQL Alchemy v2 是否仍需要此行为？restart_savepoint
SQL Alchemy v2 仍然需要吗？Session(bind=connection)

我认为以下内容同样可行：

@pytest.fixture(autouse=True)
def session(connection):
    with Session() as session, session.begin(nested=True):
        yield session

蟒蛇 SQL炼金术

在 SQLAlchemy v2 中，如果使用上下文管理器模式（），则通常不需要将会话显式绑定到连接。会话将自动管理连接生命周期。但是，如果您有特定要求，例如在多个会话中重用现有连接，您可能仍需要手动绑定。
您仍然可以在提交 6bf32f5 中找到它with Session() as session

您建议的代码：

@pytest.fixture(autouse=True)
def session(connection):
    with Session() as session, session.begin(nested=True):
        yield session

这是在 SQLAlchemy v2 中处理会话的一种更简洁、更惯用的方法。它利用上下文管理器来处理会话和事务生命周期，这是 SQLAlchemy v2 中的推荐做法。有关详细信息，请参阅会话基础知识。

上一个：使用 Hdbuserstore 通过 PHP/Python 连接到 hana DB

下一个：SQLAlchemy ORM - 如何表示视图和另一个表之间的 n 到 n 关系

嵌套会话after_transaction_end需要 SQL Alchemy v2

SQL Alchemy v2 is after_transaction_end needed for nested session

评论