Преимущество правильного использования точек сохранения в тестах заключается в том, что вы можете явно использовать фиксацию/откат внутри кода приложения во время теста и при этом иметь возможность откатить все в конце теста, используя оригинальная сделка. По сравнению с уровнем только одной транзакции вы не можете явно использовать откат и фиксацию в своем приложении (в таких случаях обычно вы откатываете или фиксируете в конце жизненного цикла запроса).

Здесь приведен пример синхронизации использования точек сохранения (и похоже, что он соответствует вашему второму примеру):

https://docs.sqlalchemy.org/en/20/orm/session_transaction.html#joining-a-session-into-an-external-transaction- such-as-for-test-suites

Я думаю, что второй пример лучше, если у вас есть postgresql. Хотя вам не нужно явно использовать фиксацию много раз в течение жизненного цикла запроса. Быстрая фиксация может помочь, если вам нужно открыть замок, как в этом выдуманном примере:

• SELECT orders.* FOR UPDATE WHERE order_id = 15
• быстро вести бухгалтерию, чтобы подготовить заказ
• order.state = "ready_to_ship"
• session.commit() освободить этот заказ для других изменений
• выполнить другую соответствующую работу, но без блокировки

В этом случае первый пример, вероятно, не будет работать, потому что ваш явный вызов фиксации нарушит откат вашего теста. В то время как во втором примере будет зафиксирована точка сохранения, поэтому внешняя истинная транзакция будет отменена.

Обновлять

На самом деле существуют еще более тонкие различия между «conditional_savepoint» (по умолчанию), «rollback_only», «create_savepoint» и «control_complete».

По умолчанию при использовании postgresql, если вы еще не создали точку сохранения, используется «rollback_only». При этом игнорируются фиксации внутри активной транзакции, но откаты распространяются на внешнюю транзакцию. Таким образом, первый пример выше не позволит последующим фиксациям или взаимодействию работать. Тогда как при использовании «create_savepoint» сеанс продолжит работать.

В приведенном ниже примере показан случай, когда код завершится сбоем:

import os
import contextlib

import pytest
from sqlalchemy import (
    Column,
    String,
    BigInteger,
    create_engine,
)
from sqlalchemy.sql import (
    select,
    or_,
)
from sqlalchemy.orm import (
    declarative_base,
    Session,
)


def get_engine(env):
    return create_engine(f"postgresql+psycopg2://{env['DB_USER']}:{env['DB_PASSWORD']}@{env['DB_HOST']}:{env['DB_PORT']}/{env['DB_NAME']}", echo=True, echo_pool='debug')

Base = declarative_base()


class Order(Base):
    __tablename__ = 'orders'
    id = Column(BigInteger, primary_key=True)
    state = Column(String, nullable=False)


def main():
    #
    # We run this script to setup the testing database.
    #
    engine = get_engine(os.environ)

    with engine.begin() as conn:
        Base.metadata.create_all(conn)

    with engine.connect() as conn:
        populate(conn)

    run_test(engine)


def populate(conn):
    with Session(conn) as session:
        session.add_all([Order(id=1, state='start'), Order(id=2, state='processing'), Order(id=3, state='finished')])
        session.commit()


def run_test(engine):

    # the session we get should already be bound to a connection
    # with a active transaction
    with our_session_maker(engine) as session:
        order = session.execute(select(Order)).scalars().first()
        order.state = 'error'
        # Actually write changes to the db.
        session.flush()
        # Now try to roll them back.
        session.rollback()
        # Try to keep using the session.
        # Now try to make another change but commit this time.
        order.state = 'fixed'
        session.commit()

    # When this session is created it should be "clean".
    # Ie. not in a transaction.
    with Session(engine) as session:
        # The state change should never occur.
        assert not session.execute(select(Order).where(or_(Order.state == 'error', Order.state == 'fixed'))).scalars().all()


@contextlib.contextmanager
def our_session_maker(engine):
    with engine.connect() as connection:
        with connection.begin() as transaction:
            # one of conditional_savepoint, create_savepoint, control_fully, rollback_only.
            # The default is join_transaction_mode='conditional_savepoint'.
            with Session(bind=connection) as s:
                yield s
                transaction.rollback()


if __name__ == '__main__':
    main()

Это создает подобную обратную трассировку (сокращенную). Вы можете видеть, что исключение возникает во втором коммите.

Traceback (most recent call last):
  File "/app/scripts/testing_sync_session_pytest.py", line 89, in <module>
    main()
  File "/app/scripts/testing_sync_session_pytest.py", line 45, in main
    run_test(engine)
  File "/app/scripts/testing_sync_session_pytest.py", line 68, in run_test
    session.commit()
...
sqlalchemy.exc.InvalidRequestError: Can't operate on closed transaction inside context manager.  Please complete the context manager before emitting further commands.

Журналы в этом случае прекращаются после отката, но вы можете видеть это:

• транзакция начата
• база данных изменена (через сброс)
• происходит откат
• немедленная ошибка, когда сеанс снова пытается получить доступ к базе данных

2024-08-04 22:51:49,974 INFO sqlalchemy.engine.Engine BEGIN (implicit)
2024-08-04 22:51:49,975 INFO sqlalchemy.engine.Engine SELECT orders.id, orders.state 
FROM orders
2024-08-04 22:51:49,975 INFO sqlalchemy.engine.Engine [generated in 0.00015s] {}
2024-08-04 22:51:49,976 INFO sqlalchemy.engine.Engine UPDATE orders SET state=%(state)s WHERE orders.id = %(orders_id)s
2024-08-04 22:51:49,976 INFO sqlalchemy.engine.Engine [generated in 0.00008s] {'state': 'error', 'orders_id': 1}
2024-08-04 22:51:49,977 INFO sqlalchemy.engine.Engine ROLLBACK

Если используется join_transaction_mode='create_savepoint', этот пример работает как положено. Т.е.

@contextlib.contextmanager
def our_session_maker(engine):
    with engine.connect() as connection:
        with connection.begin() as transaction:
            with Session(bind=connection, join_transaction_mode='create_savepoint') as s:
                yield s
                transaction.rollback()

Глядя на журналы, которые вы можете увидеть (при использовании create_savepoint):

• транзакция начинается
• запускается точка сохранения
• делается первое изменение (через сброс)
• это изменение отменяется
• запускается другая точка сохранения
• это изменение зафиксировано (точка сохранения выпущена)
• вся транзакция откатывается (как нам хотелось бы при использовании pytest)

2024-08-04 22:48:12,020 INFO sqlalchemy.engine.Engine BEGIN (implicit)
2024-08-04 22:48:12,020 INFO sqlalchemy.engine.Engine SAVEPOINT sa_savepoint_1
2024-08-04 22:48:12,020 INFO sqlalchemy.engine.Engine [no key 0.00006s] {}
2024-08-04 22:48:12,020 INFO sqlalchemy.engine.Engine SELECT orders.id, orders.state 
FROM orders
2024-08-04 22:48:12,020 INFO sqlalchemy.engine.Engine [generated in 0.00007s] {}
2024-08-04 22:48:12,021 INFO sqlalchemy.engine.Engine UPDATE orders SET state=%(state)s WHERE orders.id = %(orders_id)s
2024-08-04 22:48:12,021 INFO sqlalchemy.engine.Engine [generated in 0.00008s] {'state': 'error', 'orders_id': 1}
2024-08-04 22:48:12,022 INFO sqlalchemy.engine.Engine ROLLBACK TO SAVEPOINT sa_savepoint_1
2024-08-04 22:48:12,022 INFO sqlalchemy.engine.Engine [no key 0.00007s] {}
2024-08-04 22:48:12,022 INFO sqlalchemy.engine.Engine SAVEPOINT sa_savepoint_2
2024-08-04 22:48:12,022 INFO sqlalchemy.engine.Engine [no key 0.00006s] {}
2024-08-04 22:48:12,023 INFO sqlalchemy.engine.Engine SELECT orders.id AS orders_id 
FROM orders 
WHERE orders.id = %(pk_1)s
2024-08-04 22:48:12,023 INFO sqlalchemy.engine.Engine [generated in 0.00007s] {'pk_1': 1}
2024-08-04 22:48:12,023 INFO sqlalchemy.engine.Engine UPDATE orders SET state=%(state)s WHERE orders.id = %(orders_id)s
2024-08-04 22:48:12,023 INFO sqlalchemy.engine.Engine [cached since 0.002251s ago] {'state': 'fixed', 'orders_id': 1}
2024-08-04 22:48:12,024 INFO sqlalchemy.engine.Engine RELEASE SAVEPOINT sa_savepoint_2
2024-08-04 22:48:12,024 INFO sqlalchemy.engine.Engine [no key 0.00005s] {}
2024-08-04 22:48:12,024 INFO sqlalchemy.engine.Engine ROLLBACK

Заключение

Было бы довольно легко использовать значение по умолчанию, "conditional_savepoint", и не заметить этого, потому что откат происходит реже, чем фиксация. Мой оригинальный сфабрикованный пример на самом деле работает нормально, но если вы используете "control_fully", он потерпит неудачу. Кажется, что для детерминированного поведения во время тестов вы бы хотели использовать "create_savepoint", если бы это было вам доступно, иначе код, который действительно работал бы в производстве, не будет работать правильно.

Больше информации

Некоторый контекст есть в разделе документации «Что нового в версии 2.0» по адресу new-transaction-join-modes-for-session

Также, как упоминалось выше, в документации есть пример присоединения-к-сессии-во-внешнюю-транзакцию-например-для-тестовых-комплектов