Python の with 文でデータベース接続を管理する

データベース接続は、開いたら必ず閉じる必要があるリソースの代表例です。with 文を使うことで、接続のクローズやトランザクションのコミット・ロールバックを自動化できます。

sqlite3 の基本

Python 標準の sqlite3 モジュールは、接続オブジェクトがコンテキストマネージャをサポートしています。

import sqlite3

with sqlite3.connect("test.db") as conn:
    cursor = conn.cursor()
    cursor.execute("CREATE TABLE IF NOT EXISTS users (id INTEGER, name TEXT)")
    cursor.execute("INSERT INTO users VALUES (1, 'Alice')")
    # ブロック終了時に自動コミット

with ブロックを正常に抜けると自動的にコミットされ、例外が発生するとロールバックされます。

トランザクション管理

正常終了

自動的に COMMIT される

例外発生

自動的に ROLLBACK される

import sqlite3

try:
    with sqlite3.connect("test.db") as conn:
        cursor = conn.cursor()
        cursor.execute("INSERT INTO users VALUES (2, 'Bob')")
        raise ValueError("エラー発生")  # ここで例外
except ValueError:
    print("ロールバックされました")

# Bob は挿入されていない

カスタム接続マネージャ

より細かい制御が必要な場合は、カスタムのコンテキストマネージャを作成します。

import sqlite3
from contextlib import contextmanager

@contextmanager
def db_connection(db_path):
    conn = sqlite3.connect(db_path)
    try:
        yield conn
    except Exception:
        conn.rollback()
        raise
    else:
        conn.commit()
    finally:
        conn.close()

with db_connection("test.db") as conn:
    cursor = conn.cursor()
    cursor.execute("SELECT * FROM users")
    print(cursor.fetchall())

カーソルもコンテキストマネージャに

接続とカーソルの両方を管理するコンテキストマネージャです。

from contextlib import contextmanager
import sqlite3

@contextmanager
def db_cursor(db_path):
    conn = sqlite3.connect(db_path)
    cursor = conn.cursor()
    try:
        yield cursor
    except Exception:
        conn.rollback()
        raise
    else:
        conn.commit()
    finally:
        cursor.close()
        conn.close()

with db_cursor("test.db") as cursor:
    cursor.execute("SELECT * FROM users")
    for row in cursor.fetchall():
        print(row)

接続プールとの連携

実際のアプリケーションでは、接続プールを使うことが多いです。

from contextlib import contextmanager
import sqlite3
from queue import Queue

class ConnectionPool:
    def __init__(self, db_path, pool_size=5):
        self.db_path = db_path
        self.pool = Queue(maxsize=pool_size)
        for _ in range(pool_size):
            self.pool.put(sqlite3.connect(db_path))
    
    @contextmanager
    def get_connection(self):
        conn = self.pool.get()
        try:
            yield conn
            conn.commit()
        except Exception:
            conn.rollback()
            raise
        finally:
            self.pool.put(conn)

pool = ConnectionPool("test.db")

with pool.get_connection() as conn:
    cursor = conn.cursor()
    cursor.execute("SELECT COUNT(*) FROM users")
    print(cursor.fetchone()[0])

他のデータベースライブラリ

多くのデータベースライブラリがコンテキストマネージャをサポートしています。

# psycopg2 (PostgreSQL)
import psycopg2

with psycopg2.connect("dbname=test") as conn:
    with conn.cursor() as cursor:
        cursor.execute("SELECT * FROM users")

# MySQL Connector
import mysql.connector

with mysql.connector.connect(host="localhost", database="test") as conn:
    with conn.cursor() as cursor:
        cursor.execute("SELECT * FROM users")

データベース操作では必ず with 文を使いましょう。接続リークを防ぎ、トランザクションを安全に管理できます。