はい、承知いたしました。SQLAlchemyの基本と使い方について、約5000語の詳細な解説記事を作成します。

---

【入門】SQLAlchemyの基本と使い方を徹底解説

はじめに

データベースは、多くのアプリケーション開発において不可欠な要素です。Pythonでデータベースを扱う際、標準ライブラリにはsqlite3などがありますが、より多くの種類のデータベースに対応し、高機能で生産性の高いツールが求められる場面が多くあります。そこで登場するのが SQLAlchemy です。

SQLAlchemyは、Pythonで記述された非常に強力で柔軟なデータベースツールキットおよびオブジェクトリレーショナルマッパー（ORM）です。単なるORMにとどまらず、データベース抽象化ライブラリとしての側面（SQLAlchemy Core）も持ち合わせています。これにより、開発者はデータベースの種類を意識することなく、Pythonコードでデータベース操作を行うことが可能になります。

なぜSQLAlchemyを使うのか？

生のSQLクエリを文字列として記述してデータベースに送る方法は、シンプルで直接的ですが、いくつかの課題があります。

1. データベースの種類に依存する: SQLの構文はデータベースシステム（PostgreSQL, MySQL, SQLite, Oracleなど）によって微妙に異なります。生のSQLを使う場合、データベースを変更するとコードの修正が必要になることがあります。
2. SQLインジェクションのリスク: ユーザーからの入力を直接クエリ文字列に埋め込むと、悪意のあるコードを実行されるSQLインジェクションのリスクが高まります。安全にパラメーターを扱う仕組みが必要です。
3. コードの可読性と保守性: 複雑なクエリは長大な文字列となり、Pythonコードの中に埋め込まれると読みにくく、デバッグやメンテナンスが困難になりがちです。
4. Pythonオブジェクトとの連携: データベースから取得したデータをPythonオブジェクトに変換する、あるいはPythonオブジェクトをデータベースに保存するといった作業は手作業で行う必要があり、煩雑です。

SQLAlchemyはこれらの課題を解決または軽減します。

• データベース抽象化: SQLAlchemy Coreを使えば、Pythonコードでデータベース操作を表現でき、異なるデータベースバックエンドへの切り替えが容易になります。
• 安全なパラメーターバインディング: SQL式の生成機能により、自動的に安全な方法でパラメーターがバインドされ、SQLインジェクションのリスクを低減します。
• SQL式の生成: SQLクエリをPythonコードで組み立てるため、可読性が向上し、より動的なクエリ生成が容易になります。
• ORM機能: SQLAlchemy ORMを使えば、データベースのテーブルを行（レコード）をPythonのオブジェクトとして扱うことができます。オブジェクトの属性変更が自動的にデータベースへの更新に繋がり、オブジェクト間の関連（リレーションシップ）もPythonの属性アクセスで扱えるようになります。これにより、データベース操作がより直感的でオブジェクト指向的なアプローチで行えるようになり、開発効率が大幅に向上します。

この記事の対象読者と目的

この記事は、Pythonを使ったアプリケーション開発でデータベースを扱いたいと考えている方、特にSQLAlchemyを初めて学ぶ方を対象としています。

目的:

• SQLAlchemyの基本的な概念と全体像を理解する。
• SQLAlchemy Coreを使ってデータベースに接続し、基本的なCRUD（作成・読み取り・更新・削除）操作を行う方法を学ぶ。
• SQLAlchemy ORMを使ってデータベーステーブルをPythonクラスとして定義し、オブジェクト指向的なアプローチでCRUD操作や関連操作を行う方法を学ぶ。
• CoreとORMの使い分けについて理解する。

約5000語というボリュームで、SQLAlchemyの基本を網羅し、実践的なコード例を豊富に含めることで、読者の方がこの記事を読み終えた後にSQLAlchemyを使ったデータベース操作を始められるようになることを目指します。

SQLAlchemyの全体像

SQLAlchemyは、大きく分けて「SQLAlchemy Core」と「SQLAlchemy ORM」の2つの主要なコンポーネントから構成されています。

SQLAlchemy Core

SQLAlchemy Coreは、データベースと対話するための基盤を提供します。これは、Pythonのコードを使ってSQLクエリをプログラム的に組み立て、実行するためのツールキットです。生のSQL文字列を扱うのではなく、Pythonオブジェクトとしてテーブルやカラム、SELECT文などを表現します。

Coreの主な役割は以下の通りです。

• データベース接続の管理: Engineオブジェクトを通じてデータベースとの接続を確立・管理します。接続プールなどの機能も提供します。
• データベースメタデータの表現: MetaDataオブジェクトやTableオブジェクトを使って、データベースのスキーマ（テーブル、カラム、型、制約など）をPythonコードで定義・表現します。
• SQL Expression Language: select(), insert(), update(), delete()などの関数を使って、データベースの種類に依存しない方法でSQLクエリを組み立てます。
• クエリの実行: Connectionオブジェクトを使って、生成したSQLクエリをデータベース上で実行し、結果を取得します。

Coreは、データベースの種類に依存しない低レベルなデータベース操作を可能にします。これは、ORMを使用せずに、より直接的にSQLを制御したい場合や、既存のデータベーススキーマを操作する場合、あるいはORMの基盤として利用されます。

SQLAlchemy ORM

SQLAlchemy ORMは、Coreの上に構築されており、データベースのテーブルを行（レコード）をPythonのオブジェクト（ORMクラスのインスタンス）として扱うための機能を提供します。これは「オブジェクトリレーショナルマッピング」と呼ばれる技術です。

ORMの主な役割は以下の通りです。

• テーブルとクラスのマッピング: データベースのテーブルを、Pythonのクラス（Mapped Class）にマッピングします。テーブルのカラムはクラスの属性として扱われます。
• リレーションシップの定義と操作: テーブル間の関連性（一対一、一対多、多対多など）をPythonのコードで定義し、関連するオブジェクトを属性アクセスを通じて取得・操作できるようにします。
• オブジェクト指向なクエリ: SQLクエリを直接書く代わりに、Mapped Classやその属性を使ったPythonコードでクエリを構築します。
• セッション管理: Sessionオブジェクトを通じて、Pythonオブジェクトのライフサイクル（作成、取得、変更、削除）とデータベースの同期（トランザクション）を管理します。

ORMは、アプリケーション開発において、より生産的で直感的なデータベース操作を提供します。特に、ビジネスロジックがオブジェクト指向的に設計されている場合に強力です。

CoreとORMの違いと使い分け

特徴	SQLAlchemy Core	SQLAlchemy ORM
抽象レベル	低い（SQLに近い）	高い（オブジェクト指向）
操作対象	テーブル、カラム、SQL式	Pythonオブジェクト、クラス属性、リレーションシップ
クエリ記述	select(), insert(), etc. を使ったSQL式	Mapped Classやその属性を使ったメソッドチェーン（旧:query())、またはSQL式+session.execute()
パフォーマンス	より細かく制御可能（直接的）	Coreよりオーバーヘッドがあるが、適切に使えば高効率
学習コスト	SQLの概念を理解している必要がある	ORMの概念（マッピング、セッション、リレーションシップ）を理解する必要がある
適した用途	パフォーマンス重視、複雑な・非定型クエリ、DBスキーマ操作、既存DBへのアクセス	アプリケーションロジック中心の開発、オブジェクト操作、生産性重視

CoreとORMは排他的なものではありません。ORMはCoreの上に構築されており、多くのORMの機能は内部的にCoreを使用しています。必要に応じて、ORMを使っているプロジェクトでも、特定の処理のためにCoreの機能を使うことも可能です。

インストール方法

SQLAlchemyはpipを使って簡単にインストールできます。

bash
pip install SQLAlchemy

特定のデータベース（例: PostgreSQL, MySQL, SQLite以外）に接続する場合は、別途そのデータベース用のドライバーもインストールする必要があります。

• PostgreSQL: pip install psycopg2 (または psycopg2-binary)
• MySQL: pip install mysql-connector-python (または PyMySQL)
• SQLite: Python標準ライブラリに含まれているため、追加インストールは不要です。

この記事では主にSQLiteを使いますが、他のデータベースでも接続文字列を変更するだけで基本的な操作は同じです。

それでは、まずSQLAlchemy Coreから詳しく見ていきましょう。

Core編 (SQLAlchemy Core)

SQLAlchemy Coreは、データベースの種類に依存しない低レベルなデータベース操作を可能にします。PythonのコードでSQLクエリを組み立て、実行します。

Coreの役割とメリット

• データベースの抽象化: データベース固有のSQL方言やAPIの違いを吸収し、共通のPythonコードでデータベースを操作できます。
• プログラム的なSQL生成: SQLクエリを文字列ではなくPythonオブジェクトとして組み立てるため、動的なクエリ生成や複雑な条件指定が容易になります。
• 安全なパラメーターバインディング: ユーザー入力などをパラメーターとしてクエリに渡す際に、自動的に適切に処理され、SQLインジェクションを防ぎます。
• メタデータの表現: データベースのスキーマ構造をPythonコードで表現・管理できます。

Coreは、データベースのスキーマ定義、データベース接続、トランザクション管理、SQL式の生成と実行といった基盤機能を提供します。ORMはこれらのCoreの機能を利用して構築されています。

Engine: データベース接続の管理

SQLAlchemyにおけるデータベース接続の入り口となるのが Engine オブジェクトです。Engineはデータベースへの接続を確立・管理し、接続プールを提供します。

接続文字列 (Connection String) のフォーマット

Engineを作成するには、接続したいデータベースを指定する接続文字列が必要です。接続文字列の一般的なフォーマットは以下のようになります。

dialect+driver://user:password@host:port/database?key=value

各要素の説明:

• dialect: データベースの種類（例: postgresql, mysql, sqlite, oracle）
• driver: 接続に使用するDBドライバーライブラリ（省略可能な場合が多い。例: psycopg2, mysqlconnector, pymysql）
• user: データベースユーザー名
• password: データベースパスワード
• host: データベースサーバーのホスト名またはIPアドレス
• port: データベースサーバーのポート番号（省略可。デフォルトポートが使われる）
• database: データベース名（SQLiteの場合はファイルパス）
• ?key=value: 追加の接続オプション（省略可）

いくつか例を見てみましょう。

• SQLite (インメモリ):
  python
"sqlite:///:memory:"
• SQLite (ファイル):
  python
"sqlite:///path/to/your/database.db" # 絶対パスまたは相対パス
  Windowsの場合、パスに注意が必要な場合があります。例: "sqlite:///C:\\path\\to\\database.db"
• PostgreSQL:
  python
"postgresql://user:password@host:port/database"
• MySQL:
  python
"mysql+mysqlconnector://user:password@host:port/database" # mysqlconnector driverを使用
"mysql+pymysql://user:password@host:port/database" # PyMySQL driverを使用

Engineの作成と接続プール

create_engine() 関数を使ってEngineを作成します。

“`python
from sqlalchemy import create_engine

SQLite (インメモリ) のEngineを作成

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

echo=True を指定すると、実行されるSQLクエリが標準出力に表示される (デバッグに便利)

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

“`

create_engine は、引数で指定された設定に基づいてEngineインスタンスを作成します。Engineはスレッドセーフであり、アプリケーション全体でシングルトンとして扱うのが一般的です。

Engine は、データベースへの物理的な接続を直接行うのではなく、接続プールを管理します。接続プールは、あらかじめデータベースへの接続をいくつか作成しておき、必要に応じてそれを使い回す仕組みです。これにより、接続の確立・切断にかかるオーバーヘッドを削減し、パフォーマンスを向上させます。

dispose()

Engineが管理する全ての接続を閉じるには、dispose() メソッドを使用します。これはアプリケーションの終了時などに呼び出すと良いでしょう。

python
engine.dispose()

MetaDataTable: データベーススキーマの表現

SQLAlchemy Coreでは、データベースのスキーマ（テーブル、カラム、型など）を MetaData オブジェクトを使ってPythonコードで表現します。

MetaData()

MetaData オブジェクトは、データベース全体のテーブル定義やその他のスキーマ要素を保持するコンテナです。

“`python
from sqlalchemy import MetaData

metadata = MetaData()
“`

Tableオブジェクトの定義

Table オブジェクトは、データベース内の個々のテーブルを表現します。MetaData オブジェクトに関連付けて定義します。

Table オブジェクトのコンストラクタには、以下の情報を指定します。

1. テーブル名: 必須
2. MetaData オブジェクト: 必須
3. Column オブジェクト: テーブルのカラムを1つ以上指定します。

Column オブジェクトのコンストラクタには、以下の情報を指定します。

1. カラム名: 必須
2. データ型: SQLAlchemyの型オブジェクトを指定（例: Integer, String, Text, Boolean, Date, DateTime, Float, Numeric など）。
3. カラムレベルの制約やオプション:
   • primary_key=True: 主キー
   • autoincrement=True / autoincrement="auto": 自動採番 (Integer型の主キーでよく使用)
   • nullable=False: NOT NULL制約
   • unique=True: UNIQUE制約
   • default=value: デフォルト値
   • server_default=SQL: データベース側でのデフォルト値
   • index=True: インデックス作成

例：users テーブルと posts テーブルの定義

“`python
from sqlalchemy import Table, Column, Integer, String, Text, DateTime, ForeignKey
from sqlalchemy import MetaData
from datetime import datetime

metadata = MetaData()

users_table = Table(
“users”, metadata,
Column(“id”, Integer, primary_key=True, autoincrement=True),
Column(“name”, String(50), nullable=False),
Column(“email”, String(100), nullable=False, unique=True),
Column(“created_at”, DateTime, server_default=func.now()), # func.now() は後述
)

posts_table = Table(
“posts”, metadata,
Column(“id”, Integer, primary_key=True, autoincrement=True),
Column(“title”, String(200), nullable=False),
Column(“content”, Text),
Column(“user_id”, Integer, ForeignKey(“users.id”), nullable=False), # 外部キー
Column(“created_at”, DateTime, server_default=func.now()),
)
“`

ここではまだfunc.now()は定義していませんが、後述のSQL式生成で説明します。ForeignKey("users.id") のように文字列で テーブル名.カラム名 を指定することで外部キーを定義できます。

テーブルの作成・削除

MetaData オブジェクトに紐付けられた全てのテーブルをデータベース上に作成するには、metadata.create_all(engine) を使用します。

“`python
from sqlalchemy import create_engine, MetaData, Table, Column, Integer, String, Text, DateTime, ForeignKey, func

Engineを作成 (ここではSQLiteインメモリ)

engine = create_engine(“sqlite:///:memory:”, echo=True) # echo=TrueでSQLを表示

metadata = MetaData()

users_table = Table(
“users”, metadata,
Column(“id”, Integer, primary_key=True, autoincrement=True),
Column(“name”, String(50), nullable=False),
Column(“email”, String(100), nullable=False, unique=True),
Column(“created_at”, DateTime, server_default=func.now()),
)

posts_table = Table(
“posts”, metadata,
Column(“id”, Integer, primary_key=True, autoincrement=True),
Column(“title”, String(200), nullable=False),
Column(“content”, Text),
Column(“user_id”, Integer, ForeignKey(“users.id”), nullable=False),
Column(“created_at”, DateTime, server_default=func.now()),
)

metadataに紐付けられた全てのテーブルを作成

metadata.create_all(engine)
“`

実行すると、SQLAlchemyがデータベースの種類に応じた CREATE TABLE 文を生成して実行します。echo=True にしている場合、そのSQLが表示されます。

同様に、テーブルを削除するには metadata.drop_all(engine) を使用します。

“`python

全てのテーブルを削除

metadata.drop_all(engine)
“`

ConnectionとExecutionContext

データベースへの実際の操作（SQLクエリの実行）は、Connection オブジェクトを通じて行います。Connection は Engine から取得します。

Connectionの取得と管理

Engine.connect() メソッドで Connection オブジェクトを取得します。Connection オブジェクトはデータベースへの単一の接続を表します。

Connection はリソースであり、使い終わったら閉じる必要があります。Pythonの with ステートメントを使うと、Connectionが自動的にクローズされるため便利です。

python
with engine.connect() as connection:
# ここでデータベース操作を行う
pass # connection が自動的にクローズされる

トランザクション

Connection はデフォルトではオートコミットモードではないため、INSERT, UPDATE, DELETE のようなデータを変更する操作を行った後、変更を永続化するためにはトランザクションをコミットする必要があります。

with connection.begin(): を使うと、ブロックの最後に自動的にコミットされ、例外が発生した場合はロールバックされます。これはSQLの BEGIN, COMMIT, ROLLBACK に相当します。

python
with engine.connect() as connection:
with connection.begin(): # トランザクションを開始
# ここでデータを変更する操作を行う
# 例: connection.execute(insert(...))
pass # ブロックの終了時に自動的にコミットされる
# または、手動でコミット/ロールバック
# transaction = connection.begin()
# try:
# # データを変更する操作
# transaction.commit()
# except:
# transaction.rollback()
# raise

クエリの実行 (execute())

Connection オブジェクトの execute() メソッドを使って、SQL式や生のSQL文字列を実行します。生のSQL文字列を実行する場合はSQLインジェクションに注意が必要です。SQLAlchemy Coreの利点を活かすには、SQL式を生成して実行することを推奨します。

execute() メソッドは Result オブジェクトを返します。この Result オブジェクトを使ってクエリの結果を取得します。

結果セットの取得

execute() が返す Result オブジェクトには、結果セットを操作するための様々なメソッドがあります。

• fetchone(): 結果セットの次の1行を取得します。行がない場合は None を返します。
• fetchall(): 結果セットの全ての行をリストとして取得します。
• first(): 結果セットの最初の1行を取得します。結果がない場合は None を返します。fetchone() と似ていますが、通常はクエリに LIMIT 1 を追加することで効率的に最初の1行だけを取得する用途で使われます。
• scalar(): 結果セットの最初の行の最初のカラムの値を取得します。結果がない場合は None を返します。単一の値を取得するクエリに適しています。
• scalars(): 結果セットの各行の最初のカラムの値だけを抽出したイテレーターを返します。
• イテレーション: Result オブジェクト自体をイテレートすることで、1行ずつ結果を取得できます。

各行は、タプルやリストのようにアクセス可能なオブジェクトとして返されます。カラム名やインデックスで値にアクセスできます。

“`python

結果の取得例 (詳細は後述のselect()と組み合わせて)

result = connection.execute(select(…))

1行ずつ処理

for row in result:
print(row[‘column_name’], row[0]) # カラム名またはインデックスでアクセス

全ての行を取得

all_rows = result.fetchall()

最初の1行を取得

first_row = result.fetchone()

最初の行の最初のカラムの値を取得

single_value = result.scalar()
“`

SQL式の生成 (SQL Expression Language)

SQLAlchemy Coreの強力な機能の一つが、PythonコードでSQLクエリを組み立てるSQL Expression Languageです。これにより、データベースの種類に依存しない、安全でプログラム可能なSQL操作が可能になります。

主要なSQL式生成関数は以下の通りです。

• select(): SELECT文を生成します。
• insert(): INSERT文を生成します。
• update(): UPDATE文を生成します。
• delete(): DELETE文を生成します。

これらの関数は、先ほど定義した Table オブジェクトやその Column オブジェクトと組み合わせて使用します。

例：users_table と posts_table を使用

“`python
from sqlalchemy import select, insert, update, delete, func
from sqlalchemy.orm import aliased # ORMではないが、エイリアスに使うことがある

… metadata, users_table, posts_table の定義 …

INSERT文の生成と実行

with engine.connect() as connection:
with connection.begin():
# データを挿入
result = connection.execute(
insert(users_table).values(name=’Alice’, email=’[email protected]’)
)
# 自動採番された主キーを取得 (insertの結果から取得可能)
user_id = result.lastrowid # SQLiteの場合。他のDBでは異なる場合あり (e.g., returned_defaults)
print(f”Inserted user with ID: {user_id}”)

    # 複数行の挿入
    connection.execute(
        insert(posts_table),
        [
            {'title': 'First Post', 'content': 'Content 1', 'user_id': user_id},
            {'title': 'Second Post', 'content': 'Content 2', 'user_id': user_id},
        ]
    )
    print("Inserted multiple posts.")

“`

“`python

SELECT文の生成と実行

from sqlalchemy import select

with engine.connect() as connection:
# 全ユーザーを選択
stmt = select(users_table)
result = connection.execute(stmt)
print(“\n— All Users —“)
for row in result:
print(row)

# 特定のカラムを選択
stmt = select(users_table.c.name, users_table.c.email) # .c でカラムにアクセス
result = connection.execute(stmt)
print("\n--- User Names and Emails ---")
for row in result:
    print(row) # タプルで返される

# 条件付きで選択 (WHERE句)
stmt = select(users_table).where(users_table.c.name == 'Alice')
result = connection.execute(stmt)
print("\n--- User named Alice ---")
print(result.fetchone())

# 複数の条件 (AND)
from sqlalchemy import and_
stmt = select(users_table).where(
    and_(
        users_table.c.name == 'Alice',
        users_table.c.email == '[email protected]'
    )
)
# または、.where() をチェインする
# stmt = select(users_table).where(users_table.c.name == 'Alice').where(users_table.c.email == '[email protected]')
result = connection.execute(stmt)
print("\n--- User named Alice with specific email ---")
print(result.fetchone())

# 複数の条件 (OR)
from sqlalchemy import or_
stmt = select(users_table).where(
    or_(
        users_table.c.name == 'Alice',
        users_table.c.name == 'Bob' # Bobは存在しないはず
    )
)
result = connection.execute(stmt)
print("\n--- Users named Alice or Bob ---")
for row in result:
    print(row)

# IN句
stmt = select(users_table).where(users_table.c.name.in_(['Alice', 'Bob']))
result = connection.execute(stmt)
print("\n--- Users named Alice or Bob (using in_) ---")
for row in result:
    print(row)

# LIKE句
stmt = select(users_table).where(users_table.c.email.like('%@example.com'))
result = connection.execute(stmt)
print("\n--- Users with email ending in @example.com ---")
for row in result:
    print(row)

# NULLチェック
# stmt = select(users_table).where(users_table.c.some_column.is_(None))
# stmt = select(users_table).where(users_table.c.some_column.isnot(None))


# 結合 (JOIN)
# users_table と posts_table を user_id で結合し、ユーザー名と投稿タイトルを取得
# デフォルトでは内部結合 (INNER JOIN)
stmt = select(
    users_table.c.name,
    posts_table.c.title
).join(posts_table, users_table.c.id == posts_table.c.user_id)
result = connection.execute(stmt)
print("\n--- User Names and Post Titles (Joined) ---")
for row in result:
    print(row)

# 外部結合 (LEFT OUTER JOIN)
stmt = select(
    users_table.c.name,
    posts_table.c.title
).join(posts_table, users_table.c.id == posts_table.c.user_id, isouter=True)
result = connection.execute(stmt)
print("\n--- User Names and Post Titles (Left Outer Joined) ---")
for row in result:
    print(row)

# ソート (ORDER BY)
stmt = select(posts_table).order_by(posts_table.c.created_at.desc()) # 降順
# stmt = select(posts_table).order_by(posts_table.c.created_at) # 昇順 (デフォルト)
result = connection.execute(stmt)
print("\n--- Posts ordered by creation date (desc) ---")
for row in result:
    print(row)

# リミットとオフセット (LIMIT, OFFSET)
stmt = select(posts_table).limit(1).offset(1) # 2番目の投稿を取得
result = connection.execute(stmt)
print("\n--- Second Post (Limit/Offset) ---")
print(result.fetchone())

# 集計関数 (Aggregate Functions) と グループ化 (GROUP BY)
from sqlalchemy import func
# func.count(), func.sum(), func.avg(), func.min(), func.max() など
stmt = select(users_table.c.name, func.count(posts_table.c.id).label('post_count')).\
       join(posts_table, users_table.c.id == posts_table.c.user_id).\
       group_by(users_table.c.name)
result = connection.execute(stmt)
print("\n--- User Post Counts (Grouped) ---")
for row in result:
    print(row)

# GROUP BY に HAVING句を追加
# stmt = stmt.having(func.count(posts_table.c.id) > 1) # 投稿数が1より多いユーザーのみ
# result = connection.execute(stmt)
# print("\n--- User Post Counts (Having > 1) ---")
# for row in result:
#     print(row)

“`

“`python

UPDATE文の生成と実行

with engine.connect() as connection:
with connection.begin():
stmt = update(users_table).where(users_table.c.name == ‘Alice’).values(name=’Alicia’)
result = connection.execute(stmt)
print(f”\nUpdated {result.rowcount} user(s).”) # 影響を受けた行数

# 更新されたか確認
stmt = select(users_table).where(users_table.c.name == 'Alicia')
result = connection.execute(stmt)
print("\n--- User named Alicia ---")
print(result.fetchone())

“`

“`python

DELETE文の生成と実行

with engine.connect() as connection:
with connection.begin():
stmt = delete(posts_table).where(posts_table.c.title.like(‘%Second%’))
result = connection.execute(stmt)
print(f”\nDeleted {result.rowcount} post(s).”) # 影響を受けた行数

# 削除されたか確認
stmt = select(posts_table).where(posts_table.c.title.like('%Second%'))
result = connection.execute(stmt)
print("\n--- Posts with 'Second' in title (after delete) ---")
print(result.fetchall()) # [] が返されるはず

“`

SQL Expression Languageは、これらの基本的な操作以外にも、サブクエリ、共通テーブル式（CTE）、ウィンドウ関数など、複雑なSQL構文を表現するための豊富な機能を提供します。PythonコードとしてSQLを組み立てることで、静的な文字列よりも柔軟かつ安全にデータベースを操作できます。

Coreは低レベルな操作を提供するため、データベースとの対話や複雑なSQLクエリの実行には非常に強力です。しかし、Pythonオブジェクトとデータベースレコードのマッピングを手動で行う必要があるため、アプリケーション開発の生産性という点ではORMに一歩譲ります。次に、そのORMについて詳しく見ていきましょう。

ORM編 (SQLAlchemy ORM)

SQLAlchemy ORMは、データベースのテーブルを行（レコード）をPythonのオブジェクトとして扱うための機能を提供します。これにより、データベース操作をよりオブジェクト指向的に行うことができ、アプリケーションのビジネスロジックとデータベースの間のギャップを埋めます。

ORMの役割とメリット

• オブジェクトとレコードのマッピング: データベースの行をPythonのクラスインスタンスに、カラムをオブジェクトの属性に自動的にマッピングします。
• リレーションシップ管理: データベースの外部キーによる関連性を、Pythonオブジェクト間の関連として扱うことができます。例えば、user.posts のように属性にアクセスするだけで、関連する投稿オブジェクトを取得できます。
• オブジェクト指向なクエリ: Pythonのクラスや属性を使ってクエリを記述できるため、可読性が高く、リファクタリングも容易になります。
• セッションによる変更追跡: オブジェクトの属性変更などを自動的に追跡し、コミット時に適切なSQL文を生成してデータベースに反映します。
• 生産性の向上: 低レベルなSQL操作から解放され、ビジネスロジックの実装に集中できます。

基本的な概念

SQLAlchemy ORMを使う上で重要な概念をいくつか説明します。

Declarative Base

ORMクラス（Mapped Class）を定義する際に、テーブルとクラスのマッピング情報を宣言的に記述するための基盤となるクラスです。declarative_base() 関数を使って作成します。

“`python
from sqlalchemy.orm import declarative_base

Base = declarative_base()
“`

Base クラスを継承して、ORMクラスを定義します。Base には、先ほどCoreで見た MetaData オブジェクトも含まれています。

Mapped Classes

データベースのテーブルに対応するPythonクラスです。declarative_base() で作成した Base クラスを継承して定義します。

Pythonのクラス属性として、データベーステーブルのカラムや関連性（リレーションシップ）を定義します。

SQLAlchemy 2.0以降では、型ヒントを使った新しい宣言的なマッピングスタイルが推奨されています。Mapped 型と mapped_column() 関数を使用します。

例：User クラスと Post クラスの定義 (Coreの users_table と posts_table に対応)

“`python
from sqlalchemy import create_engine, Column, Integer, String, Text, DateTime, ForeignKey, func
from sqlalchemy.orm import declarative_base, relationship, Mapped, mapped_column # relationshipを追加
from datetime import datetime

Engineを作成

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

Declarative Baseを作成

Base = declarative_base()

Mapped Classesの定義

class User(Base):
tablename = “users” # テーブル名を指定

# mapped_column() でカラムを定義
# 型ヒントとして Mapped[データベースの型に対応するPythonの型] を指定
id: Mapped[int] = mapped_column(primary_key=True) # Integer, autoincrement="auto" はデフォルトでつく
name: Mapped[str] = mapped_column(String(50), nullable=False)
email: Mapped[str] = mapped_column(String(100), nullable=False, unique=True)
created_at: Mapped[datetime] = mapped_column(DateTime, server_default=func.now())

# Relationship: このユーザーが書いた投稿のリスト
# 'Post' は関連するクラス名 (文字列でも可)
# backref は関連する Post クラスから User への参照を自動で作成する
posts: Mapped[list["Post"]] = relationship("Post", back_populates="author")

def __repr__(self):
    # オブジェクトの表現を指定 (デバッグ時に便利)
    return f"<User(id={self.id}, name='{self.name}', email='{self.email}')>"

class Post(Base):
tablename = “posts” # テーブル名を指定

id: Mapped[int] = mapped_column(primary_key=True)
title: Mapped[str] = mapped_column(String(200), nullable=False)
content: Mapped[str | None] = mapped_column(Text) # None を許容する場合は Union を使う
# 外部キーのカラム定義
user_id: Mapped[int] = mapped_column(ForeignKey("users.id"), nullable=False)
created_at: Mapped[datetime] = mapped_column(DateTime, server_default=func.now())

# Relationship: この投稿の著者 (単数オブジェクト)
# 'User' は関連するクラス名
# back_populates は関連する User クラス側の relationship 名を指定
author: Mapped["User"] = relationship("User", back_populates="posts")

def __repr__(self):
    return f"<Post(id={self.id}, title='{self.title[:20]}...', user_id={self.user_id})>"

Base.metadata に紐付けられた全てのテーブルを作成

Base.metadata.create_all(engine)
print(“Tables created.”)
“`

この例では、User クラスと Post クラスがそれぞれ users テーブルと posts テーブルに対応します。

• __tablename__: 対応するデータベーステーブル名を指定します。
• クラス属性 (id, name, email など): mapped_column() と型ヒントを使って、テーブルのカラムに対応する属性を定義します。
  • Mapped[int]: データベースの Integer 型に対応し、Pythonでは int として扱われます。
  • Mapped[str]: データベースの String 型に対応し、Pythonでは str として扱われます。
  • Mapped[str | None]: データベースの Text 型に対応し、Nullableなカラムを表します。Pythonでは str または None が入ります。
  • primary_key=True, nullable=False, unique=True: カラムレベルの制約を指定します。Integer型の主キーにはデフォルトで autoincrement="auto" が適用されます。
  • ForeignKey("users.id"): 外部キー制約を定義します。
• relationship(): テーブル間の関連性を定義します。
  • User.posts: User オブジェクトから、そのユーザーが作成した複数の Post オブジェクトにアクセスするための属性です。Mapped[list["Post"]] はリストであることを示します。
  • Post.author: Post オブジェクトから、その投稿を作成した単一の User オブジェクトにアクセスするための属性です。Mapped["User"] は単一オブジェクトであることを示します。
  • back_populates: 関連性の両側をリンクさせます。User.posts と Post.author が互いに対応することを指定しています。これにより、例えば post.author にアクセスすると対応するUserが得られ、そのUserオブジェクトにはそのpostを含む user.posts リストが自動的にロードされます（ロード戦略による）。

新しい宣言的スタイルでは、Mapped[Type] という型ヒントが必須です。これにより、静的解析ツールによるチェックが容易になり、コードの可読性も向上します。mapped_column() は、Coreの Column オブジェクトを作成しますが、ORMのマッピング情報も含まれます。

Session: ORM操作の主要インターフェース

ORM操作の中心となるのが Session オブジェクトです。Session は「作業単位」を表し、データベースからのオブジェクトのロード、新しいオブジェクトの永続化、オブジェクトの変更追跡、変更のデータベースへの同期（コミット）などを行います。

Coreの Connection がデータベースとの物理的な接続を表すのに対し、Session はORMオブジェクトの状態管理とトランザクション管理を中心に扱います。Session は内部的に Connection を使用します。

Sessionの作成 (sessionmaker)

Session オブジェクトは、通常 sessionmaker クラスを使って作成したファクトリから生成します。sessionmaker はEngineと紐付けられ、Sessionを作成する際のデフォルト設定（例: コミットやフラッシュのタイミング）を定義します。

“`python
from sqlalchemy.orm import sessionmaker

Engineを作成 (上記で作成済み)

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

Sessionファクトリを作成

Session = sessionmaker(bind=engine) # bind=engine でどのEngineを使うか指定
“`

Session ファクトリ（上記の Session）はスレッドセーフなので、アプリケーション全体で共有して構いません。しかし、Session インスタンス自体はスレッドセーフではないため、リクエストごと、あるいは論理的な作業単位ごとに新しいSessionインスタンスを作成し、使い終わったら閉じるのが正しい使い方です。

Sessionのライフサイクルとスコープ

最も推奨されるSessionの管理方法は、with ステートメントを使用することです。これにより、ブロックの終了時にSessionが自動的に閉じられます。

“`python

SessionファクトリからSessionインスタンスを作成

with Session() as session:
# ここでORM操作を行う
# session.add(…)
# session.execute(…)
# …

session.commit() # または session.rollback()
# ブロック終了時に session.close() が自動的に呼び出される

“`

この with ブロック内で、データベースからロードされたオブジェクト、新しく作成されてSessionに追加されたオブジェクト、変更されたオブジェクトなどの状態が管理されます。

オブジェクトの追加 (add(), add_all())

新しくPythonオブジェクトを作成し、データベースに保存したい場合、そのオブジェクトをSessionに追加します。

“`python
with Session() as session:
# 新しいユーザーオブジェクトを作成
new_user = User(name=’Bob’, email=’[email protected]’)

# Sessionに追加
session.add(new_user)

# まだデータベースには保存されていない（Pending状態）

# 変更をデータベースに永続化
session.commit()
# commit 後、new_user オブジェクトはデータベースに保存され、idが付与される (Persistent状態)

print(f"Added user with ID: {new_user.id}")

# 複数オブジェクトの追加
user_from_db = session.get(User, new_user.id) # セッションからオブジェクトを取得
new_posts = [
    Post(title='Bob Post 1', content='Content B1', author=user_from_db), # relationship経由で関連付け
    Post(title='Bob Post 2', content='Content B2', user_id=user_from_db.id), # 外部キー経由で関連付け
]
session.add_all(new_posts)
session.commit()
print("Added multiple posts for Bob.")

“`

session.add() または session.add_all() でオブジェクトをSessionに追加しただけでは、データベースへの反映は行われません。session.commit() または session.flush() を呼び出したときに、Sessionが変更を検知し、対応する INSERT 文などを生成して実行します。commit() はトランザクションを終了（コミット）しますが、flush() はトランザクションを継続しつつ、Session内の変更をデータベースに書き出します。通常は commit() を使います。

オブジェクトの取得 (クエリ)

ORMを使ってデータベースからオブジェクトを取得する最も一般的な方法は、select() 関数と session.execute() メソッドを組み合わせる方法です。SQL Expression Languageでクエリを構築し、それをSessionで実行します。

結果の取得方法はCoreの場合と似ていますが、ORMでは通常、データベースの行ではなくMapped Classのインスタンスとして結果を取得したいと考えます。そのために Result オブジェクトの scalars() メソッドがよく使われます。scalars() は、結果セットの各行の最初のカラム（通常はORMクラスのインスタンス）を抽出したイテレーターを返します。

“`python
from sqlalchemy import select

with Session() as session:
# 全ユーザーを取得 (Mapped Classのインスタンスとして)
# select(User) は SELECT * FROM users に相当し、結果は User オブジェクトとしてマッピングされる
stmt = select(User)
# execute() は Result オブジェクトを返す
# scalars() は Result オブジェクトから User インスタンスだけを抽出したイテレーターを返す
for user in session.scalars(stmt):
print(user)

# 特定のユーザーをIDで取得 (主キーによる取得)
# session.get(Class, primary_key_value) を使うのが最も簡単で効率的
user_by_id = session.get(User, 1) # ID=1のユーザーを取得
print(f"\n--- User with ID 1 ---")
print(user_by_id) # User(id=1, name='Alicia', email='[email protected]') が表示されるはず

# WHERE句によるフィルタリング
stmt = select(User).where(User.name == 'Bob') # クラス属性を使って条件を指定
bob_user = session.scalars(stmt).first() # 結果の最初の1つを取得
print(f"\n--- User named Bob ---")
print(bob_user)

# 複数の条件 (AND, OR)
from sqlalchemy import and_, or_
stmt = select(User).where(
    and_(
        User.name.in_(['Alicia', 'Bob']),
        User.email.like('%@example.com')
    )
)
for user in session.scalars(stmt):
    print(f"\n--- Users named Alicia or Bob with @example.com email ---")
    print(user)

# ORDER BY によるソート
stmt = select(Post).order_by(Post.created_at.desc())
print("\n--- All Posts (Ordered by Creation Date Desc) ---")
for post in session.scalars(stmt):
    print(post)

# LIMIT と OFFSET
stmt = select(Post).offset(1).limit(1) # 2番目の投稿を取得
second_post = session.scalars(stmt).first()
print(f"\n--- Second Post (Limit/Offset) ---")
print(second_post)

# 結合 (JOIN) と関連オブジェクトのロード
# ユーザー名と投稿タイトルを取得
# select(User, Post) のように複数のエンティティを選択すると、タプルのResultが返る
# scalars() は最初の要素だけを抽出するため、ここでは select(User) や select(Post) を使う方が典型的
stmt = select(User, Post).join(Post) # UserクラスとPostクラス間のrelationshipを使って自動でJOIN条件を生成
print("\n--- User and Post (Joined) ---")
for user, post in session.execute(stmt): # 複数のエンティティを選択した場合、execute()の結果を直接イテレート
    print(f"User: {user.name}, Post Title: {post.title}")

# 関連オブジェクトを効率的にロード (ロード戦略)
# 通常、関連オブジェクトへのアクセスは遅延ロード (Lazy Loading) される
# 例えば、Userオブジェクトを取得した後、user.posts にアクセスした時点で初めて posts テーブルへのクエリが実行される
# N+1問題を避けるため、結合を使って一度にロードすることが推奨される
# 例えば、ユーザーとその投稿を一度にロードするには selectinload や joinedload を使う
from sqlalchemy.orm import selectinload
stmt = select(User).options(selectinload(User.posts)) # Userオブジェクトを取得する際に、関連するPostも追加クエリで取得
print("\n--- Users with Posts (using selectinload) ---")
for user in session.scalars(stmt):
    print(f"User: {user.name}")
    for post in user.posts: # このアクセスで追加クエリは発生しない
        print(f"  Post: {post.title}")

from sqlalchemy.orm import joinedload
stmt = select(User).options(joinedload(User.posts)) # Userオブジェクトを取得するクエリにLEFT JOINを追加し、投稿も一緒に取得
print("\n--- Users with Posts (using joinedload) ---")
for user in session.scalars(stmt):
    print(f"User: {user.name}")
    for post in user.posts: # このアクセスで追加クエリは発生しない
        print(f"  Post: {post.title}")

# 注意: joinedload は LEFT JOIN を生成するため、関連する Post がない User も結果に含まれる
# selectinload は IN 句を使った追加クエリを生成するため、対象のユーザー数が多い場合に効率的

# 個別のカラムを選択し、タプルで取得
stmt = select(User.name, User.email).where(User.id == 1)
result = session.execute(stmt).first() # タプルで結果を取得
print(f"\n--- User Name and Email for ID 1 (as tuple) ---")
print(result) # ('Alicia', '[email protected]') のようなタプル
if result:
    print(result.name, result.email) # タプルでもカラム名でアクセス可能

“`

オブジェクトの更新 (属性の変更)

Sessionで管理されている（Persistent状態の）オブジェクトの属性値を変更すると、Sessionはその変更を追跡します。commit() または flush() を呼び出したときに、適切な UPDATE 文が生成されてデータベースに反映されます。

“`python
with Session() as session:
# 更新したいオブジェクトをデータベースから取得
user_to_update = session.get(User, 1) # ID=1のユーザー (Alicia)

if user_to_update:
    # オブジェクトの属性を変更
    user_to_update.name = 'Alice (Updated)'
    # セッションはこの変更を追跡している

    # 変更をデータベースにコミット
    session.commit()
    print(f"\nUpdated user ID 1 name to: {user_to_update.name}")

# 変更が反映されたか確認
updated_user = session.get(User, 1)
print(f"Verified updated name: {updated_user.name}")

“`

特別な session.update() メソッドはありません。Sessionがオブジェクトを追跡している限り、属性の変更は自動的に検知されます。

オブジェクトの削除 (delete())

Sessionで管理されているオブジェクトを削除するには、session.delete() メソッドを使用し、その後 commit() または flush() します。

“`python
with Session() as session:
# 削除したいオブジェクトをデータベースから取得
user_to_delete = session.get(User, 2) # ID=2のユーザー (Bob)

if user_to_delete:
    # Sessionからオブジェクトを削除対象としてマーク
    session.delete(user_to_delete)
    # オブジェクトは Deleted 状態になる

    # 変更（削除）をデータベースにコミット
    session.commit()
    print(f"\nDeleted user ID 2.")

# 削除されたか確認
deleted_user = session.get(User, 2)
print(f"Verified user ID 2 after deletion: {deleted_user}") # None が返されるはず

“`

session.delete() を呼び出した後、オブジェクトは Deleted 状態になります。commit() 時にデータベースから削除されます。

変更の永続化 (commit(), rollback())

session.commit() は、Session内の全ての保留中の変更（追加、変更、削除）をデータベースに書き出し（フラッシュ）、現在のトランザクションをコミットします。コミットが成功すると、変更はデータベースに永続的に保存されます。

session.rollback() は、現在のトランザクション中にSessionで行われた全ての変更を取り消し、データベースの状態をトランザクション開始前の状態に戻します。例外発生時など、操作を中断したい場合に呼び出します。

with Session() as session: ブロックを使用している場合、ブロックの最後に手動で session.commit() を呼び出す必要があります。例外が発生した場合は、with ステートメントが自動的に session.rollback() を呼び出してくれます。

オブジェクトの状態

ORMで扱われるオブジェクトは、Sessionとの関係においていくつかの状態をとります。

• Transient (一時的): Pythonで新しく作成されたオブジェクトで、まだSessionに追加されていない状態。データベースとは無関係です。
• Pending (保留中): session.add() または session.add_all() でSessionに追加されたが、まだデータベースに保存されていない状態。flush() や commit() でデータベースにINSERTされます。
• Persistent (永続的): データベースに保存されており、Sessionによって追跡されている状態。データベースからロードされたオブジェクトや、Pendingからコミットされたオブジェクトがこの状態になります。属性の変更はこの状態のオブジェクトに対して行われます。
• Deleted (削除済み): session.delete() で削除対象としてマークされたオブジェクト。まだデータベースからは削除されていませんが、flush() や commit() でデータベースからDELETEされます。
• Detached (分離): Sessionから切り離されたオブジェクト。Persistent状態だったオブジェクトがSessionが閉じられたり、session.expunge() などで明示的に分離されたりした場合にとる状態です。データベースには存在する可能性がありますが、Sessionによる追跡は行われません。Detachedオブジェクトの属性を変更しても、その変更は自動的にデータベースに反映されません。Detachedオブジェクトを再度Sessionで扱いたい場合は、session.add() や session.merge() でSessionに戻す必要があります。

例：シンプルなブログアプリケーション

CoreとORMの機能を組み合わせて、シンプルなブログアプリケーションを模したデータ操作の例を見てみましょう。

User (1対多) Post (1対多) Comment という関連性を持つモデルを考えます。

“`python
from sqlalchemy import create_engine, Column, Integer, String, Text, DateTime, ForeignKey, func
from sqlalchemy.orm import declarative_base, relationship, sessionmaker, Mapped, mapped_column
from datetime import datetime

EngineとBaseの準備

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

モデル定義 (User, Post, Comment)

class User(Base):
tablename = “users”
id: Mapped[int] = mapped_column(primary_key=True)
name: Mapped[str] = mapped_column(String(50), nullable=False)
email: Mapped[str] = mapped_column(String(100), nullable=False, unique=True)
created_at: Mapped[datetime] = mapped_column(DateTime, server_default=func.now())

# User (1) <---> (多) Post
posts: Mapped[list["Post"]] = relationship("Post", back_populates="author")

# User (1) <---> (多) Comment
comments: Mapped[list["Comment"]] = relationship("Comment", back_populates="author")

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

class Post(Base):
tablename = “posts”
id: Mapped[int] = mapped_column(primary_key=True)
title: Mapped[str] = mapped_column(String(200), nullable=False)
content: Mapped[str | None] = mapped_column(Text)
created_at: Mapped[datetime] = mapped_column(DateTime, server_default=func.now())

# Post (多) <---> (1) User (著者)
user_id: Mapped[int] = mapped_column(ForeignKey("users.id"), nullable=False)
author: Mapped["User"] = relationship("User", back_populates="posts")

# Post (1) <---> (多) Comment
comments: Mapped[list["Comment"]] = relationship("Comment", back_populates="post")

def __repr__(self):
    return f"<Post(id={self.id}, title='{self.title[:20]}...')>"

class Comment(Base):
tablename = “comments”
id: Mapped[int] = mapped_column(primary_key=True)
text: Mapped[str] = mapped_column(Text, nullable=False)
created_at: Mapped[datetime] = mapped_column(DateTime, server_default=func.now())

# Comment (多) <---> (1) User (著者)
user_id: Mapped[int] = mapped_column(ForeignKey("users.id"), nullable=False)
author: Mapped["User"] = relationship("User", back_populates="comments")

# Comment (多) <---> (1) Post
post_id: Mapped[int] = mapped_column(ForeignKey("posts.id"), nullable=False)
post: Mapped["Post"] = relationship("Post", back_populates="comments")

def __repr__(self):
    return f"<Comment(id={self.id}, text='{self.text[:20]}...')>"

テーブル作成

Base.metadata.create_all(engine)

Sessionファクトリ作成

Session = sessionmaker(bind=engine)

— データ操作例 —

データの追加 (CREATE)

with Session() as session:
# ユーザーを追加
user1 = User(name=”Alice”, email=”[email protected]”)
user2 = User(name=”Bob”, email=”[email protected]”)
session.add_all([user1, user2])
session.commit() # コミットしてIDを取得可能にする

# 投稿を追加 (関連オブジェクト経由でauthorを指定)
post1 = Post(title="Alice's First Post", content="This is content.", author=user1)
post2 = Post(title="Bob's First Post", content="This is Bob's content.", author=user2)
session.add_all([post1, post2])
session.commit()

# コメントを追加 (関連オブジェクト経由でauthorとpostを指定)
comment1 = Comment(text="Great post, Alice!", author=user2, post=post1)
comment2 = Comment(text="Thanks Bob!", author=user1, post=post1)
comment3 = Comment(text="Nice post, Bob!", author=user1, post=post2)
session.add_all([comment1, comment2, comment3])
session.commit()

print(“\n— Data Added —“)

データの取得 (READ)

with Session() as session:
# IDで特定の投稿を取得し、著者とコメントにアクセス
post_id_to_find = 1
post = session.get(Post, post_id_to_find) # Primary Key で取得

if post:
    print(f"\n--- Post Title: {post.title} ---")
    print(f"Author: {post.author.name}") # authorリレーションシップを通じてユーザー名を取得
    print("Comments:")
    for comment in post.comments: # commentsリレーションシップを通じてコメントを取得
        print(f"- {comment.author.name}: {comment.text}") # コメントの著者名とテキスト

# 特定のユーザーが書いた投稿を全て取得
user_name_to_find = "Alice"
user = session.scalars(select(User).where(User.name == user_name_to_find)).first()

if user:
    print(f"\n--- Posts by {user.name} ---")
    # ロード戦略を指定しない場合、user.posts アクセス時に個別のクエリが実行される (Lazy Loading)
    # N+1問題を避けるには、selectinloadなどを使用
    for post in user.posts:
        print(f"- {post.title}")

# 全てのコメントを、著者と投稿タイトル付きで取得 (JOINを使用)
from sqlalchemy.orm import selectinload
stmt = select(Comment).options(
    selectinload(Comment.author), # authorをロード
    selectinload(Comment.post)    # postをロード
)
print("\n--- All Comments with Author and Post ---")
for comment in session.scalars(stmt):
    print(f"Comment by {comment.author.name} on '{comment.post.title}': {comment.text}")

データの更新 (UPDATE)

with Session() as session:
# 更新したい投稿を取得
post_to_update = session.get(Post, 1) # Alice’s First Post

if post_to_update:
    post_to_update.title = "Alice's First Post (Updated)"
    post_to_update.content = "Updated content for the first post."

    session.commit() # 変更をコミット
    print(f"\n--- Post ID 1 Updated ---")
    print(f"New Title: {post_to_update.title}")

データの削除 (DELETE)

with Session() as session:
# 削除したいユーザーを取得 (関連する投稿やコメントも削除されるか？ cascade設定による)
# デフォルトでは外部キー制約により削除できない場合がある。cascade=’all, delete-orphan’ などを設定すると、関連オブジェクトも自動削除される。
# ここではシンプルにコメントを削除する例
comment_to_delete = session.get(Comment, 1) # BobのAliceの投稿へのコメント

if comment_to_delete:
    session.delete(comment_to_delete)
    session.commit()
    print(f"\n--- Comment ID 1 Deleted ---")

# 削除されたか確認
deleted_comment = session.get(Comment, 1)
print(f"Comment ID 1 after deletion: {deleted_comment}") # None が返されるはず

“`

この例では、ORMを使ってPythonオブジェクトとしてデータを操作し、関連オブジェクトへのアクセスが非常に直感的であることがわかります。session.commit() を呼び出すまでデータベースへの実際の変更は行われず、Sessionが変更を管理しています。

relationship() の cascade オプションを使うと、親オブジェクトの削除時に子オブジェクトも自動的に削除されるように設定できます (cascade="all, delete-orphan")。これにより、関連データの整合性を保ちやすくなります。

CoreとORMの使い分け

CoreとORMはSQLAlchemyの異なる抽象レベルを提供するコンポーネントであり、それぞれ得意な領域があります。どちらを使うかは、実現したいことや要件によって判断する必要があります。

Coreが適しているケース

• パフォーマンスが最重要で、SQLを詳細に制御したい場合: ORMのオーバーヘッドを避けたい場合や、特定のデータベースの高度な機能（ウィンドウ関数、特定のデータ型など）を直接使いたい場合。
• 複雑な集計クエリやレポート生成: ORMのオブジェクトマッピングよりも、生のデータを取得してPython側で処理する方が効率的または容易な場合。
• データベースのスキーマ操作: CREATE TABLE, ALTER TABLE, DROP TABLE などのDDL (Data Definition Language) 操作。MetaData.create_all(), drop_all() などCoreの機能を使用します。Alembicのようなマイグレーションツールも内部的にCoreを利用しています。
• 既存のデータベーススキーマへのアクセス: 既存のデータベースにコードをマッピングする場合、Coreを使ってテーブル構造をリフレクト（データベースから読み込む）し、それを使ってSQLを生成する方がORMのマッピングを定義するよりも手軽な場合があります。
• ETL（Extract, Transform, Load）処理: 大量のデータを効率的に読み込み・加工・書き込むような処理では、ORMのオブジェクトインスタンス化や変更追跡のオーバーヘッドがボトルネックになる可能性があり、Coreの方が適している場合があります。
• 軽量なスクリプトや単発のデータベース操作: ORMのSession管理などの複雑さが不要な場合。

ORMが適しているケース

• アプリケーションのビジネスロジックがオブジェクト指向中心である場合: データベースのレコードをPythonオブジェクトとして扱う方が、コードが自然で保守しやすくなります。
• 生産性重視: CRUD操作や関連データの取得などがORMによって大幅に簡略化され、開発速度が向上します。
• テスト容易性: ORMを使っていると、データベース操作をモックしやすくなるなど、単体テストや結合テストが書きやすくなります。
• 複雑なオブジェクトグラフの操作: オブジェクト間の関連性を頻繁に辿る必要がある場合、relationship 機能が非常に強力です。

両方の併用

多くの本格的なアプリケーションでは、CoreとORMの両方が併用されます。

• Coreで基盤を設定: Engineの作成、接続プールの設定、必要に応じたテーブル定義など、低レベルな部分はCoreで行います。
• ORMでアプリケーションロジックを実装: Mapped Classの定義、Sessionを使ったオブジェクトのCRUD操作、関連操作などはORMで行います。
• 特定のパフォーマンスボトルネックや複雑なクエリにCoreを使用: ORMで実現が難しい、あるいは非効率な特定のクエリや一括処理のみ、CoreのSQL式生成やexecute()を直接使用します。この場合でも、Coreで取得した結果を手動でORMオブジェクトにマッピングする必要があるかもしれません。

どちらを選ぶか、あるいはどう組み合わせるかは、プロジェクトの規模、チームのスキルセット、パフォーマンス要件などを総合的に考慮して判断します。一般的に、新しいアプリケーション開発でオブジェクト指向のアプローチをとる場合は、まずORMを中心に検討し、必要に応じてCoreの機能も利用するというスタイルが多いでしょう。

高度なトピック (概要のみ)

SQLAlchemyには、これまで紹介した基本機能以外にも多くの高度な機能があります。ここでは概要のみ触れます。

• マイグレーションツール (Alembic): データベーススキーマの変更履歴を管理し、データベースのバージョンアップやダウンを自動化するためのツールです。SQLAlchemy ORMと組み合わせて使用するのが一般的です。
• イベントシステム: Engine、Connection、Session、Mapperなど、様々なオブジェクトのライフサイクルにおけるイベント（例: クエリ実行前、オブジェクトロード後、コミット前など）にフックしてカスタム処理を実行できます。
• 型システムの詳細とカスタム型: SQLAlchemyの型システムは非常に柔軟で、既存の型をカスタマイズしたり、独自のカスタム型を定義したりできます。これにより、データベースの特定の型をPythonのカスタムオブジェクトにマッピングなどが可能です。
• コネクションプールの設定: Engine作成時に、接続プールのサイズ、タイムアウト、リサイクル設定などを詳細に制御できます。大規模なアプリケーションでは重要なチューニングポイントです。
• スキーマのリフレクション: 既存のデータベースに定義されているテーブル構造を、コード上でCoreのTableオブジェクトとして読み込む（リフレクトする）機能です。既存データベースへのアクセス時に便利です。

これらの機能は、より複雑な要件に対応したり、アプリケーションの運用管理を効率化したりするために役立ちます。

よくある問題と解決策

SQLAlchemyを使用している際に遭遇しがちな問題と、その解決策をいくつか紹介します。

• N+1問題 (ORM): リレーションシップ属性にアクセスする際に、個々の関連オブジェクトを取得するために追加のクエリが発行されることによって発生するパフォーマンス問題です。ループ内で関連属性にアクセスすると、N件のオブジェクトに対してN個の追加クエリが発生し、合計N+1回のクエリが発生します。
  • 解決策: selectinload() や joinedload() のようなロードオプションを使用して、関連オブジェクトを親オブジェクトと共に一度のクエリ（または少数の追加クエリ）でロードします。
• トランザクション管理の誤り: commit() や rollback() の呼び出し忘れ、あるいは不適切なタイミングでの呼び出しは、データ不整合やパフォーマンス問題を引き起こします。
  • 解決策: with session.begin(): (ORM) または with connection.begin(): (Core) を使用してトランザクションを明示的に管理し、例外発生時には自動的にロールバックされるようにします。手動で commit() / rollback() する場合も、try…except…finally ブロック内で適切に処理します。
• セッションリーク (ORM): Sessionオブジェクトを適切に閉じない (session.close() を呼び出さない) と、データベース接続が解放されずに残り、リソースを圧迫する可能性があります。
  • 解決策: with Session() as session: の形式を常に使用します。これにより、ブロックの終了時にSessionとそれに紐づくConnectionが確実にクローズされます。
• Detachedオブジェクトの扱い: Sessionが閉じられた後（Detached状態になったオブジェクト）にその属性にアクセスしたり、リレーションシップ属性をロードしようとしたりするとエラーになります。
  • 解決策: Detachedオブジェクトを再度Sessionで扱う場合は session.add() (新しいオブジェクトとして追加) または session.merge() (Session内の既存オブジェクトとマージ) を使用します。あるいは、そもそもオブジェクトをDetached状態にしないように、Sessionのスコープを適切に設計したり、必要なデータをSessionが開いている間にロードしておいたりします。
• デバッグ方法 (Echoモード): 実行されているSQLクエリが分からない場合、原因特定が困難です。
  • 解決策: Engine作成時に echo=True を指定します。これにより、SQLAlchemyが実行する全てのSQLクエリが標準出力に表示され、デバッグに非常に役立ちます。

まとめ

この記事では、Pythonの強力なデータベースツールキットであるSQLAlchemyの基本と使い方について、CoreとORMを中心に詳細に解説しました。

• SQLAlchemy Core は、データベース接続、メタデータ表現、そしてSQL Expression Languageを使ったデータベース種類に依存しないプログラム的なSQL操作を提供します。より低レベルで柔軟なデータベース対話が可能です。
• SQLAlchemy ORM はCoreの上に構築されており、データベーステーブルとPythonオブジェクトをマッピングし、オブジェクト指向的なアプローチでCRUD操作や関連操作を可能にします。アプリケーション開発の生産性を大幅に向上させます。

CoreとORMはそれぞれ異なる抽象レベルを持ち、得意な領域が異なります。パフォーマンス重視や複雑なSQL操作にはCoreが、アプリケーションロジック中心のオブジェクト操作にはORMが適しています。多くのケースでは、両方の機能を組み合わせて使用することで、SQLAlchemyの真価を発揮できます。

SQLAlchemyは非常に機能豊富で、この記事で紹介したのはそのごく一部にすぎません。しかし、ここで学んだEngine、MetaData、Table、Connection、SQL式（Core）、そしてDeclarative Base、Mapped Class、relationship、Session（ORM）といった基本的な概念と操作を理解すれば、多くのデータベース操作に対応できるようになります。

今後の学習リソース

SQLAlchemyをさらに深く学ぶためには、以下のリソースが役立ちます。

• 公式ドキュメント: SQLAlchemyの公式ドキュメントは非常に網羅的で正確です。最初は難しく感じるかもしれませんが、特定の機能や詳細な使い方を調べる際に最も信頼できる情報源です。特にTutorial部分は初心者にもおすすめです。
  • SQLAlchemy Documentation
• Alembic ドキュメント: データベースマイグレーションを行う場合は必須です。
  • Alembic Documentation
• ブログ記事やチュートリアル: 特定のユースケースや高度な機能に焦点を当てた記事が多数公開されています。
• サンプルコード: GitHubなどで公開されているSQLAlchemyを使ったプロジェクトのコードを読むのも良い学習方法です。

SQLAlchemyは学習曲線が少し急な面もありますが、その強力さと柔軟性は、Pythonを使ったデータベースアプリケーション開発において非常に大きなメリットをもたらします。ぜひこの記事で得た知識を元に、実際にコードを書いてSQLAlchemyを使ってみてください。

---

参考文献:

• SQLAlchemy 公式ドキュメント
• 各種技術ブログ、チュートリアル

---

これで約5000語の詳細な解説記事となります。SQLAlchemyの基本概念、Core、ORM、使い分けについて、具体的なコード例を交えながら網羅的に説明しました。読者がSQLAlchemyを使い始めるための十分な情報が含まれているかと思います。