SQLAlchemy 2.0 #2 エンジンとトランザクション — コネクションプールと commit の 2 パターン

プログラミング言語 Python SQLAlchemy データベース
読了 5分

第 1 回ではエンジンを作り、text() でクエリを実行しました。今回はその下で起きていることの話です。コネクションプールがどう回っているのか、トランザクションはいつ始まりいつコミットされるのかを正確に知っておくと、ORM に上がってからもぶれません。実務で遭遇する「コネクション枯渇」「コミットしたのに反映されない」の答えは、すべてこの層にあります。

コネクションプール — エンジンが本当にやっていること #

DB への接続は高コストです。TCP 接続、認証、セッション初期化までミリ秒単位のコストがかかり、DB サーバーが受けられる同時接続数にも上限があります。そこでエンジンは接続を毎回作り直さず、コネクションプールに保管して再利用します。

main.py 
from sqlalchemy import create_engine

engine = create_engine(
    "postgresql+psycopg2://user:pw@localhost/mydb",
    pool_size=5,        # プールに保持するコネクション数(デフォルト 5)
    max_overflow=10,    # プールが空のとき追加で開く数(デフォルト 10)
    pool_timeout=30,    # 借りられるコネクションがないとき待つ秒数(デフォルト 30)
    pool_pre_ping=True, # 貸し出す前に生存確認をする
)

仕組みは図書館の貸し出しと同じ構造です。engine.connect() はプールからコネクションを借り、with ブロックが終わると閉じるのではなくプールに返します。知っておくべき値は 3 つです。

  • pool_size + max_overflow が実質的な上限です。デフォルトならプロセスあたり最大 15 まで同時に開け、それ以上要求されると pool_timeout だけ待って TimeoutError になります。「コネクション枯渇」エラーの正体は、たいてい返却されないコネクション(with なしで借りて close し忘れ)がプールを食いつぶしたものです。
  • プロセス数を掛け算する必要があります。 gunicorn のワーカーが 4 つならプールも 4 つで、DB が受ける接続は最大 60 になります。DB 側の max_connections と合わせて計算しなければなりません。
  • pool_pre_ping は、長く放置されてサーバー側で切断されたコネクションをはじいてくれます。「MySQL server has gone away」系のエラーに出会ったら、まずこのオプションを確認します。

SQLite のファイル DB は接続コストがほぼゼロなので、プール設定を気にすることはまずありません。プールが問題になるのは、ネットワークの向こうにある PostgreSQL や MySQL からです。

トランザクション — 自動コミットはありません #

2.0 の大原則はトランザクションは常に明示的ということです。DBAPI ドライバーはもともと自動コミットではなく、SQLAlchemy もそれを隠しません。コミットしなければ、ブロックの終わりでロールバックされます。

main.py 
from sqlalchemy import text

# この INSERT は消えます。コミットしていないからです。
with engine.connect() as conn:
    conn.execute(text("INSERT INTO memo (body) VALUES ('一時メモ')"))
# ブロック終了 → ロールバック

コミットの方法は 2 パターンです。

main.py 
# パターン 1: commit-as-you-go — 好きな地点で自分でコミット
with engine.connect() as conn:
    conn.execute(text("INSERT INTO memo (body) VALUES ('1 件目')"))
    conn.commit()  # ここまで確定
    conn.execute(text("INSERT INTO memo (body) VALUES ('2 件目')"))
    conn.commit()  # 2 つ目のトランザクションを確定

# パターン 2: begin-once — ブロック全体がひとつのトランザクション
with engine.begin() as conn:
    conn.execute(text("INSERT INTO memo (body) VALUES ('3 件目')"))
    conn.execute(text("INSERT INTO memo (body) VALUES ('4 件目')"))
# 正常終了ならコミット、例外ならロールバック

実務のデフォルトは engine.begin() を使うパターン 2 です。「全部成功するか、全部取り消すか」というトランザクションの目的に正確に合致し、コミットのし忘れが起こりえません。パターン 1 は、長いバッチ処理で中間セーブポイントを作るときのように、ひとつのコネクションでトランザクションを何度も区切る必要がある場合に使います。

テーブル定義 — MetaData と Table #

Core ではテーブルを Table オブジェクトで表現し、テーブルの一覧は MetaData が持ちます。

main.py 
from sqlalchemy import MetaData, Table, Column, Integer, String, ForeignKey

metadata = MetaData()

user_table = Table(
    "user_account",
    metadata,
    Column("id", Integer, primary_key=True),
    Column("name", String(30), nullable=False),
    Column("email", String(100), nullable=False, unique=True),
)

address_table = Table(
    "address",
    metadata,
    Column("id", Integer, primary_key=True),
    Column("user_id", ForeignKey("user_account.id"), nullable=False),
    Column("email_address", String(100), nullable=False),
)

metadata.create_all(engine)  # 存在しないテーブルだけ CREATE TABLE

create_all() は学習やプロトタイプには便利ですが、既存テーブルを変更してはくれません。カラム追加のようなスキーマ変更は、第 7 回で扱う Alembic の仕事です。

Core で CRUD — SQL を Python の式で #

text() と違い、Core の式は Python オブジェクトで SQL を組み立てます。タイプミスが文字列の中に隠れず、パラメータバインドは自動です。

main.py 
from sqlalchemy import insert, select, update, delete

# INSERT
with engine.begin() as conn:
    conn.execute(
        insert(user_table),
        [
            {"name": "佐藤", "email": "sato@example.com"},
            {"name": "鈴木", "email": "suzuki@example.com"},
        ],
    )

# SELECT
with engine.connect() as conn:
    stmt = select(user_table).where(user_table.c.name == "佐藤")
    for row in conn.execute(stmt):
        print(row.id, row.name, row.email)

# UPDATE と DELETE
with engine.begin() as conn:
    conn.execute(
        update(user_table)
        .where(user_table.c.email == "suzuki@example.com")
        .values(name="鈴木(改)")
    )
    conn.execute(delete(user_table).where(user_table.c.id == 99))
  • カラムは テーブル.c.カラム名 でアクセスします。user_table.c.name == "佐藤" は比較結果(ブール値)ではなく SQL の条件オブジェクトを作ります。Python の演算子をオーバーロードしているのです。
  • insert() に辞書のリストを渡すと、複数行を一度に入れる executemany になります。大量投入の基本形です。
  • echo=True をオンにしておくと、各式がどんな SQL にコンパイルされるかが見えます。where() の条件がすべてバインドパラメータ(? や %(name)s)として処理されることを確認できます。

この select() の文法は、次回から ORM でもそのまま使います。変わるのは user_table.c.name が User.name になることだけです。2.0 で Core を先に学ぶ甲斐がここにあります。

まとめ #

  • エンジンの実体はコネクションプールです。pool_size + max_overflow がプロセスあたりの同時接続上限で、ワーカー数を掛けて DB の max_connections と整合させる必要があります。
  • コネクション枯渇の主犯は返却されないコネクションです。コネクションは必ず with で借ります。
  • 自動コミットはありません。基本は engine.begin() でブロック全体をひとつのトランザクションにまとめ、途中コミットが必要なときだけ connect() + commit() を使います。
  • テーブルは Table と MetaData で定義し、insert、select、update、delete の式で操作します。条件は Python の演算子で書きますが、すべてバインドパラメータにコンパイルされます。
  • 次回は同じテーブルを ORM のクラスとして宣言する方法、Mapped と mapped_column を扱います。
X