SQLAlchemy 徹底解説: Python 開発での DB 連携を効率化

Python 開発において、データベース（DB）との連携は避けて通れない重要な要素です。特にリレーショナルデータベースを扱う場合、SQL を記述してデータの操作を行うことが一般的ですが、アプリケーションコード内に生の SQL 文字列が散りばめられると、保守性の低下やセキュリティリスク（SQL インジェクション）といった問題が生じやすくなります。また、DB スキーマの変更に合わせてコードを修正する手間も増大します。

これらの課題を解決し、Python での DB 連携を効率的かつ安全に行うための強力なライブラリが、SQLAlchemy です。SQLAlchemy は、Python における最も人気があり、機能豊富なデータベースツールキットおよび Object-Relational Mapper (ORM) として広く利用されています。

本記事では、SQLAlchemy を徹底的に解説します。SQLAlchemy の全体像から、低レベルなデータベース抽象化を提供する Core、そしてオブジェクト指向での DB 操作を可能にする ORM の使い方、さらに実践的な応用例やパフォーマンスに関する考慮事項まで、詳細に見ていきましょう。この記事を読むことで、SQLAlchemy を活用して Python アプリケーションでの DB 連携を次のレベルに引き上げることができるはずです。

1. はじめに: なぜ SQLAlchemy を使うのか？

Python でデータベースを操作する方法はいくつかあります。標準ライブラリには DB-API 2.0 に準拠した sqlite3 モジュールがあり、他のデータベース（PostgreSQL, MySQL, SQLite など）についてもそれぞれのデータベースベンダーやサードパーティが提供する DB-API 実装（例: psycopg2, mysql-connector-python, PyMySQL, aiosqlite など）を利用することで、Python コードから SQL を実行できます。

しかし、これらの DB-API 実装を直接使う場合、開発者は主に以下のような課題に直面します。

SQL 文字列の管理: アプリケーションコード中に SQL 文字列が直書きされることが多くなります。これは可読性、保守性を低下させ、SQL インジェクションのリスクを高めます。
データベース間の差異の吸収: データベースシステムによって SQL の方言やデータ型、接続方法などに微妙な違いがあります。複数のデータベースに対応する場合、これらの差異を吸収するコードを書く必要が出てきます。
オブジェクトとのマッピング: データベースのテーブル構造と Python のオブジェクト構造（クラスのインスタンス）は異なります。取得したデータをオブジェクトに変換したり、オブジェクトの状態をデータベースに保存したりするための煩雑な処理を自前で実装する必要があります。

SQLAlchemy は、これらの課題に対する包括的な解決策を提供します。

データベース抽象化レイヤー: 様々なデータベースシステム（PostgreSQL, MySQL, SQLite, Oracle, SQL Server など）に対応し、データベース間の差異を吸収します。接続プーリングやトランザクション管理なども効率的に行います。
SQL Expression Language: 生の SQL 文字列ではなく、Python オブジェクトとメソッドを使って SQL 文を構築できます。これにより、コードの可読性、保守性が向上し、SQL インジェクションのリスクを低減できます。
ORM (Object-Relational Mapper): データベースのテーブルを Python のクラスに、テーブルの行をクラスのインスタンスに、カラムをインスタンスの属性にマッピングします。これにより、オブジェクト指向のパラダイムでデータベースを操作できるようになり、データ操作コードが直感的で記述しやすくなります。

SQLAlchemy は、これらの機能を通じて、Python 開発におけるデータベース連携を大幅に効率化します。小規模なスクリプトから大規模な Web アプリケーションまで、幅広いプロジェクトでその真価を発揮します。

SQLAlchemy は大きく分けて以下の二つの主要なコンポーネントから構成されています。

SQLAlchemy Core: データベース抽象化レイヤーの中核をなす部分です。Python オブジェクトを使って SQL 文を構築・実行し、結果を扱うためのツールを提供します。ORM の基盤でもあります。生に近いレベルでデータベースを操作したい場合に適しています。
SQLAlchemy ORM: Core の上に構築され、Python オブジェクトとデータベースのリレーショナルデータをマッピングします。オブジェクト指向的なアプローチでデータを永続化、取得、操作できます。多くのアプリケーション開発では ORM が主に利用されます。

以降では、まず Core から解説し、その後 ORM について詳しく見ていきます。

2. Part 1: SQLAlchemy Core – データベース抽象化の基盤

SQLAlchemy Core は、生の SQL よりも安全かつプログラム的に SQL を構築・実行するための基盤を提供します。特定のデータベースに依存しない抽象化された方法でテーブル定義やクエリ操作を行うことができます。ORM を使う場合でも、その内部では Core の機能が利用されています。Core を理解することは、SQLAlchemy を深く理解する上で非常に重要です。

2.1 データベース接続 (Engines)

SQLAlchemy でデータベースに接続するには、create_engine() 関数を使って Engine オブジェクトを作成します。Engine は、データベースへの接続と、その接続を管理するためのプーリングやダイアレクト（特定のデータベースシステムに対応するためのアダプター）をカプセル化したものです。

create_engine() には、接続したいデータベースを指定する接続文字列 (Connection String) を渡します。接続文字列の形式はデータベースの種類によって異なりますが、一般的には以下のようになります。

dialect+driver://user:password@host:port/database

dialect: 使用するデータベースの種類（例: postgresql, mysql, sqlite, oracle, mssql など）
driver: 使用する DB-API ドライバー（例: psycopg2, mysqlconnector, sqlite3 など）。省略可能な場合が多いです。
user, password: データベースに接続するためのユーザー名とパスワード。
host, port: データベースサーバーのホスト名とポート番号。
database: 接続するデータベース名。

SQLite の場合は、ファイルパスを指定することもできます。

“`python
from sqlalchemy import create_engine

SQLite (インメモリデータベース)

engine_sqlite_memory = create_engine(“sqlite:///:memory:”)

SQLite (ファイル)

engine_sqlite_file = create_engine(“sqlite:///./my_database.db”)

PostgreSQL

engine_postgres = create_engine(“postgresql://user:password@host:port/database”)

MySQL (PyMySQL ドライバーを使用)

engine_mysql = create_engine(“mysql+pymysql://user:password@host:port/database”)

“`

create_engine() のオプションとして、echo=True を指定すると、SQLAlchemy が実行するすべての SQL 文が標準出力にログとして表示されるため、デバッグ時に非常に役立ちます。

python engine = create_engine("sqlite:///:memory:", echo=True)

接続プーリング

Engine はデフォルトで接続プーリングを有効にしています。これにより、接続の確立・切断のオーバーヘッドを減らし、効率的に接続を再利用できます。プーリングに関する設定は create_engine() の引数で調整可能です（例: pool_size, max_overflow, pool_timeout など）。

非同期 IO (AsyncIO) のサポート

SQLAlchemy 1.4 から非同期 IO がサポートされ、2.0 で完成しました。asyncio と連携してノンブロッキングなデータベース操作を行うことができます。非同期エンジンを作成するには create_async_engine() を使用します。

“`python

from sqlalchemy.ext.asyncio import create_async_engine

async_engine = create_async_engine(“postgresql+asyncpg://user:password@host:port/database”)

“`

非同期機能については後述の「Part 3: 実践的な使い方と応用」で詳しく触れます。

2.2 メタデータとテーブル定義 (MetaData, Table, Column)

データベースのスキーマ情報（テーブル名、カラム名、データ型、制約など）は、SQLAlchemy ではメタデータとして管理されます。Core では、MetaData オブジェクトを使ってこれらの情報を定義します。

MetaData オブジェクトは、既にあるデータベースからスキーマ情報をリフレクション（読み込み）することも、Python コード内でスキーマ情報を定義することもできます。ここでは、Python コードで定義する方法を見てみましょう。

まず、MetaData オブジェクトを作成します。

“`python
from sqlalchemy import MetaData

metadata = MetaData()
“`

次に、Table オブジェクトを使ってテーブルを定義します。Table コンストラクタは、テーブル名、そのテーブルが所属する MetaData オブジェクト、そして一つ以上の Column オブジェクトを引数に取ります。

Column オブジェクトは、カラム名、データ型、そして各種制約（主キー、外部キー、NOT NULL、デフォルト値など）を引数に取ります。

“`python
from sqlalchemy import Table, Column, Integer, String, ForeignKey

users_table = Table(
“users”, # テーブル名
metadata, # 所属する MetaData オブジェクト
Column(“id”, Integer, primary_key=True), # カラム: id (整数, 主キー)
Column(“name”, String(50), nullable=False), # カラム: name (文字列, NOT NULL)
Column(“age”, Integer), # カラム: age (整数)
Column(“email”, String(100), unique=True) # カラム: email (文字列, ユニーク制約)
)

addresses_table = Table(
“addresses”,
metadata,
Column(“id”, Integer, primary_key=True),
Column(“user_id”, Integer, ForeignKey(“users.id”), nullable=False), # 外部キー
Column(“email_address”, String(100), nullable=False)
)
“`

MetaData オブジェクトには、定義したすべてのテーブルが登録されます。metadata.tables で確認できます。

“`python
print(metadata.tables.keys())

出力例: dict_keys([‘users’, ‘addresses’])

“`

これらの定義をもとに、実際にデータベース上にテーブルを作成することができます。MetaData.create_all() メソッドは、指定した Engine を使って、メタデータに登録されているすべてのテーブルを作成します。

“`python

定義したテーブルをデータベースに作成 (もし存在しなければ)

engine = create_engine(“sqlite:///:memory:”, echo=True)

metadata.create_all(engine)

“`

2.3 SQL 表現言語 (SQL Expression Language)

SQLAlchemy Core の最も強力な機能の一つが、SQL 表現言語です。これにより、Python オブジェクトとメソッドの組み合わせによって、生の SQL 文字列を記述することなく SQL 文を構築できます。これは、可読性、保守性、そして何よりも SQL インジェクションからの保護に役立ちます。

SQL 表現言語は、select(), insert(), update(), delete() といった関数を使ってクエリ構造を作成します。これらの関数は、テーブルオブジェクトやカラムオブジェクトを引数に取り、実行可能なクエリ構造体を返します。

以降の例では、前のセクションで定義した users_table と addresses_table を使用します。

python from sqlalchemy import select, insert, update, delete, join from sqlalchemy import func # 集約関数用

SELECT 文

データを選択するには select() 関数を使用します。引数には取得したいカラムを指定します。カラムはテーブルオブジェクトの属性としてアクセスできます。

“`python

全てのユーザーを取得するSELECT文

s = select(users_table)
print(s)

例: SELECT users.id, users.name, users.age, users.email FROM users

“`

select() にテーブルオブジェクトを直接渡すと、そのテーブルのすべてのカラムが選択されます。

python s = select(users_table) # SELECT users.id, users.name, users.age, users.email FROM users

WHERE 句 (フィルタリング)

データをフィルタリングするには、クエリ構造体に対して .where() メソッドを使用します。条件は比較演算子を使って表現します。比較演算子は、カラムオブジェクトに対して Python の演算子 (==, !=, >, <, >=, <=) を使うことで生成できます。これらの演算子は SQLAlchemy によってオーバーロードされています。

“`python

IDが1のユーザーを取得

s = select(users_table).where(users_table.c.id == 1)

例: SELECT users.id, users.name, users.age, users.email FROM users WHERE users.id = :id_1

年齢が30歳以上のユーザーを取得

s = select(users_table).where(users_table.c.age >= 30)

名前が’Alice’で、かつ年齢が25歳以下のユーザーを取得 (AND)

s = select(users_table).where(
(users_table.c.name == ‘Alice’) & (users_table.c.age <= 25)
) # AND条件は & 演算子を使う

名前が’Bob’または年齢が40歳以上のユーザーを取得 (OR)

from sqlalchemy import or_ # OR条件は or_ 関数を使う

s = select(users_table).where(
or_(users_table.c.name == ‘Bob’, users_table.c.age >= 40)
)

名前が’A’で始まるユーザーを取得 (LIKE)

s = select(users_table).where(users_table.c.name.like(‘A%’))

メールアドレスがNULLでないユーザーを取得 (IS NOT NULL)

s = select(users_table).where(users_table.c.email.is_not(None))

複数のIDのいずれかに一致するユーザーを取得 (IN)

s = select(users_table).where(users_table.c.id.in_([1, 3, 5]))
“`

.where() メソッドは複数回チェーンすることができます。複数の .where() は AND 条件として結合されます。

“`python
s = select(users_table).where(users_table.c.name == ‘Alice’).where(users_table.c.age <= 25)

これは (users.name = ‘Alice’ AND users.age <= 25) と同じ

“`

ORDER BY 句 (順序付け)

結果を特定のカラムで並べ替えるには .order_by() メソッドを使用します。デフォルトは昇順（ASC）です。降順（DESC）にするには、カラムオブジェクトに対して .desc() メソッドを使います。

“`python

年齢の昇順でユーザーを取得

s = select(users_table).order_by(users_table.c.age)

年齢の降順、名前の昇順でユーザーを取得

s = select(users_table).order_by(users_table.c.age.desc(), users_table.c.name)
“`

LIMIT 句, OFFSET 句 (制限)

取得する行数を制限するには .limit() メソッドを、スキップする行数を指定するには .offset() メソッドを使用します。これらはページネーションなどに利用できます。

“`python

最初の10人のユーザーを取得

s = select(users_table).limit(10)

11番目から20番目までのユーザーを取得 (最初の10件をスキップして次の10件)

s = select(users_table).offset(10).limit(10)
“`

JOIN 句 (結合)

複数のテーブルを結合するには .join() メソッドを使用します。デフォルトでは内部結合（INNER JOIN）です。

“`python

ユーザーとそのアドレスを結合

users_table と addresses_table を user_id と id で結合

s = select(users_table, addresses_table).join(addresses_table, users_table.c.id == addresses_table.c.user_id)

例: SELECT users.id, users.name, users.age, users.email, addresses.id, addresses.user_id, addresses.email_address FROM users JOIN addresses ON users.id = addresses.user_id

結合するテーブルと結合条件を明確に指定することもできます

s = select(users_table, addresses_table).join(addresses_table, users_table.c.id == addresses_table.c.user_id)

外部結合 (LEFT OUTER JOIN など)

users テーブルにアドレスがあるかどうかに関わらず、全てのユーザーとそのアドレス（あれば）を取得

s = select(users_table, addresses_table).join(addresses_table, users_table.c.id == addresses_table.c.user_id, isouter=True) # isouter=True で LEFT OUTER JOIN
“`

集約関数とグループ化 (func, group_by)

COUNT, SUM, AVG, MAX, MIN といった集約関数は sqlalchemy.sql.func モジュールを通じて利用できます。

“`python

ユーザーの総数をカウント

s = select(func.count()).select_from(users_table)

平均年齢を計算

s = select(func.avg(users_table.c.age)).select_from(users_table)

各年齢層ごとのユーザー数をカウント (GROUP BY)

s = select(users_table.c.age, func.count()).group_by(users_table.c.age)

HAVING 句 (`group_by` の結果をフィルタリング)

年齢層ごとのユーザー数が5人以上のものを取得

s = select(users_table.c.age, func.count()).group_by(users_table.c.age).having(func.count() >= 5)
“`

サブクエリと CTE (Common Table Expressions)

複雑なクエリのためにサブクエリや CTE を使うことも可能です。SQL 表現言語はこれらの高度な概念もサポートしています。

“`python

サブクエリの例: 平均年齢より若いユーザーを取得

subquery = select(func.avg(users_table.c.age)).scalar_subquery()
s = select(users_table).where(users_table.c.age < subquery)

CTE の例 (with_cte() は SQLA 2.0+ で推奨されるスタイル)

user_counts というCTEを定義し、各年齢層のユーザー数をカウント

user_counts_cte = select(users_table.c.age, func.count().label(“count”)).group_by(users_table.c.age).cte(“user_counts”)

CTE を使って、ユーザー数が5人以上の年齢層を取得

s = select(user_counts_cte).where(user_counts_cte.c.count >= 5)

WITH clause (従来の with_xxx() メソッド)

user_counts_cte = select(users_table.c.age, func.count().label(“count”)).group_by(users_table.c.age).alias(“user_counts_alias”)

s = select(user_counts_cte).where(user_counts_cte.c.count >= 5) # Aliasを使う場合は c.count でアクセス

“`

INSERT, UPDATE, DELETE 文

データの挿入、更新、削除も SQL 表現言語で行います。

INSERT:

“`python

1件のレコードを挿入

ins = insert(users_table).values(name=’Alice’, age=30, email=’[email protected]’)
print(ins)

例: INSERT INTO users (name, age, email) VALUES (:name, :age, :email)

複数件のレコードを挿入 (リスト形式でデータを渡す)

ins = insert(users_table).values([
{‘name’: ‘Bob’, ‘age’: 25, ‘email’: ‘[email protected]’},
{‘name’: ‘Charlie’, ‘age’: 35, ‘email’: ‘[email protected]’}
])
“`

UPDATE:

“`python

IDが1のユーザーの名前を更新

upd = update(users_table).where(users_table.c.id == 1).values(name=’Alicia’)
print(upd)

例: UPDATE users SET name=:name WHERE users.id = :id_1

全ユーザーの年齢を+1更新

upd = update(users_table).values(age=users_table.c.age + 1)
“`

DELETE:

“`python

IDが2のユーザーを削除

d = delete(users_table).where(users_table.c.id == 2)
print(d)

例: DELETE FROM users WHERE users.id = :id_1

“`

2.4 実行 (Executing Statements)

構築した SQL 表現言語のステートメントを実行するには、Engine から接続を取得し、その接続オブジェクトの execute() メソッドを使用します。SQLAlchemy 2.0 以降では、execute() は接続オブジェクトの最も一般的なメソッドとなりました。

“`python

engine = create_engine(“sqlite:///:memory:”, echo=True)

metadata.create_all(engine) # テーブル作成

with engine.connect() as connection:
# INSERT 実行
ins = insert(users_table).values(name=’Alice’, age=30, email=’[email protected]’)
result = connection.execute(ins)
print(f”Inserted user with ID: {result.lastrowid}”) # 挿入されたレコードの主キーを取得

# 複数件 INSERT
connection.execute(insert(users_table).values([
    {'name': 'Bob', 'age': 25, 'email': '[email protected]'},
    {'name': 'Charlie', 'age': 35, 'email': '[email protected]'},
    {'name': 'Alice', 'age': 28, 'email': '[email protected]'}, # 同じ名前の別ユーザー
]))

# SELECT 実行
s = select(users_table).where(users_table.c.name == 'Alice')
result = connection.execute(s)

# 結果セットの取得
# fetchone(): 最初の1行を取得
# fetchall(): 全ての行をリストで取得
# イテレーション: 行ごとに順に取得

# 全て取得
# for row in result:
#     print(row) # 行はタプルまたは RowProxy オブジェクトとして取得
#     print(f"ID: {row.id}, Name: {row.name}") # カラム名でアクセス可能

# あるいは fetchall() で一度に取得
rows = result.fetchall()
print("All users named Alice:")
for row in rows:
    print(row)

# LIMIT + OFFSET 付き SELECT
s_paginated = select(users_table).order_by(users_table.c.id).limit(2).offset(1)
print("\nPaginated results (skip 1, limit 2):")
for row in connection.execute(s_paginated):
     print(row)

# UPDATE 実行
upd = update(users_table).where(users_table.c.name == 'Bob').values(age=26)
result = connection.execute(upd)
print(f"\nUpdated {result.rowcount} rows.") # 更新された行数

# DELETE 実行
d = delete(users_table).where(users_table.c.name == 'Charlie')
result = connection.execute(d)
print(f"Deleted {result.rowcount} rows.") # 削除された行数

# SELECT 再実行して確認
print("\nRemaining users:")
for row in connection.execute(select(users_table)):
    print(row)

# トランザクションの確定はデフォルトで行われる (autocommit=True相当)
# 明示的なトランザクション管理は後述

“`

with engine.connect() as connection: のように with ステートメントを使うことで、接続が自動的にクローズされることが保証されます。

execute() メソッドは Result オブジェクトを返します。Result オブジェクトから fetchone(), fetchall() メソッドで結果を取得したり、オブジェクトをイテレートしたりできます。行は RowProxy オブジェクトとして取得され、カラム名またはインデックスで値にアクセスできます（例: row['name'] や row.name）。

2.5 トランザクション管理

Connection オブジェクトは、デフォルトでは DML (INSERT, UPDATE, DELETE) ステートメントを実行するたびに自動的にコミットを行うモード（autocommit）ではありません。しかし、with engine.connect() as connection: のように with を使うと、ブロックの終了時にエラーが発生しなければ自動的にコミットが行われ、エラーが発生した場合はロールバックが行われます。これは非常に便利なデフォルト挙動です。

より細粒度なトランザクション管理が必要な場合は、connection.begin() を使って明示的にトランザクションを開始し、commit() や rollback() を呼び出します。

“`python

engine = create_engine(“sqlite:///:memory:”, echo=True)

metadata.create_all(engine) # テーブル作成

with engine.connect() as connection:
with connection.begin(): # 明示的にトランザクションを開始
# トランザクション内の操作
connection.execute(
insert(users_table).values(name=’David’, age=40, email=’[email protected]’)
)
# ここで何らかのエラーが発生したとする
# raise ValueError(“Something went wrong”)

# ここまで到達すればコミット、エラーならロールバック
print("\nUsers after explicit transaction:")
for row in connection.execute(select(users_table)):
    print(row)

あるいは try…except…finally で手動管理

connection = engine.connect()

try:

# トランザクション開始

trans = connection.begin()

connection.execute(

insert(users_table).values(name=’Eve’, age=22, email=’[email protected]’)

)

# 何らかの条件に基づいてロールバック

# if some_condition:

# trans.rollback()

# else:

# trans.commit()

trans.commit() # コミット

except Exception as e:

trans.rollback() # ロールバック

print(f”Transaction failed: {e}”)

finally:

connection.close() # 接続を閉じる

“`

2.6 Core のメリット・デメリット

メリット:

低レベルな制御: ORM よりもデータベースに近いレベルで操作できます。特定の SQL 機能（ウィンドウ関数、複雑な CTE など）を直接利用しやすいです。
パフォーマンス: ORM のオーバーヘッドがないため、単純なクエリや一括処理では ORM よりも高速になる可能性があります。
SQL の柔軟性: SQL 表現言語は生の SQL のほとんどの機能をサポートしており、複雑なクエリもプログラム的に構築できます。
DB スキーマへの依存が少ない: ORM のようにクラスとテーブルを厳密にマッピングする必要がないため、DB スキーマが頻繁に変更される場合や、既存の複雑なスキーマを扱う場合に ORM より柔軟に対応できることがあります。

デメリット:

記述量が多い: ORM に比べて、特にデータ取得後のオブジェクトへのマッピングや関連データの取得などで記述量が多くなりがちです。
オブジェクト指向の利便性が低い: データをオブジェクトとして扱うための自動的なマッピングや、リレーションシップによる関連データの自動取得といった ORM の恩恵を受けられません。
開発速度: ORM と比較すると、アプリケーションレベルでの開発速度は遅くなる傾向があります。

Core は、データベースの低レベルな機能を利用したい場合、パフォーマンスが特に重要で ORM のオーバーヘッドを避けたい場合、あるいは ORM の上にカスタムなデータマッピング層を構築したい場合などに適しています。

3. Part 2: SQLAlchemy ORM – オブジェクト指向によるDB操作

SQLAlchemy ORM は、リレーショナルデータベースのテーブルと Python オブジェクトの間でマッピングを行うレイヤーです。これにより、データベース操作をオブジェクト指向的なアプローチで行うことができます。ORM を使うことで、SQL を直接書く代わりに Python オブジェクトのメソッドや属性を操作するだけで、データの永続化、取得、更新、削除が可能になります。

多くの Web アプリケーションやビジネスロジックでは、データをオブジェクトとして扱う方が自然です。ORM は、このオブジェクト指向の世界とリレーショナルデータベースの世界の間のギャップを埋めてくれます。

3.1 ORM とは何か？なぜ ORM を使うのか？

ORM (Object-Relational Mapper) は、オブジェクト指向プログラミング言語でリレーショナルデータベースを操作するための技術です。データベースのテーブルを行ごとに Python のクラスのインスタンスにマッピングし、各カラムをインスタンスの属性にマッピングします。

ORM を使う主な理由は以下の通りです。

開発効率の向上: SQL を直接書く手間が省け、Python コードで完結したデータベース操作が可能になります。
保守性の向上: オブジェクト指向的なコードは構造化されており、保守が容易になります。SQL インジェクションのリスクも自動的に低減されます。
可読性の向上: データの取得や操作が、より直感的で分かりやすいオブジェクト操作として記述できます。
データベースの抽象化: Core と同様に、特定のデータベースシステムに依存しないコードを記述できます。

SQLAlchemy ORM は、強力かつ柔軟なマッピング機能とクエリインターフェースを提供します。

3.2 Core との関係性

SQLAlchemy ORM は、その内部で SQLAlchemy Core を利用しています。ORM で定義したマッピングやクエリは、最終的には Core の SQL 表現言語のステートメントに変換され、Core の Engine や Connection を通じてデータベースに実行されます。

ORM は Core の上に高レベルな抽象化を提供しており、開発者は通常、ORM の機能のみに焦点を当てることができます。しかし、Core の機能（例えば複雑な SQL 関数や特定のクエリ構造）が必要になった場合は、ORM のコンテキスト内で Core の機能を利用することも可能です。

3.3 マッピング (Mapping)

ORM を使うには、まずデータベースのテーブル構造に対応する Python クラスを定義し、そのクラスとテーブルをマッピングする必要があります。SQLAlchemy ORM にはいくつかマッピングの方法がありますが、最も一般的で推奨されているのは Declarative Mapping です。

Declarative Mapping では、Python クラスの中でテーブル構造とマッピング情報を同時に定義します。まず、マッピングのベースとなるクラスを定義します。

“`python
from sqlalchemy.orm import declarative_base

Base = declarative_base()
“`

Base クラスは、後で定義するすべての ORM マッピングクラスの親クラスとなります。この Base クラスは、内部的に Core の MetaData オブジェクトを管理しています。

次に、データベースのテーブルに対応する Python クラスを定義します。このクラスは Base を継承し、クラス属性としてテーブルのカラムを定義します。カラムは sqlalchemy.orm.Mapped 型アノテーションと sqlalchemy.orm.mapped_column() 関数を使って定義します。

SQLAlchemy 2.0 以降では、型ヒントを活用した Declarative Mapping が推奨されています。

“`python
from sqlalchemy import Integer, String, ForeignKey
from sqlalchemy.orm import Mapped, mapped_column, relationship # relationship は後述

Base = declarative_base() # 既に定義済み

class User(Base):
tablename = “users” # マッピングするテーブル名

# カラム定義
id: Mapped[int] = mapped_column(primary_key=True)
name: Mapped[str] = mapped_column(String(50), nullable=False)
age: Mapped[int | None] # Optional[int] と同じ意味、null を許可
email: Mapped[str | None] = mapped_column(String(100), unique=True)

# リレーションシップ定義 (後述)
# addresses: Mapped[list["Address"]] = relationship(back_populates="user")

# オブジェクトの文字列表現 (デバッグ用)
def __repr__(self):
    return f"<User(id={self.id}, name='{self.name}', age={self.age}, email='{self.email}')>"

class Address(Base):
tablename = “addresses”

id: Mapped[int] = mapped_column(primary_key=True)
user_id: Mapped[int] = mapped_column(ForeignKey("users.id")) # 外部キー
email_address: Mapped[str] = mapped_column(String(100), nullable=False)

# リレーションシップ定義 (後述)
# user: Mapped["User"] = relationship(back_populates="addresses")

def __repr__(self):
    return f"<Address(id={self.id}, email_address='{self.email_address}')>"

“`

ここで定義した User クラスと Address クラスは、それぞれ users テーブルと addresses テーブルに対応付けられます。Mapped[type] は型ヒントであり、mapped_column() が実際のマッピング情報を設定します。mapped_column() は Core の Column とほぼ同じ引数を取ります。

Base クラスは、このマッピング情報を Core の MetaData オブジェクトとして内部に保持しています。Base.metadata にアクセスすることで、このメタデータオブジェクトを取得できます。

“`python

テーブル作成は Core と同じ方法で行う

Base.metadata.create_all(engine)

“`

Dataclass との連携 (`MappedAsDataclass`)

Python の Dataclass と連携させることで、マッピングクラスをより簡潔に定義できます。

“`python
from sqlalchemy.orm import MappedAsDataclass, Mapped, mapped_column
from sqlalchemy import Integer, String

class User(MappedAsDataclass, Base): # MappedAsDataclass を継承
tablename = “users”

id: Mapped[int] = mapped_column(primary_key=True, init=False, default=None) # init=False, default=None で自動生成される __init__ に含めない
name: Mapped[str] = mapped_column(String(50))
age: Mapped[int | None] = mapped_column(default=None)
email: Mapped[str | None] = mapped_column(String(100), unique=True, default=None)

# Dataclass として扱えるようになる
# User(name="Alice", age=30) のようにインスタンスを作成できる

“`

MappedAsDataclass を使うと、__init__ メソッドや __repr__ メソッドなどが自動生成され、Dataclass の機能を利用できます。ただし、主キーのようにデータベース側で自動生成されるカラムは init=False, default=None を指定することが推奨されます。

3.4 セッション (Session)

SQLAlchemy ORM におけるデータベース操作の中心となるのが Session です。Session は、ORM オブジェクト（マッピングクラスのインスタンス）とデータベースの間での対話、つまりオブジェクトの読み込み、追加、変更、削除、そしてトランザクション管理を行います。

Session は Core の Connection のラッパーのようなもので、トランザクションの境界を表します。一つの Session は通常、一つのトランザクションに対応します。

Session を作成するには、まず sessionmaker 関数を使って Session を生成するためのファクトリを作成します。

“`python
from sqlalchemy.orm import sessionmaker

engine = create_engine(“sqlite:///:memory:”, echo=True)

Session ファクトリを作成

bind に Engine オブジェクトを渡す

Session = sessionmaker(bind=engine)
“`

sessionmaker は、セッションを生成するための設定（バインドするエンジン、autocommit 設定、autoflush 設定など）を保持します。Session クラス（ここでは Session という名前の変数に代入）のインスタンスを作成することで、個々のセッションを取得します。

“`python

Session のインスタンスを作成

session = Session()
“`

セッションのライフサイクル

セッションのライフサイクルは通常以下のようになります。

Session() を呼び出してセッションを作成する。
セッションを使ってオブジェクトを操作する（追加、クエリなど）。
session.commit() を呼び出して変更をデータベースに永続化する。または session.rollback() で変更を取り消す。
session.close() を呼び出してセッションを閉じる（接続プールに接続を戻す）。

with ステートメントを使うと、セッションのクローズとトランザクション管理（成功時のコミット、失敗時のロールバック）が自動的に行われるため便利です。

“`python

Session = sessionmaker(bind=engine) # 既に定義済み

with Session() as session: # with を使うと session.close() が自動で行われる
# セッションを使った操作
# …
session.commit() # with ブロックの終了時にエラーがなければコミット
# エラーが発生した場合は自動でロールバック
“`

3.5 オブジェクトの永続化 (Persistence)

Session を使うことで、Python オブジェクトをデータベースに保存したり、データベースから読み込んだデータをオブジェクトとして扱ったりできます。

オブジェクトの追加 (INSERT):

マッピングクラスのインスタンスを作成し、session.add() または session.add_all() を使ってセッションに追加します。追加されたオブジェクトは「Pending」状態になります。

“`python

Session = sessionmaker(bind=engine) # 既に定義済み

with Session() as session:
# User オブジェクトを作成
new_user = User(name=’David’, age=40, email=’[email protected]’)

# オブジェクトをセッションに追加
session.add(new_user)

# この時点ではまだデータベースには保存されていない (Pending 状態)
print(new_user.id) # ID はまだ None (データベースで生成される場合)

# commit() を呼び出すことでデータベースに保存される
session.commit()

# commit() 後、オブジェクトはデータベースに同期され、IDなどが設定される
print(new_user.id) # ID が設定されているはず
print(f"Added user: {new_user}")

“`

session.add_all() を使うと、複数のオブジェクトを一度に追加できます。

python with Session() as session: users_to_add = [ User(name='Eve', age=22, email='[email protected]'), User(name='Frank', age=50, email='[email protected]') ] session.add_all(users_to_add) session.commit()

オブジェクトの変更 (UPDATE):

セッションによって読み込まれたオブジェクトの属性値を変更すると、その変更はセッションによって追跡されます。commit() 時に、変更されたオブジェクトに対応するデータベースレコードが更新されます。

“`python

Session = sessionmaker(bind=engine) # 既に定義済み

with Session() as session:
# ID が 1 のユーザーを取得
user_to_update = session.get(User, 1) # ORM の get() メソッド (主キーで取得)

if user_to_update:
    # オブジェクトの属性を変更
    user_to_update.age = 31
    user_to_update.email = '[email protected]'

    # 変更はセッションによって自動追跡される
    # commit() 時に UPDATE 文が発行される
    session.commit()
    print(f"Updated user: {user_to_update}")
else:
    print("User with ID 1 not found.")

“`

オブジェクトの削除 (DELETE):

セッションによって読み込まれたオブジェクトを session.delete() に渡すと、そのオブジェクトは「Deleted」状態になります。commit() 時に、対応するデータベースレコードが削除されます。

“`python

Session = sessionmaker(bind=engine) # 既に定義済み

with Session() as session:
# ID が 2 のユーザーを取得
user_to_delete = session.get(User, 2)

if user_to_delete:
    # オブジェクトをセッションから削除対象としてマーク
    session.delete(user_to_delete)

    # commit() 時に DELETE 文が発行される
    session.commit()
    print(f"Deleted user: {user_to_delete}")
else:
    print("User with ID 2 not found.")

“`

ロールバック (Rollback)

トランザクション中にエラーが発生した場合や、変更を取り消したい場合は、session.rollback() を呼び出します。これにより、現在のトランザクションで行われたすべての操作が取り消され、セッションの状態がロールバック前の状態に戻ります。

with Session() as session: を使う場合は、ブロック内で例外が発生すると自動的にロールバックが行われます。

“`python

Session = sessionmaker(bind=engine) # 既に定義済み

try:
with Session() as session:
session.add(User(name=’Grace’, age=33, email=’[email protected]’))
# 強制的にエラーを発生させる
# assert False, “Simulating an error”
print(“This line will not be reached if error occurs”)
session.commit() # コミットは実行されない
except Exception as e:
print(f”An error occurred: {e}. Changes will be rolled back.”)
# with ブロックの外では session は閉じられているが、with 内部で自動ロールバックされている

ロールバックされたことを確認するためにデータを再度取得してみる

with Session() as session:
grace = session.execute(select(User).where(User.name == ‘Grace’)).scalar_one_or_none()
print(f”Grace user found after rollback attempt: {grace}”) # None が出力されるはず
“`

3.6 クエリ (Querying)

SQLAlchemy ORM でデータを取得するには、セッションからクエリを発行します。SQLAlchemy 2.0 以降では、Core の select() 関数を使ってクエリを構築し、それをセッションの execute() メソッドで実行するスタイルが推奨されています。これにより、Core と ORM のクエリ構築方法が統一され、よりシンプルになりました。

“`python
from sqlalchemy import select

Session = sessionmaker(bind=engine) # 既に定義済み

with Session() as session:
# 全てのユーザーを取得するクエリを構築
stmt = select(User)

# クエリを実行
result = session.execute(stmt)

# 結果セットからオブジェクトを取得
# result は Result オブジェクトを返す
# ORM では Row オブジェクトのタプルの中にマッピングされたオブジェクトが含まれる
# scalar_one(), scalar_one_or_none(), scalars().all(), scalars().first(), scalars().fetchone() などを使用

# 全ての User オブジェクトを取得
users = result.scalars().all()
print("\nAll users:")
for user in users:
    print(user)

# 最初の1件を取得
first_user = session.execute(select(User)).scalars().first()
print(f"\nFirst user: {first_user}")

# 条件付きで取得 (WHERE)
stmt = select(User).where(User.name == 'Alice')
alice_users = session.execute(stmt).scalars().all()
print("\nUsers named Alice:")
for user in alice_users:
    print(user)

# 主キーで1件取得 (便利なメソッド)
user_by_id = session.get(User, 1)
print(f"\nUser with ID 1: {user_by_id}")

# ORDER BY
stmt = select(User).order_by(User.age.desc())
users_ordered = session.execute(stmt).scalars().all()
print("\nUsers ordered by age (DESC):")
for user in users_ordered:
    print(user)

# LIMIT, OFFSET
stmt = select(User).order_by(User.id).offset(1).limit(2)
users_paginated = session.execute(stmt).scalars().all()
print("\nPaginated users (skip 1, limit 2):")
for user in users_paginated:
    print(user)

# COUNT (集約関数)
stmt = select(func.count()).select_from(User) # ORM クラスを select_from に渡す
user_count = session.execute(stmt).scalar_one() # 集約関数の結果は scalar で取得
print(f"\nTotal user count: {user_count}")

# JOIN
# User と Address を結合して、ユーザー名とメールアドレスを取得
stmt = select(User.name, Address.email_address).join(Address) # ORM クラスを指定して join
results = session.execute(stmt).all() # Row オブジェクトのタプルを取得
print("\nUser names and emails from join:")
for row in results:
    print(f"Name: {row.name}, Email: {row.email_address}") # カラム名でアクセス

# DISTINCT
stmt = select(User.age).distinct()
distinct_ages = session.execute(stmt).scalars().all()
print(f"\nDistinct ages: {distinct_ages}")

# Group By
stmt = select(User.age, func.count()).group_by(User.age)
age_counts = session.execute(stmt).all()
print("\nUser count by age:")
for age, count in age_counts:
    print(f"Age {age}: {count}")

“`

SQLAlchemy 2.0 以前の session.query() スタイルについて:

SQLAlchemy 1.x では session.query(User).filter(...) のようなスタイルが主流でしたが、2.0 以降は Core と共通の select(User).where(...) スタイルが推奨されています。新しいスタイルの方がより SQL に近く、Core との連携もスムーズです。レガシーコードや既存のドキュメントでは古いスタイルを見かけることもありますが、これから新たに開発する場合は 2.0 スタイルを使うべきです。

3.7 リレーションシップ (Relationships)

リレーショナルデータベースの大きな特徴は、テーブル間の関連性（リレーションシップ）を定義できることです。SQLAlchemy ORM では、relationship() 関数を使って、マッピングされたクラス間にリレーションシップを定義し、関連するオブジェクトを簡単に操作できるようにします。

例えば、一人のユーザーが複数のアドレスを持つ場合（一対多のリレーションシップ）、User クラスにそのユーザーに紐づく Address オブジェクトのリスト属性を追加できます。

“`python

User および Address クラスの定義を更新

from sqlalchemy import Integer, String, ForeignKey
from sqlalchemy.orm import Mapped, mapped_column, relationship, declarative_base

Base = declarative_base()

class User(Base):
tablename = “users”
id: Mapped[int] = mapped_column(primary_key=True)
name: Mapped[str] = mapped_column(String(50), nullable=False)
age: Mapped[int | None]
email: Mapped[str | None] = mapped_column(String(100), unique=True)

# User と Address のリレーションシップ (一対多)
# User オブジェクトから addresses 属性を通じて、関連する Address オブジェクトのリストにアクセスできる
addresses: Mapped[list["Address"]] = relationship(back_populates="user")

def __repr__(self):
    return f"<User(id={self.id}, name='{self.name}')>"

class Address(Base):
tablename = “addresses”
id: Mapped[int] = mapped_column(primary_key=True)
user_id: Mapped[int] = mapped_column(ForeignKey(“users.id”))
email_address: Mapped[str] = mapped_column(String(100), nullable=False)

# Address と User のリレーションシップ (多対一)
# Address オブジェクトから user 属性を通じて、関連する User オブジェクトにアクセスできる
user: Mapped["User"] = relationship(back_populates="addresses")

def __repr__(self):
    return f"<Address(id={self.id}, email_address='{self.email_address}')>"

Base.metadata.create_all(engine) # テーブル再作成または更新が必要

“`

relationship() 関数は、関連するクラス（"Address" や "User"）、そして双方向のリレーションシップを確立するための back_populates 引数をよく使います。back_populates="user" は、Address クラスの user 属性が、User クラスの addresses 属性に対応することを示します。

リレーションシップを使った操作:

リレーションシップを定義すると、オブジェクト間の関連をオブジェクト指向的に操作できます。

“`python

Session = sessionmaker(bind=engine) # 既に定義済み

事前にユーザーとアドレスを追加しておく

with Session() as session:

u1 = User(name=’Alice’, age=30)

a1 = Address(email_address=’[email protected]’, user=u1) # Address 作成時に user に User オブジェクトを渡す

a2 = Address(email_address=’[email protected]’, user=u1)

session.add_all([u1, a1, a2])

u2 = User(name=’Bob’, age=25)

a3 = Address(email_address=’[email protected]’, user=u2)

session.add_all([u2, a3])

session.commit()

with Session() as session:
# User オブジェクトを取得し、関連する addresses にアクセス
user = session.execute(select(User).where(User.name == ‘Alice’)).scalars().first()

if user:
    print(f"User: {user.name}")
    print("Addresses:")
    for address in user.addresses: # relationship を通じて関連アドレスにアクセス
        print(f"- {address.email_address}")

print("-" * 20)

# Address オブジェクトを取得し、関連する user にアクセス
address = session.execute(select(Address).where(Address.email_address == '[email protected]')).scalars().first()

if address:
    print(f"Address: {address.email_address}")
    print(f"Owned by user: {address.user.name}") # relationship を通じて関連ユーザーにアクセス

print("-" * 20)

# 新しいアドレスを追加 (User オブジェクトに紐づける)
user_to_add_address = session.get(User, 1) # Alice
if user_to_add_address:
    new_address = Address(email_address='[email protected]')
    user_to_add_address.addresses.append(new_address) # リストに追加するだけでリレーションシップが設定される
    session.commit()
    print(f"New address added for {user_to_add_address.name}: {new_address.email_address}")

print("-" * 20)

# Address を削除すると、リレーションシップも解除される (User オブジェクトの addresses リストからも削除される)
address_to_delete = session.execute(select(Address).where(Address.email_address == '[email protected]')).scalars().first()
if address_to_delete:
    session.delete(address_to_delete)
    session.commit()
    print(f"Address deleted: {address_to_delete.email_address}")

    # 再度ユーザーを取得してアドレスを確認
    user_after_delete = session.get(User, 1)
    print(f"{user_after_delete.name}'s addresses after deletion:")
    for address in user_after_delete.addresses:
        print(f"- {address.email_address}")

“`

リレーションシップの種類

一対多 (One-to-Many): User に複数の Address が紐づく例のように、一つの親オブジェクトに複数の子オブジェクトが関連する場合。親クラスに子クラスのリスト属性を定義し、子クラスに親クラスの単一オブジェクト属性を定義します。子クラス側には親への外部キーが必要です。
多対一 (Many-to-One): Address が一つの User に紐づく例のように、複数の子オブジェクトが一つの親オブジェクトに関連する場合。子クラスに親クラスの単一オブジェクト属性を定義し、子クラスには親への外部キーが必要です。親クラスに子クラスのリスト属性を定義すると一対多と多対一が双方向になります。
一対一 (One-to-One): 例えば User と UserProfile のように、各々が互いに一つだけ関連する場合。両方のクラスに単一オブジェクト属性を定義し、一方または両方にユニーク制約付きの外部キーが必要です。relationship(uselist=False) を使います。
多対多 (Many-to-Many): 例えば Student と Course のように、一人の学生が複数のコースを履修し、一つのコースに複数の学生が参加する場合。通常、中間テーブル（または結合テーブル、アソシエーションテーブル）を使って実現します。relationship() を定義する際に、中間テーブルを secondary 引数で指定します。

多対多の例（Student – Course 間に関係テーブル enrollment を介する）：

“`python

from sqlalchemy import Table, Column, Integer, ForeignKey

from sqlalchemy.orm import relationship, declarative_base

from sqlalchemy.orm import Mapped, mapped_column

Base = declarative_base()

enrollment_table = Table(

“enrollments”,

Base.metadata, # Base.metadata に登録

Column(“student_id”, Integer, ForeignKey(“students.id”), primary_key=True),

Column(“course_id”, Integer, ForeignKey(“courses.id”), primary_key=True)

)

class Student(Base):

tablename = “students”

id: Mapped[int] = mapped_column(primary_key=True)

name: Mapped[str]

# 多対多リレーションシップ

courses: Mapped[list[“Course”]] = relationship(secondary=enrollment_table, back_populates=”students”)

class Course(Base):

tablename = “courses”

id: Mapped[int] = mapped_column(primary_key=True)

title: Mapped[str]

# 多対多リレーションシップ

students: Mapped[list[“Student”]] = relationship(secondary=enrollment_table, back_populates=”courses”)

“`

3.8 ORM のメリット・デメリット

メリット:

開発速度: オブジェクト指向での直感的な操作により、開発速度が向上します。CRUD 操作が非常に簡潔に記述できます。
保守性: アプリケーションコードとデータベース操作が密結合しにくく、コードの構造が明確になります。
リレーションシップ管理: オブジェクト間の関連を容易に扱えます。
N+1 問題の対策: ロード戦略を適切に設定することで、一般的なパフォーマンス問題を回避できます。

デメリット:

学習コスト: Core と ORM の両方の概念を理解する必要があります。特にセッションの扱い、リレーションシップ、ロード戦略は習得に時間がかかる場合があります。
抽象化の限界: 非常に複雑な SQL クエリ（特定のデータベース固有の関数、高度なウィンドウ関数など）を ORM だけで表現するのが難しい場合があります。その場合は Core の機能と組み合わせるか、生 SQL を使う必要が出てきます。
パフォーマンスのオーバーヘッド: オブジェクトマッピングなどの処理により、Core を直接使う場合と比較して若干のオーバーヘッドが発生します。

多くのアプリケーション開発では ORM が主要なツールとなります。開発効率と保守性の高さがその理由です。

4. Part 3: 実践的な使い方と応用

SQLAlchemy は基本的な DB 操作だけでなく、実際のアプリケーション開発で役立つ多くの機能を提供しています。

4.1 マイグレーションツール (Alembic)

データベースのスキーマはアプリケーションの進化とともに変更されることがよくあります（例: 新しいテーブルの追加、カラムの追加・削除・変更）。これらのスキーマ変更履歴を管理し、開発環境や本番環境に安全に適用するためのツールがマイグレーションツールです。SQLAlchemy の公式なマイグレーションツールとして Alembic があります。

Alembic は、データベースの現在のスキーマと SQLAlchemy モデルの定義を比較し、差分を検知してスキーマ変更のための Python スクリプト（リビジョンファイル）を自動生成する機能（autogenerate）を持っています。開発者はこのスクリプトをレビュー・編集し、アップグレード（スキーマ変更の適用）やダウングレード（変更の取り消し）を行います。

Alembic の基本的な使い方は以下の通りです。

インストール: pip install alembic
初期化: プロジェクトのルートディレクトリで alembic init alembic を実行します。これにより alembic.ini 設定ファイルと alembic ディレクトリが作成されます。
環境設定: alembic/env.py ファイルを編集し、SQLAlchemy の Engine や Base (ORM を使う場合) を Alembic からアクセスできるように設定します。特に target_metadata に ORM の Base.metadata を設定することが重要です。
“`python
# alembic/env.py の一部
from logging.config import fileConfig

from sqlalchemy import engine_from_config
from sqlalchemy import pool

from alembic import context

プロジェクトのモデル定義をインポート

from myapp.models import Base # 例: myapp/models.py に Base とモデルクラスがある場合

alembic.ini から設定を取得

config = context.config

logging 設定

if config.config_file_name is not None:
fileConfig(config.config_file_name)

target_metadata を設定

ORM を使う場合、Base.metadata をここに設定する

target_metadata = Base.metadata # Alembic がこのメタデータを見てスキーマを比較する

… (rest of env.py)

def run_migrations_online():
“””Run migrations in ‘online’ mode. …”””
connectable = engine_from_config(
config.get_section(config.config_ini_section),
prefix=”sqlalchemy.”,
poolclass=pool.NullPool,
)
```
with connectable.connect() as connection:
    context.configure(
        connection=connection,
        target_metadata=target_metadata,
        # ...
    )

    with context.begin_transaction():
        context.run_migrations()
```
… (rest of env.py)

また、`alembic.ini` で SQLAlchemy の接続文字列を設定します。ini

alembic.ini の一部

[sqlalchemy]

url = driver://user:password@host:port/database

url = sqlite:///./my_database.db # 実際のデータベース接続文字列に置き換える
“`
リビジョンファイルの生成: モデルに変更を加えた後、alembic revision -m "create user and address tables" --autogenerate のようにコマンドを実行します。これにより、スキーマの差分に基づいたマイグレーションスクリプトが alembic/versions ディレクトリに生成されます。
リビジョンファイルの確認・編集: 生成された Python スクリプトを開き、意図通りのスキーマ変更（upgrade() 関数と downgrade() 関数）が記述されているか確認します。必要に応じて手動で編集します。
マイグレーションの適用: alembic upgrade head コマンドを実行すると、保留中のすべてのマイグレーションがデータベースに適用されます。特定のバージョンにアップグレード/ダウングレードすることも可能です。

Alembic は、開発チームが複数いる場合や、アプリケーションのデプロイメントパイプラインにおいて、データベーススキーマを確実に管理するために不可欠なツールです。

4.2 エラーハンドリング (Error Handling)

データベース操作中に発生する可能性のあるエラーには、接続エラー、SQL 構文エラー、データ整合性エラーなどがあります。SQLAlchemy は、基盤となる DB-API ドライバーから送られる例外を捕捉し、sqlalchemy.exc モジュールで定義された独自の例外クラスにラップして送出します。

一般的な例外の例:

sqlalchemy.exc.OperationalError: データベースへの接続失敗、SQL 構文エラーなど、データベース操作に関する一般的なエラー。
sqlalchemy.exc.IntegrityError: 主キー違反、ユニーク制約違反、NOT NULL 制約違反、外部キー制約違反など、データの整合性に関するエラー。
sqlalchemy.exc.DatabaseError: より一般的なデータベース関連エラーの基底クラス。

これらの例外を適切にキャッチして処理することで、堅牢なアプリケーションを構築できます。特に IntegrityError は、ユーザーからの入力がデータベースの制約に違反した場合などに発生するため、アプリケーションレベルでの検証や適切なエラーメッセージの表示が必要です。

“`python
from sqlalchemy.exc import IntegrityError

Session = sessionmaker(bind=engine) # 既に定義済み

try:
with Session() as session:
# email がユニーク制約に違反するユーザーを追加しようとする
duplicate_user = User(name=’Alice’, age=30, email=’[email protected]’)
session.add(duplicate_user)
session.commit() # ここで IntegrityError が発生する可能性
except IntegrityError as e:
print(f”Integrity Error occurred: {e}”)
# session が with で管理されていれば自動でロールバックされる
# 手動管理の場合は session.rollback() が必要
# 通常、このエラーはユーザーに何らかのフィードバックを返す必要がある
except Exception as e:
print(f”An unexpected error occurred: {e}”)

トランザクションがロールバックされたことを確認

with Session() as session:
user = session.execute(select(User).where(User.email == ‘[email protected]’)).scalars().one_or_none()
print(f”User with duplicate email found after error: {user}”) # None が出力されるはず
“`

4.3 パフォーマンスチューニング

大規模なアプリケーションや高負荷なシステムでは、データベース操作のパフォーマンスがボトルネックになることがよくあります。SQLAlchemy を使った開発でも、パフォーマンスを意識することが重要です。

N+1 問題の対策: 前述の通り、Lazy Loading による N+1 問題は一般的なパフォーマンスキラーです。関連データを取得する際は、selectinload や joinedload を積極的に活用して N+1 問題を回避しましょう。
クエリのプロファイリング: SQLAlchemy が生成する SQL 文を確認するために、Engine 作成時に echo=True を指定するのは非常に有効です。本番環境ではログレベルを調整するなどして、必要に応じて SQL をログに出力し、遅いクエリを特定します。
適切なインデックスの使用: データベースのクエリ速度はインデックスの有無に大きく依存します。よくフィルタリングや結合条件に使われるカラムにはインデックスを作成することを検討しましょう。SQLAlchemy の Column 定義で index=True や unique=True を指定できます。
大量データの効率的な操作:
- 一括挿入/更新/削除には、Core の insert(), update(), delete() や ORM の一括操作機能（例: session.execute(insert(User), data_list)）が ORM オブジェクトを一つずつ操作するよりも効率的な場合があります。
- 大量データを読み込む場合は、結果セットを一度にすべて fetchall() するのではなく、イテレーションを利用してメモリ消費を抑えることができます。
セッションの適切な管理: セッションはトランザクションの境界を表します。不要になったセッションは速やかにクローズし、長時間実行されるセッションや巨大なセッション（多くのオブジェクトを保持する）は避けるべきです。

4.4 非同期サポート (AsyncIO)

近年の Python Web フレームワーク（FastAPI, Starlette など）では非同期プログラミング（AsyncIO）が広く利用されています。データベース操作は I/O バウンドな処理であるため、非同期で実行することでアプリケーションのスループットを向上させることができます。

SQLAlchemy はバージョン 1.4 から非同期 IO をサポートし、2.0 でより使いやすくなりました。非同期対応のデータベースドライバー（例: asyncpg for PostgreSQL, aiomysql for MySQL, aiosqlite for SQLite）と組み合わせて使用します。

非同期エンジンとセッションを作成するには create_async_engine と async_sessionmaker を使用します。クエリ実行やトランザクション操作は await 可能なメソッドで行います。

“`python

import asyncio

from sqlalchemy.ext.asyncio import create_async_engine, async_sessionmaker, AsyncSession

from sqlalchemy import select

from sqlalchemy.orm import Mapped, mapped_column, declarative_base

Base = declarative_base()

class User(Base):

tablename = “users”

id: Mapped[int] = mapped_column(primary_key=True)

name: Mapped[str]

async def async_main():

# 非同期エンジンを作成

# async_engine = create_async_engine(“postgresql+asyncpg://user:password@host:port/database”, echo=True)

# テーブル作成 (非同期)

# async with async_engine.begin() as conn:

# await conn.run_sync(Base.metadata.create_all) # run_sync を使って同期処理を実行

# 非同期セッションファクトリを作成

# AsyncSessionLocal = async_sessionmaker(async_engine, expire_on_commit=False)

# 非同期セッションを使って操作

# async with AsyncSessionLocal() as session:

# # INSERT (非同期)

# new_user = User(name=’Async Alice’)

# session.add(new_user)

# await session.commit()

# # SELECT (非同期)

# stmt = select(User).where(User.name == ‘Async Alice’)

# result = await session.execute(stmt)

# user = result.scalars().first()

# print(f”Found user: {user}”)

# asyncio.run(async_main())

“`

非同期 SQLAlchemy は、非同期アプリケーションのパフォーマンスを向上させる上で強力な選択肢となりますが、await の使い方など、非同期プログラミングの知識が必要です。

4.5 テスト (Testing)

データベースを使用するアプリケーションのテストは重要です。SQLAlchemy を使ったコードのテストでは、以下のようなアプローチが考えられます。

インメモリ SQLite: 開発中やテスト時には、高速なインメモリ SQLite (sqlite:///:memory:) を利用すると便利です。テストごとに新しいインメモリデータベースを作成し、スキーマを構築することで、クリーンな状態でテストを実行できます。
トランザクションとロールバック: 各テストケースを一つのトランザクション内で実行し、テストの最後に必ずロールバックすることで、テストによるデータベースの変更が他のテストに影響しないようにできます。
テスト用のフィクスチャデータ: テストに必要なデータを事前にデータベースに投入（フィクスチャ）してからテストを実行します。

4.6 Web フレームワークとの連携

SQLAlchemy は、Flask, FastAPI, Django といった様々な Python Web フレームワークと簡単に連携できます。

Flask: Flask-SQLAlchemy 拡張が提供されており、Flask アプリケーション内での SQLAlchemy の設定やセッション管理をより簡単にできます。
FastAPI/Starlette: FastAPI の公式ドキュメントでは、SQLAlchemy ORM と AsyncIO を組み合わせたデータベース連携の方法が詳しく解説されています。依存性注入 (Dependency Injection) を使ってセッションを管理するパターンが一般的です。
Django: Django は独自の ORM を持っていますが、SQLAlchemy を Django プロジェクト内で併用することも技術的には可能です（ただし一般的ではありません）。

これらのフレームワークと連携する場合、フレームワークのドキュメントやコミュニティで推奨されている SQLAlchemy の使い方や統合パターンを確認することが重要です。

5. Core vs ORM: どちらを選ぶべきか？

SQLAlchemy Core と ORM は、それぞれ異なる目的とユースケースを持っています。どちらを選ぶべきか、あるいは組み合わせて使うべきかは、プロジェクトの要件や開発者の好みによって異なります。

SQLAlchemy Core が適しているケース:
- パフォーマンスが最優先される、大量データの一括処理。
- 特定のデータベースの低レベルな機能や複雑な SQL 構造を積極的に利用したい場合。
- 既存の複雑なデータベーススキーマを扱う必要があり、厳密な ORM マッピングが難しい場合。
- ORM の上に独自のデータマッピング層を構築したい場合。
- 比較的単純なデータベース操作のみを行う小規模なスクリプト。
SQLAlchemy ORM が適しているケース:
- ほとんどの Web アプリケーションやビジネスロジック開発。
- オブジェクト指向のパラダイムでデータベース操作を行いたい場合。
- コードの可読性、保守性、開発速度を重視する場合。
- テーブル間のリレーションシップを頻繁に扱う場合。
- N+1 問題などの一般的なパフォーマンス問題を標準的な方法で回避したい場合。

多くのプロジェクトでは ORM がデフォルトの選択肢となります。開発の効率性、保守性、そしてオブジェクト指向の利便性は、Core の提供する低レベルな制御よりも一般的なアプリケーション開発では優先されるからです。

しかし、ORM で表現するのが難しい複雑なクエリや、特定のパフォーマンス要件を満たす必要がある場合は、ORM のクエリ内で Core の要素（例えば func や literal_column、あるいは select オブジェクトそのもの）を組み合わせて使用したり、あるいは Core を直接使用してバッチ処理を行ったりすることも可能です。SQLAlchemy は Core と ORM の間を行き来できる柔軟性を持っています。

重要なのは、どちらか一方だけを使うと決めつけず、それぞれの得意な部分を理解し、必要に応じて使い分けることです。

6. まとめと今後の学習

SQLAlchemy は、Python におけるデータベース連携を効率化するための非常に強力で柔軟なツールキットです。Core が提供するデータベース抽象化と SQL 表現言語は、生の SQL を書くことによる多くの課題を解決し、ORM はオブジェクト指向のパラダイムでデータベースを操作する道を拓きます。

本記事では、SQLAlchemy の基本から、データベース接続、テーブル定義、Core での SQL 表現言語、実行とトランザクション管理、ORM でのマッピング、セッション、永続化、クエリ、リレーションシップとロード戦略、さらにはマイグレーション、エラーハンドリング、パフォーマンス、非同期サポートといった実践的なトピックまでを網羅的に解説しました。

SQLAlchemy は機能が非常に豊富であるため、この記事だけで全てを理解することは難しいかもしれません。さらに学習を進めるためには、以下のリソースが非常に役立ちます。

SQLAlchemy 公式ドキュメント: 最も正確で詳細な情報源です。特に Tutorials セクションは、Core と ORM の基本的な使い方を学ぶのに最適です。最新の SQLAlchemy 2.0 のドキュメントを参照することをお勧めします。
- https://docs.sqlalchemy.org/
Alembic 公式ドキュメント: マイグレーションの詳細について学ぶにはこちらを参照します。
- https://alembic.sqlalchemy.org/
Python コミュニティのリソース: Stack Overflow や Python 関連のフォーラム、ブログなどには、SQLAlchemy の利用例やトラブルシューティングに関する情報が豊富にあります。

SQLAlchemy をマスターすることで、Python アプリケーションにおけるデータベース連携のコードはより安全に、より効率的に、そしてより保守しやすくなります。ぜひ、ご自身のプロジェクトで SQLAlchemy を活用してみてください。

7. 付録: サンプルコード (簡易な CRUD 操作)

記事中で紹介した概念をまとめた簡易的な CRUD 操作のサンプルコード（SQLite, 同期版）を以下に示します。

“`python
import os
from sqlalchemy import create_engine, text # text は生SQL実行用
from sqlalchemy import MetaData, Table, Column, Integer, String, ForeignKey, select, insert, update, delete
from sqlalchemy.orm import declarative_base, sessionmaker, Mapped, mapped_column, relationship
from sqlalchemy.exc import IntegrityError

— データベース接続設定 —

SQLite (ファイル) を使用。もしファイルが存在すればそれを使う

DATABASE_FILE = “./sqlalchemy_example.db”
DATABASE_URL = f”sqlite:///{DATABASE_FILE}”

Engine の作成 (echo=True で実行されるSQLを表示)

engine = create_engine(DATABASE_URL, echo=True)

— Core によるテーブル定義 —

metadata = MetaData()

users_table_core = Table(
“users_core”,
metadata,
Column(“id”, Integer, primary_key=True),
Column(“name”, String(50), nullable=False),
Column(“age”, Integer),
Column(“email”, String(100), unique=True)
)

— ORM によるテーブル定義 (Declarative Mapping) —

Base = declarative_base()

class UserORM(Base):
tablename = “users_orm”
id: Mapped[int] = mapped_column(primary_key=True)
name: Mapped[str] = mapped_column(String(50), nullable=False)
age: Mapped[int | None]
email: Mapped[str | None] = mapped_column(String(100), unique=True)

# リレーションシップは省略（シンプル化のため）

def __repr__(self):
    return f"<UserORM(id={self.id}, name='{self.name}', age={self.age})>"

— データベース初期化 (テーブル作成) —

def setup_database():
# Core のメタデータと ORM のメタデータ (Base.metadata) の両方を使ってテーブルを作成
print(“\n— Setting up database (creating tables) —“)
# Core のテーブルを作成
metadata.create_all(engine)
# ORM のテーブルを作成 (Base.metadata は内部的に Core の MetaData を保持している)
Base.metadata.create_all(engine)
print(“Database setup complete.”)

— Core による CRUD 操作 —

def core_crud_example():
print(“\n— SQLAlchemy Core CRUD Example —“)

with engine.connect() as connection:
    # INSERT
    print("\nInserting data...")
    insert_stmt = insert(users_table_core).values([
        {"name": "Core Alice", "age": 30, "email": "[email protected]"},
        {"name": "Core Bob", "age": 25, "email": "[email protected]"},
    ])
    result = connection.execute(insert_stmt)
    print(f"Inserted {result.rowcount} rows.")
    connection.commit() # with blocks commit by default if no exception

    # SELECT
    print("\nSelecting all data...")
    select_stmt_all = select(users_table_core)
    for row in connection.execute(select_stmt_all):
        print(row)

    # SELECT with WHERE
    print("\nSelecting data with WHERE...")
    select_stmt_where = select(users_table_core).where(users_table_core.c.name == "Core Alice")
    alice = connection.execute(select_stmt_where).fetchone()
    print(f"Found Core Alice: {alice}")

    # UPDATE
    print("\nUpdating data...")
    update_stmt = update(users_table_core).where(users_table_core.c.name == "Core Bob").values(age=26)
    result = connection.execute(update_stmt)
    print(f"Updated {result.rowcount} rows.")
    connection.commit()

    # SELECT to verify update
    print("\nSelecting data after update...")
    select_stmt_bob = select(users_table_core).where(users_table_core.c.name == "Core Bob")
    bob = connection.execute(select_stmt_bob).fetchone()
    print(f"Core Bob after update: {bob}")

    # DELETE
    print("\nDeleting data...")
    delete_stmt = delete(users_table_core).where(users_table_core.c.name == "Core Alice")
    result = connection.execute(delete_stmt)
    print(f"Deleted {result.rowcount} rows.")
    connection.commit()

    # SELECT all after delete
    print("\nSelecting all data after delete...")
    select_stmt_all = select(users_table_core)
    for row in connection.execute(select_stmt_all):
        print(row)

    # Using text() for raw SQL (use with caution!)
    print("\nExecuting raw SQL with text()...")
    raw_sql_select = text("SELECT COUNT(*) FROM users_core;")
    count = connection.execute(raw_sql_select).scalar_one()
    print(f"Count using raw SQL: {count}")
    connection.commit() # Commit may not be needed for SELECT, but good practice in transaction

— ORM による CRUD 操作 —

def orm_crud_example():
print(“\n— SQLAlchemy ORM CRUD Example —“)

# Session ファクトリを作成
SessionLocal = sessionmaker(bind=engine)

# Session を使って操作
with SessionLocal() as session: # with で自動的に close(), commit()/rollback()
    # INSERT
    print("\nInserting data...")
    user1 = UserORM(name="ORM Carol", age=40, email="[email protected]")
    user2 = UserORM(name="ORM Dave", age=35, email="[email protected]")
    session.add_all([user1, user2])
    # session.add(user1)
    # session.add(user2)
    session.commit() # commit 時に INSERT 文が発行される
    print(f"Inserted users with IDs: {user1.id}, {user2.id}")

with SessionLocal() as session:
    # SELECT (All)
    print("\nSelecting all data...")
    # ORM クラスに対して select() を使い、scalars() で ORM オブジェクトとして結果を取得
    stmt_all = select(UserORM)
    users = session.execute(stmt_all).scalars().all()
    for user in users:
        print(user)

    # SELECT with WHERE
    print("\nSelecting data with WHERE...")
    stmt_where = select(UserORM).where(UserORM.name == "ORM Carol")
    carol = session.execute(stmt_where).scalars().one_or_none() # 1件取得、なければ None
    print(f"Found ORM Carol: {carol}")

    # SELECT by primary key (using get)
    print("\nSelecting by primary key...")
    if carol:
        user_by_id = session.get(UserORM, carol.id) # 主キーで1件取得
        print(f"Found user by ID {carol.id}: {user_by_id}")

with SessionLocal() as session:
    # UPDATE
    print("\nUpdating data...")
    dave = session.execute(select(UserORM).where(UserORM.name == "ORM Dave")).scalars().one_or_none()
    if dave:
        dave.age = 36 # オブジェクトの属性値を変更
        # 変更はセッションによって追跡される
        session.commit() # commit 時に UPDATE 文が発行される
        print(f"Updated ORM Dave: {dave}")

with SessionLocal() as session:
    # SELECT to verify update
    print("\nSelecting data after update...")
    dave_after_update = session.get(UserORM, dave.id) if dave else None
    print(f"ORM Dave after update: {dave_after_update}")

with SessionLocal() as session:
    # DELETE
    print("\nDeleting data...")
    carol_to_delete = session.execute(select(UserORM).where(UserORM.name == "ORM Carol")).scalars().one_or_none()
    if carol_to_delete:
        session.delete(carol_to_delete) # オブジェクトを削除対象としてマーク
        session.commit() # commit 時に DELETE 文が発行される
        print(f"Deleted ORM Carol: {carol_to_delete}")

with SessionLocal() as session:
    # SELECT all after delete
    print("\nSelecting all data after delete...")
    stmt_all = select(UserORM)
    users = session.execute(stmt_all).scalars().all()
    for user in users:
        print(user)

— メイン実行部 —

if name == “main“:
# 前回のデータベースファイルを削除 (テスト実行をクリーンにするため)
if os.path.exists(DATABASE_FILE):
os.remove(DATABASE_FILE)

setup_database()
core_crud_example()
orm_crud_example()

# 終了処理は with block や SessionLocal() が担当
print("\n--- Examples Finished ---")

“`

このサンプルコードは、まず Core と ORM それぞれでテーブルを定義し、データベースを初期化します。その後、Core の機能を使って users_core テーブルに対して CRUD 操作を行い、次に ORM の機能を使って users_orm テーブルに対して同様の操作を行います。echo=True により、各操作で SQLAlchemy が生成・実行する SQL 文がコンソールに表示されるため、Core と ORM の違いや内部の挙動を理解するのに役立ちます。

SQLAlchemy 徹底解説: Python 開発での DB 連携を効率化

1. はじめに: なぜ SQLAlchemy を使うのか？

2. Part 1: SQLAlchemy Core – データベース抽象化の基盤

2.1 データベース接続 (Engines)

SQLite (インメモリデータベース)

SQLite (ファイル)

PostgreSQL

engine_postgres = create_engine(“postgresql://user:password@host:port/database”)

MySQL (PyMySQL ドライバーを使用)

engine_mysql = create_engine(“mysql+pymysql://user:password@host:port/database”)

接続プーリング

非同期 IO (AsyncIO) のサポート

from sqlalchemy.ext.asyncio import create_async_engine

async_engine = create_async_engine(“postgresql+asyncpg://user:password@host:port/database”)

2.2 メタデータとテーブル定義 (MetaData, Table, Column)

出力例: dict_keys([‘users’, ‘addresses’])

定義したテーブルをデータベースに作成 (もし存在しなければ)

engine = create_engine(“sqlite:///:memory:”, echo=True)

metadata.create_all(engine)

2.3 SQL 表現言語 (SQL Expression Language)

SELECT 文

全てのユーザーを取得するSELECT文

例: SELECT users.id, users.name, users.age, users.email FROM users

WHERE 句 (フィルタリング)

IDが1のユーザーを取得

例: SELECT users.id, users.name, users.age, users.email FROM users WHERE users.id = :id_1

年齢が30歳以上のユーザーを取得

名前が’Alice’で、かつ年齢が25歳以下のユーザーを取得 (AND)

名前が’Bob’または年齢が40歳以上のユーザーを取得 (OR)

名前が’A’で始まるユーザーを取得 (LIKE)

メールアドレスがNULLでないユーザーを取得 (IS NOT NULL)

複数のIDのいずれかに一致するユーザーを取得 (IN)

これは (users.name = ‘Alice’ AND users.age <= 25) と同じ

ORDER BY 句 (順序付け)

年齢の昇順でユーザーを取得

年齢の降順、名前の昇順でユーザーを取得

LIMIT 句, OFFSET 句 (制限)

最初の10人のユーザーを取得

11番目から20番目までのユーザーを取得 (最初の10件をスキップして次の10件)

JOIN 句 (結合)

ユーザーとそのアドレスを結合

users_table と addresses_table を user_id と id で結合

例: SELECT users.id, users.name, users.age, users.email, addresses.id, addresses.user_id, addresses.email_address FROM users JOIN addresses ON users.id = addresses.user_id

結合するテーブルと結合条件を明確に指定することもできます

外部結合 (LEFT OUTER JOIN など)

users テーブルにアドレスがあるかどうかに関わらず、全てのユーザーとそのアドレス（あれば）を取得

集約関数とグループ化 (func, group_by)

ユーザーの総数をカウント

平均年齢を計算

各年齢層ごとのユーザー数をカウント (GROUP BY)

HAVING 句 (group_by の結果をフィルタリング)

年齢層ごとのユーザー数が5人以上のものを取得

サブクエリと CTE (Common Table Expressions)

サブクエリの例: 平均年齢より若いユーザーを取得

CTE の例 (with_cte() は SQLA 2.0+ で推奨されるスタイル)

user_counts というCTEを定義し、各年齢層のユーザー数をカウント

CTE を使って、ユーザー数が5人以上の年齢層を取得

WITH clause (従来の with_xxx() メソッド)

user_counts_cte = select(users_table.c.age, func.count().label(“count”)).group_by(users_table.c.age).alias(“user_counts_alias”)

s = select(user_counts_cte).where(user_counts_cte.c.count >= 5) # Aliasを使う場合は c.count でアクセス

INSERT, UPDATE, DELETE 文

1件のレコードを挿入

例: INSERT INTO users (name, age, email) VALUES (:name, :age, :email)

複数件のレコードを挿入 (リスト形式でデータを渡す)

IDが1のユーザーの名前を更新

例: UPDATE users SET name=:name WHERE users.id = :id_1

全ユーザーの年齢を+1更新

IDが2のユーザーを削除

例: DELETE FROM users WHERE users.id = :id_1

2.4 実行 (Executing Statements)

engine = create_engine(“sqlite:///:memory:”, echo=True)

metadata.create_all(engine) # テーブル作成

2.5 トランザクション管理

engine = create_engine(“sqlite:///:memory:”, echo=True)

metadata.create_all(engine) # テーブル作成

あるいは try…except…finally で手動管理

connection = engine.connect()

try:

# トランザクション開始

trans = connection.begin()

HAVING 句 (`group_by` の結果をフィルタリング)

3.1 ORM とは何か？なぜ ORM を使うのか？

Dataclass との連携 (`MappedAsDataclass`)