Транзакции и управление соединениями¶

Использование SAVEPOINT¶

Транзакции SAVEPOINT, если они поддерживаются базовым механизмом, могут быть разграничены с помощью метода Session.begin_nested():

Session = sessionmaker()

with Session.begin() as session:
    session.add(u1)
    session.add(u2)

    nested = session.begin_nested()  # establish a savepoint
    session.add(u3)
    nested.rollback()  # rolls back u3, keeps u1 and u2

# commits u1 and u2

При каждом вызове Session.begin_nested() в базу данных в рамках текущей транзакции базы данных (если она еще не началась) передается новая команда «BEGIN SAVEPOINT» и возвращается объект типа SessionTransaction, который представляет собой хэндл к этому SAVEPOINT. При вызове метода .commit() на этом объекте в базу данных передается сообщение «RELEASE SAVEPOINT», а если вместо этого вызывается метод .rollback(), то передается сообщение «ROLLBACK TO SAVEPOINT». Охватывающая транзакция базы данных продолжает выполняться.

Session.begin_nested() обычно используется в качестве менеджера контекста, где могут быть пойманы конкретные ошибки отдельных экземпляров, в сочетании с откатом, созданным для этой части состояния транзакции, без отката всей транзакции, как в примере ниже:

for record in records:
    try:
        with session.begin_nested():
            session.merge(record)
    except:
        print("Skipped record %s" % record)
session.commit()

Когда менеджер контекста, переданный Session.begin_nested(), завершает работу, он «фиксирует» точку сохранения, что включает в себя обычное поведение очистки всех ожидающих состояний. При возникновении ошибки точка сохранения откатывается, и состояние Session, локальное для объектов, которые были изменены, становится недействительным.

Этот шаблон идеально подходит для таких ситуаций, как использование PostgreSQL и перехват IntegrityError для обнаружения дублирующихся строк; PostgreSQL обычно прерывает всю транзакцию при возникновении такой ошибки, однако при использовании SAVEPOINT внешняя транзакция сохраняется. В приведенном ниже примере список данных сохраняется в базе данных, при этом случайные записи с «дублирующимся первичным ключом» пропускаются, без отката всей операции:

from sqlalchemy import exc

with session.begin():
    for record in records:
        try:
            with session.begin_nested():
                obj = SomeRecord(id=record["identifier"], name=record["name"])
                session.add(obj)
        except exc.IntegrityError:
            print(f"Skipped record {record} - row already exists")

Когда вызывается Session.begin_nested(), Session сначала сбрасывает все текущее состояние в базу данных; это происходит безоговорочно, независимо от значения параметра Session.autoflush, который обычно может использоваться для отключения автоматического сбрасывания. Такое поведение объясняется тем, что при откате этой вложенной транзакции Session может уничтожить все состояния в памяти, которые были созданы в рамках SAVEPOINT, гарантируя при этом, что при обновлении объектов с истекшим сроком хранения состояние графа объектов до начала SAVEPOINT будет доступно для повторной загрузки из базы данных.

В современных версиях SQLAlchemy при откате SAVEPOINT, инициированного командой Session.begin_nested(), состояние объекта в памяти, которое было изменено с момента создания SAVEPOINT, уничтожается, однако состояние объекта, которое не было изменено с момента создания SAVEPOINT, сохраняется. Это делается для того, чтобы последующие операции могли продолжать использовать незатронутые данные без необходимости их обновления из базы данных.

Управление транзакциями на уровне сеанса и на уровне двигателя¶

Connection в Core и _session.Session в ORM имеют эквивалентную транзакционную семантику, как на уровне sessionmaker против Engine, так и Session против Connection. В следующих разделах подробно описаны эти сценарии, основанные на следующей схеме:

ORM                                           Core
-----------------------------------------     -----------------------------------
sessionmaker                                  Engine
Session                                       Connection
sessionmaker.begin()                          Engine.begin()
some_session.commit()                         some_connection.commit()
with some_sessionmaker() as session:          with some_engine.connect() as conn:
with some_sessionmaker.begin() as session:    with some_engine.begin() as conn:
with some_session.begin_nested() as sp:       with some_connection.begin_nested() as sp:

engine = create_engine("postgresql+psycopg2://user:pass@host/dbname")

with engine.connect() as conn:
    conn.execute(
        some_table.insert(),
        [
            {"data": "some data one"},
            {"data": "some data two"},
            {"data": "some data three"},
        ],
    )
    conn.commit()

Session = sessionmaker(engine)

with Session() as session:
    session.add_all(
        [
            SomeClass(data="some data one"),
            SomeClass(data="some data two"),
            SomeClass(data="some data three"),
        ]
    )
    session.commit()

engine = create_engine("postgresql+psycopg2://user:pass@host/dbname")

with engine.begin() as conn:
    conn.execute(
        some_table.insert(),
        [
            {"data": "some data one"},
            {"data": "some data two"},
            {"data": "some data three"},
        ],
    )
# commits and closes automatically

Session = sessionmaker(engine)

with Session.begin() as session:
    session.add_all(
        [
            SomeClass(data="some data one"),
            SomeClass(data="some data two"),
            SomeClass(data="some data three"),
        ]
    )
# commits and closes automatically

engine = create_engine("postgresql+psycopg2://user:pass@host/dbname")

with engine.begin() as conn:
    savepoint = conn.begin_nested()
    conn.execute(
        some_table.insert(),
        [
            {"data": "some data one"},
            {"data": "some data two"},
            {"data": "some data three"},
        ],
    )
    savepoint.commit()  # or rollback

# commits automatically

Session = sessionmaker(engine)

with Session.begin() as session:
    savepoint = session.begin_nested()
    session.add_all(
        [
            SomeClass(data="some data one"),
            SomeClass(data="some data two"),
            SomeClass(data="some data three"),
        ]
    )
    savepoint.commit()  # or rollback
# commits automatically

Явное начало¶

Session имеет поведение «автозапуска», означающее, что как только операции начинают выполняться, он обеспечивает наличие SessionTransaction для отслеживания текущих операций. Эта операция завершается при вызове Session.commit().

Часто желательно, особенно при интеграции фреймворков, контролировать момент, когда происходит операция «begin». Для этого Session использует стратегию «autobegin», так что метод Session.begin() может быть вызван непосредственно для Session, у которого еще не началась транзакция:

Session = sessionmaker(bind=engine)
session = Session()
session.begin()
try:
    item1 = session.get(Item, 1)
    item2 = session.get(Item, 2)
    item1.foo = "bar"
    item2.bar = "foo"
    session.commit()
except:
    session.rollback()
    raise

Приведенный выше шаблон более идиоматично вызывается с помощью менеджера контекста:

Session = sessionmaker(bind=engine)
session = Session()
with session.begin():
    item1 = session.get(Item, 1)
    item2 = session.get(Item, 2)
    item1.foo = "bar"
    item2.bar = "foo"

Метод Session.begin() и процесс «autobegin» сессии используют одну и ту же последовательность шагов для начала транзакции. Это включает вызов события SessionEvents.after_transaction_create() при его возникновении; этот хук используется фреймворками для интеграции собственных транзакционных процессов с процессами ORM Session.

Включение двухфазной фиксации¶

Для баз данных, поддерживающих двухфазную работу (в настоящее время это MySQL и PostgreSQL), сессии можно указать использовать семантику двухфазной фиксации. Это позволит координировать фиксацию транзакций между базами данных таким образом, что транзакция будет либо зафиксирована, либо откачена во всех базах данных. Вы также можете Session.prepare() использовать сессию для взаимодействия с транзакциями, не управляемыми SQLAlchemy. Для использования двухфазных транзакций установите флаг twophase=True на сессии:

engine1 = create_engine("postgresql+psycopg2://db1")
engine2 = create_engine("postgresql+psycopg2://db2")

Session = sessionmaker(twophase=True)

# bind User operations to engine 1, Account operations to engine 2
Session.configure(binds={User: engine1, Account: engine2})

session = Session()

# .... work with accounts and users

# commit.  session will issue a flush to all DBs, and a prepare step to all DBs,
# before committing both transactions
session.commit()

Настройка уровней изоляции транзакций / DBAPI AUTOCOMMIT¶

Большинство DBAPI поддерживают концепцию настраиваемых уровней транзакций isolation. Традиционно это четыре уровня «READ UNCOMMITTED», «READ COMMITTED», «REPEATABLE READ» и «SERIALIZABLE». Они обычно применяются к соединению DBAPI перед началом новой транзакции, при этом следует отметить, что большинство DBAPI начинают транзакцию неявно при первом выполнении SQL-запросов.

DBAPI, поддерживающие уровни изоляции, обычно также поддерживают концепцию истинного «автокоммита», что означает, что само соединение DBAPI будет переведено в нетранзакционный режим автокоммита. Это обычно означает, что типичное поведение DBAPI, заключающееся в автоматической передаче «BEGIN» в базу данных, больше не происходит, но оно может включать и другие директивы. При использовании этого режима ** DBAPI не использует транзакцию ни при каких обстоятельствах**. Методы SQLAlchemy, такие как .begin(), .commit() и .rollback(), проходят молча.

Диалекты SQLAlchemy поддерживают настраиваемые режимы изоляции на основе per-Engine или per-Connection, используя флаги как на уровне create_engine(), так и на уровне Connection.execution_options().

При использовании ORM Session он действует как фасад для движков и соединений, но не обеспечивает изоляцию транзакций напрямую. Поэтому для того, чтобы повлиять на уровень изоляции транзакций, мы должны действовать на Engine или Connection в зависимости от ситуации.

from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker

eng = create_engine(
    "postgresql+psycopg2://scott:tiger@localhost/test",
    isolation_level="REPEATABLE READ",
)

Session = sessionmaker(eng)

from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker

eng = create_engine("postgresql+psycopg2://scott:tiger@localhost/test")

autocommit_engine = eng.execution_options(isolation_level="AUTOCOMMIT")

transactional_session = sessionmaker(eng)
autocommit_session = sessionmaker(autocommit_engine)

with autocommit_session() as session:
    some_objects = session.execute("<statement>")
    some_other_objects = session.execute("<statement>")

# closes connection

plain_engine = create_engine("postgresql+psycopg2://scott:tiger@localhost/test")

autocommit_engine = plain_engine.execution_options(isolation_level="AUTOCOMMIT")

# will normally use plain_engine
Session = sessionmaker(plain_engine)

# make a specific Session that will use the "autocommit" engine
with Session(bind=autocommit_engine) as session:
    # work with session
    ...

with Session() as session:
    session.bind_mapper(User, autocommit_engine)

from sqlalchemy.orm import Session

# assume session just constructed
sess = Session(bind=engine)

# call connection() with options before any other operations proceed.
# this will procure a new connection from the bound engine and begin a real
# database transaction.
sess.connection(execution_options={"isolation_level": "SERIALIZABLE"})

# ... work with session in SERIALIZABLE isolation level...

# commit transaction.  the connection is released
# and reverted to its previous isolation level.
sess.commit()

# subsequent to commit() above, a new transaction may be begun if desired,
# which will proceed with the previous default isolation level unless
# it is set again.

sess = Session(bind=engine)

with sess.begin():
    # call connection() with options before any other operations proceed.
    # this will procure a new connection from the bound engine and begin a
    # real database transaction.
    sess.connection(execution_options={"isolation_level": "SERIALIZABLE"})

    # ... work with session in SERIALIZABLE isolation level...

# outside the block, the transaction has been committed.  the connection is
# released and reverted to its previous isolation level.

Отслеживание состояния транзакции с помощью событий¶

Обзор доступных крючков событий для изменения состояния транзакции сеанса см. в разделе События транзакции.