Transactions and Connection Management

Managing Transactions

Session は、 SessionTransaction というオブジェクトを使って、一度に1つの”仮想”トランザクションの状態を追跡します。このオブジェクトは、必要に応じて Connection オブジェクトを使って実際の接続レベルのトランザクションを開始するために、基礎となる Engine または Session オブジェクトがバインドされているエンジンを利用します。

この”仮想”トランザクションは、必要に応じて自動的に作成されるか、または Session.begin() メソッドを使用して開始することができます。Pythonコンテキストマネージャの使用は、可能な限り、 Session オブジェクトを作成するレベルと、 SessionTransaction のスコープを維持するレベルの両方でサポートされています。

以下では、 Session から始めるものとします。:

from sqlalchemy.orm import Session

session = Session(engine)

コンテキストマネージャを使用して、境界設定されたトランザクション内で操作を実行できるようになりました。:

with session.begin():
    session.add(some_object())
    session.add(some_other_object())
# commits transaction at the end, or rolls back if there
# was an exception raised

上記のコンテキストの最後で、例外が発生しなかったと仮定すると、保留中のオブジェクトはデータベースにフラッシュされ、データベーストランザクションがコミットされます。上記のブロック内で例外が発生した場合、トランザクションはロールバックされます。どちらの場合も、ブロックを終了した後の上記の Session は、後続のトランザクションで使用できるようになります。

Session.begin() メソッドはオプションです。また、 Session はcommit-as-you-goアプローチでも使用できます。このアプローチでは、必要に応じて自動的にトランザクションが開始されます。コミットまたはロールバックのみが必要です:

session = Session(engine)

session.add(some_object())
session.add(some_other_object())

session.commit()  # commits

# will automatically begin again
result = session.execute(text("< some select statement >"))
session.add_all([more_objects, ...])
session.commit()  # commits

session.add(still_another_object)
session.flush()  # flush still_another_object
session.rollback()  # rolls back still_another_object

Session 自体には Session.close() メソッドがあります。まだコミットもロールバックもされていないトランザクション内で Session が開始された場合、このメソッドはそのトランザクションをキャンセル(つまりロールバック)し、 Session オブジェクトの状態に含まれるすべてのオブジェクトを削除します。 Session が、 Session.commit() または Session.rollback() の呼び出しが保証されないような方法で使用されている場合(例えばコンテキストマネージャ内などで)、 close メソッドを使用して、すべてのリソースが確実に解放されるようにすることができます:

# expunges all objects, releases all transactions unconditionally
# (with rollback), releases all database connections back to their
# engines
session.close()

最後に、セッションの構築/終了プロセス自体をコンテキストマネージャ経由で実行することができます。これは、 Session オブジェクトの使用のスコープが固定ブロック内でスコープされることを保証する最善の方法です。最初に Session コンストラクタで説明します:

with Session(engine) as session:
    session.add(some_object())
    session.add(some_other_object())

    session.commit()  # commits

    session.add(still_another_object)
    session.flush()  # flush still_another_object

    session.commit()  # commits

    result = session.execute(text("<some SELECT statement>"))

# remaining transactional state from the .execute() call is
# discarded

同様に、 sessionmaker も同じように使えます:

Session = sessionmaker(engine)

with Session() as session:
    with session.begin():
        session.add(some_object)
    # commits

# closes the Session

sessionmaker 自体には sessionmaker.begin() メソッドがあり、両方の操作を同時に行うことができます:

with Session.begin() as session:
    session.add(some_object)

Using 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”コマンドがデータベースに発行され(まだ進行中でなければ開始されます)、このSAVEPOINTへのハンドルを表す SessionTransaction 型のオブジェクトが返されます。このオブジェクトの .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では、 Session.begin_nested() によって開始されたSAVEPOINTがロールバックされると、SAVEPOINTが作成された後に変更されたメモリ内のオブジェクト状態は期限切れになりますが、SAVEPOINTが開始された後に変更されなかった他のオブジェクト状態は維持されます。これは、後続の操作が、データベースからデータを更新することなく、影響を受けていないデータを引き続き使用できるようにするためです。

See also

Connection.begin_nested() - Core SAVEPOINT API

Session-level vs. Engine level transaction control

Coreの Connection とORMの _session.Session は、どちらも 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:

Commit as you go

Session と Connection の両方の機能 Connection.commit() メソッドと Connection.rollback() メソッドです。SQLAlchemy 2.0スタイルの操作を使用すると、これらのメソッドはすべての場合において**最も外側の**トランザクションに影響します。 Session の場合、 Session.autobegin はデフォルト値の True のままであると想定されます。

Engine:

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:

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()

Begin Once

sessionmaker と Engine はどちらも Engine.begin() メソッドを備えています。このメソッドは、SQL文を実行するための新しいオブジェクト(それぞれ Session と Connection )を取得し、そのオブジェクトのbegin/commit/rollbackコンテキストを保持するコンテキストマネージャを返します。

Engine:

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:

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

Nested Transaction

Session.begin_nested() または Connection.begin_nested() メソッドを介してSAVEPOINTを使用する場合、返されたトランザクションオブジェクトを使用してSAVEPOINTをコミットまたはロールバックする必要があります。 Session.commit() または Connection.commit() メソッドを呼び出すと、常に 最も外側の トランザクションがコミットされます。これはSQLAlchemy 2.0固有の動作で、1.xシリーズとは逆です。

Engine:

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:

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

Explicit Begin

Session は”autobegin”動作を特徴とします。つまり、操作が開始されるとすぐに、進行中の操作を追跡するための SessionTransaction が存在することを保証します。このトランザクションは Session.commit() が呼ばれた時に完了します。

特にフレームワーク統合では、”begin”操作が発生するポイントを制御することが望ましいことがよくあります。これに対応するために、 Session は”autobegin”戦略を使用しています。たとえば、まだトランザクションが開始されていない Session に対して、 Session.begin() メソッドを直接呼び出すことができます:

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 のトランザクション処理と統合するために使用されます。

Enabling Two-Phase Commit

2フェーズ操作をサポートするバックエンド(現在はMySQLとPostgreSQL)では、セッションに2フェーズコミットセマンティクスを使用するように指示できます。これにより、トランザクションがすべてのデータベースでコミットまたはロールバックされるように、データベース間でトランザクションのコミットが調整されます。また、SQLAlchemyで管理されていないトランザクションと対話するために、セッションを Session.prepare() することもできます。2フェーズトランザクションを使用するには、セッションにフラグ 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()

Setting Transaction Isolation Levels / DBAPI AUTOCOMMIT

ほとんどのDBAPIは、設定可能なトランザクション isolation レベルの概念をサポートしています。これらは伝統的に”READ UNCOMMITTED”、”READ COMMITTED”、”REPEATABLE READ”、”SERIALIZABLE”の4つのレベルです。これらは通常、新しいトランザクションを開始する前にDBAPI接続に適用されますが、ほとんどのDBAPIはSQL文が最初に発行されたときに暗黙的にこのトランザクションを開始することに注意してください。

独立性レベルをサポートするDBAPIは通常、真の “autocommit” の概念もサポートしています。これは、DBAPI接続自体が非トランザクション・オートコミット・モードになることを意味します。これは通常、データベースに BEGIN を自動的に発行するという一般的なDBAPIの動作が発生しなくなることを意味しますが、他のディレクティブが含まれる場合もあります。このモードを使用する場合、 DBAPIはいかなる状況においてもトランザクションを使用しません 。SQLAlchemyのメソッド(例えば .begin() 、 .commit() 、 .rollback() など)は黙って通過します。

SQLAlchemyのダイアレクトは、 Engine または Connection ごとに設定可能な分離モードをサポートしており、 create_engine() レベルと Connection.execution_options() レベルの両方でフラグを使用します。

ORM Session を使用する場合、これはエンジンとコネクションの ファサード として動作しますが、トランザクションの独立性を直接公開することはありません。したがって、トランザクションの独立性レベルに影響を与えるためには、 Engine または Connection を適切に操作する必要があります。

See also

Setting Transaction Isolation Levels including DBAPI Autocommit - be sure to review how isolation levels work at the level of the SQLAlchemy Connection object as well.

Setting Isolation For A Sessionmaker / Engine Wide

特定の独立性レベルを持つ Session または sessionmaker をグローバルに設定するための最初のテクニックは、 Engine を特定の独立性レベルに対して構築し、それを Session や sessionmaker の接続元として使用することです:

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)

異なる独立性レベルを持つ2つのエンジンが同時に存在する場合に便利なもう1つのオプションは、 Engine.execution_options() メソッドを使用することです。このメソッドは、親エンジンと同じ接続プールを共有する元の Engine のシャローコピーを生成します。これは、操作が「トランザクション」操作と「オートコミット」操作に分離される場合によく好まれます。:

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)

上記では、 “eng” と "autocommit_engine" は同じダイアレクトと接続プールを共有しています。しかし、 autocommit_engine から取得された接続には AUTOCOMMIT モードが設定されます。2つの sessionmaker オブジェクトである transactional_session と "autocommit_session" は、データベース接続で動作するときにこれらの特性を継承します。

Session.commit() や Session.rollback() などの”autocommit_session”は、 トランザクションのセマンティクスを保持しています 。これらのセマンティクスは、オブジェクトの”コミット”や”ロールバック”と見なされますが、トランザクションは暗黙的に存在しません。このため、 厳密には必須ではありませんが、AUTOCOMMITで分離されたセッションを読み取り専用で使用するのが一般的です 。つまり:

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

# closes connection

Setting Isolation for Individual Sessions

コンストラクタを直接使用して、または sessionmaker によって生成された呼び出し可能オブジェクトを呼び出して、新しい Session を作成する場合、既存のバインドを上書きして、直接 bind 引数を渡すことができます。たとえば、デフォルトの sessionmaker から Session を作成し、オートコミット用のエンジンセットを渡すことができます:

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
    ...

Session または sessionmaker が複数の”bind”で設定されている場合、引数``binds``を完全に再指定するか、特定のbindだけを置き換えたい場合は Session.bind_mapper() または Session.bind_table() メソッドを使用します:

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

Setting Isolation for Individual Transactions

独立性レベルに関する重要な注意点は、トランザクションがすでに開始されている Connection では、設定を安全に変更できないことです。データベースは進行中のトランザクションの独立性レベルを変更することはできず、一部のDB APIとSQLAlchemyダイアレクトでは、この領域での動作に一貫性がありません。

したがって、必要な独立性レベルを持つエンジンに前もってバインドされた Session を使用することをお勧めします。ただし、トランザクションの開始時に Session.connection() メソッドを使用すると、接続ごとの独立性レベルが影響を受ける可能性があります。:

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.

上記では、まずコンストラクタか sessionmaker を使って Session を生成します。次に Session.connection() を呼び出して、データベースレベルのトランザクションの開始を明示的に設定します。これは、データベースレベルのトランザクションが開始される前に接続に渡される実行オプションを提供します。トランザクションは、この選択された独立性レベルで進行します。トランザクションが完了すると、接続が接続プールに戻される前に、接続の独立性レベルがデフォルトにリセットされます。

Session.begin() メソッドは、 Session レベルのトランザクションを開始するためにも使用できます。この呼び出しの後に続く Session.connection() の呼び出しは、接続トランザクションごとの独立性レベルを設定するために使用できます:

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.

Tracking Transaction State with Events

セッショントランザクション状態の変更に利用可能なイベントフックの概要については、 Transaction Events を参照してください。

Joining a Session into an External Transaction (such as for test suites)

すでにトランザクション状態にある(つまり Transaction が確立されている) Connection が使用されている場合、 Session をその Connection にバインドするだけで、そのトランザクション内に Session を参加させることができます。これの一般的な理論的根拠は、ORMコードが Session と自由に連携できるようにするテストスイートであり、これには Session.commit() を呼び出す機能が含まれ、その後、データベースの相互作用全体がロールバックされます。

Changed in version 2.0: “外部トランザクションへの結合”レシピは、2.0で再び新たに改善されました。ネストされたトランザクションを「リセット」するイベントハンドラは不要になりました。

このレシピは、トランザクション内で Connection を確立し、オプションでSAVEPOINTを確立してから、それを Session に”bind”として渡すことで動作します。 Session.join_transaction_mode パラメータには、 Session のBEGIN/COMMIT/ROLLBACKを実装するために新しいSAVEPOINTを作成する必要があることを示す "create_savepoint" 設定が渡されます。これにより、外部トランザクションは渡されたときと同じ状態になります。

テストが終了すると、外部トランザクションがロールバックされ、テスト中に変更されたデータが元に戻されます。:

from sqlalchemy.orm import sessionmaker
from sqlalchemy import create_engine
from unittest import TestCase

# global application scope.  create Session class, engine
Session = sessionmaker()

engine = create_engine("postgresql+psycopg2://...")

class SomeTest(TestCase):
    def setUp(self):
        # connect to the database
        self.connection = engine.connect()

        # begin a non-ORM transaction
        self.trans = self.connection.begin()

        # bind an individual Session to the connection, selecting
        # "create_savepoint" join_transaction_mode
        self.session = Session(
            bind=self.connection, join_transaction_mode="create_savepoint"
        )

    def test_something(self):
        # use the session in tests.

        self.session.add(Foo())
        self.session.commit()

    def test_something_with_rollbacks(self):
        self.session.add(Bar())
        self.session.flush()
        self.session.rollback()

        self.session.add(Foo())
        self.session.commit()

    def tearDown(self):
        self.session.close()

        # rollback - everything that happened with the
        # Session above (including calls to commit())
        # is rolled back.
        self.trans.rollback()

        # return connection to the Engine
        self.connection.close()

上記のレシピはSQLAlchemy独自のCIの一部であり、期待通りに動作することを保証します。