【入門】Python向けORM SQLAlchemyを始めよう

By /

はい、承知いたしました。Python向けORMであるSQLAlchemyの入門記事を、約5000語の詳細な説明とコード例を含めて記述します。

【入門】Python向けORM SQLAlchemyを始めよう

はじめに：ORMとは何か、そしてSQLAlchemyがなぜ必要か

現代の多くのアプリケーション開発において、データベースは不可欠な要素です。データを永続化し、管理するためにリレーショナルデータベース（RDB）が広く利用されています。しかし、プログラム（特にオブジェクト指向言語）とリレーショナルデータベースの間には、根本的な考え方の違いがあります。プログラムは「オブジェクト」としてデータを扱いますが、データベースは「テーブル」と「行」としてデータを扱います。このギャップを「インピーダンスミスマッチ」と呼びます。

このミスマッチを解消し、データベース操作をより効率的かつ直感的に行うためのツールが ORM (Object-Relational Mapper) です。ORMは、プログラム内のオブジェクトとデータベースのテーブルをマッピングする役割を果たします。これにより、開発者はSQLクエリを直接書く代わりに、オブジェクト指向のコードを使ってデータベース操作を行うことができます。

例えば、UserというPythonクラスがあれば、ORMを使うことで、user = User(name='Alice', age=30)のようにオブジェクトを生成し、session.add(user)としてデータベースに保存したり、session.query(User).filter_by(name='Alice').first()のようにオブジェクトを取得したりできます。これはSQLで INSERT INTO users (name, age) VALUES ('Alice', 30) や SELECT * FROM users WHERE name = 'Alice' LIMIT 1 と書くよりも、Pythonのコードに馴染みやすく、可読性も高まります。

Pythonにおける代表的なORMの一つが SQLAlchemy です。SQLAlchemyは非常に強力で柔軟なライブラリであり、Pythonistaの間で広く利用されています。多くのWebフレームワーク（例えばFlaskやPyramid）で標準的に、あるいは拡張機能としてSQLAlchemyがサポートされています。

SQLAlchemyの大きな特徴は、その設計思想にあります。SQLAlchemyは、単なるORMとしてだけでなく、低レベルなデータベース操作を抽象化する SQLAlchemy Core と、オブジェクトマッピングを提供する SQLAlchemy ORM の二つの主要なコンポーネントを持っています。これにより、開発者は高いレベルの抽象化を利用できる一方で、必要に応じてSQLを直接操作するような低レベルな制御も行うことができます。

SQLAlchemyを利用するメリット

  1. Pythonicなコード: データベース操作をPythonのオブジェクトや構文で行えるため、コードの可読性と保守性が向上します。
  2. データベースの抽象化: データベースの種類（PostgreSQL, MySQL, SQLite, Oracleなど）によるSQL方言の違いを吸収します。一度SQLAlchemyでコードを書けば、設定を変更するだけで異なるデータベースに接続できる可能性が高まります。
  3. SQL注入攻撃への耐性: SQLAlchemyのAPIを使うことで、パラメータのバインディングが自動的に行われるため、手書きのSQLで発生しがちなSQL注入攻撃のリスクを低減できます。
  4. 柔軟性: CoreとORMの両方を提供することで、複雑なクエリやパフォーマンスが重要な部分ではCoreを、一般的なデータ操作ではORMを使うといった使い分けが可能です。
  5. 関係の自動管理: ORMを使用する場合、テーブル間のリレーションシップ（一対多、多対多など）をPythonのオブジェクト間の関係として定義・管理できます。
  6. アクティブレコード vs データマッパー: SQLAlchemyは、特定のフレームワークに強く依存する「アクティブレコード」（例: Django ORM, Ruby on Rails ActiveRecord）スタイルではなく、より独立性の高い「データマッパー」スタイルを採用しています。これにより、ビジネスロジックとデータアクセスロジックを分離しやすくなります。

この記事では、SQLAlchemyの基本的な使い方を、CoreとORMの両面から丁寧に解説していきます。初めてSQLAlchemyに触れる方でも理解できるよう、環境構築からCRUD（Create, Read, Update, Delete）操作、そしてリレーションシップの定義までを順に追っていきます。

準備：環境構築

SQLAlchemyを使い始めるために、まずは必要なライブラリをインストールしましょう。Pythonがインストールされていることを前提とします。仮想環境での作業をおすすめします。

“`bash

仮想環境の作成（例）

python -m venv .venv
source .venv/bin/activate # macOS/Linux

.venv\Scripts\activate # Windows

SQLAlchemyのインストール

pip install SQLAlchemy

SQLite以外のデータベースを使う場合は、そのデータベースアダプターも必要です

PostgreSQLの場合: pip install psycopg2-binary

MySQLの場合: pip install PyMySQL

など

“`

この記事の例では、インストールが簡単でファイルベースのデータベースであるSQLiteを使用します。SQLiteを使う場合、Python標準ライブラリの sqlite3 が使用されるため、追加のインストールは不要です。

SQLAlchemy Coreの基本

SQLAlchemy Coreは、データベーススキーマをプログラムで定義し、SQL式を構築・実行するための低レベルAPIを提供します。これはORMの基盤ともなりますが、単独でも十分に強力です。ORMのようなオブジェクトマッピングは行いませんが、データベースとの対話の基礎を理解する上で非常に重要です。

1. エンジン (Engine)

データベースに接続し、対話するためのエントリーポイントとなるのが「エンジン」です。エンジンは接続プールを管理し、実際のデータベース接続を抽象化します。エンジンは create_engine() 関数を使って作成します。

create_engine() の引数には、データベースへの接続文字列を指定します。接続文字列のフォーマットはデータベースによって異なりますが、一般的な形式は dialect+driver://user:password@host:port/database です。

SQLiteの場合、ファイルパスまたはインメモリデータベースを指定できます。

  • インメモリSQLite: sqlite:///:memory:
  • ファイルパスSQLite: sqlite:///path/to/your/database.db

例：インメモリSQLiteデータベースに接続するエンジンを作成

“`python
from sqlalchemy import create_engine

インメモリSQLiteデータベースに接続するエンジンを作成

echo=True を設定すると、SQLAlchemyが実行する全てのSQLクエリがコンソールに出力される

engine = create_engine(‘sqlite:///:memory:’, echo=True)
“`

echo=True はデバッグ時に非常に便利です。SQLAlchemyがバックグラウンドでどのようなSQLを実行しているかを確認できます。プロダクション環境では通常 False に設定します。

2. メタデータ (MetaData)

メタデータオブジェクトは、データベーススキーマ全体、つまりどのテーブルが存在し、それらがどのような構造を持っているかといった情報を保持するコンテナです。テーブル定義は、このメタデータに関連付けられます。

“`python
from sqlalchemy import MetaData

メタデータオブジェクトを作成

metadata = MetaData()
“`

3. テーブル (Table) とカラム (Column) の定義

データベースのテーブルは Table オブジェクトで定義し、テーブルのカラムは Column オブジェクトで定義します。Table オブジェクトを作成する際には、テーブル名、関連付ける MetaData オブジェクト、そしてそのテーブルに含まれる Column オブジェクト群を指定します。

Column オブジェクトを作成する際には、カラム名、データ型、そして各種制約（主キー、外部キー、NULL許容など）を指定します。

SQLAlchemyのデータ型は、データベース固有の型を抽象化したものです。一般的な型には Integer, String, Text, Boolean, Date, DateTime, Numeric などがあります。

例：users テーブルを定義する

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

メタデータオブジェクトを作成

metadata = MetaData()

‘users’ テーブルを定義

users_table = Table(
‘users’, # テーブル名
metadata, # メタデータオブジェクト
Column(‘id’, Integer, primary_key=True), # idカラム: 整数型, 主キー
Column(‘name’, String(50), nullable=False), # nameカラム: 文字列型(最大50文字), NOT NULL制約
Column(‘age’, Integer) # ageカラム: 整数型, NULL許容（デフォルト）
)

テーブル定義は、metadataオブジェクトに関連付けられる

metadata.tables を確認すると、定義したテーブルが含まれているのがわかる

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

別のテーブルを定義し、外部キーを設定する例：products テーブルと orders テーブル

“`python
from sqlalchemy import ForeignKey, Numeric

‘products’ テーブルを定義

products_table = Table(
‘products’,
metadata,
Column(‘id’, Integer, primary_key=True),
Column(‘name’, String(100), nullable=False),
Column(‘price’, Numeric(10, 2)) # 価格カラム: 数値型 (合計10桁, 小数点以下2桁)
)

‘orders’ テーブルを定義 (ユーザーと製品への外部キーを持つ)

orders_table = Table(
‘orders’,
metadata,
Column(‘id’, Integer, primary_key=True),
Column(‘user_id’, Integer, ForeignKey(‘users.id’), nullable=False), # user_idカラム: 整数型, usersテーブルのidカラムへの外部キー
Column(‘product_id’, Integer, ForeignKey(‘products.id’), nullable=False), # product_idカラム: 整数型, productsテーブルのidカラムへの外部キー
Column(‘quantity’, Integer, default=1), # quantityカラム: 整数型, デフォルト値1
Column(‘order_date’, DateTime) # order_dateカラム: 日時型
)

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

4. テーブルの作成 (DDL操作)

MetaData オブジェクトに定義されたテーブル構造に基づいて、実際のデータベース内にテーブルを作成するには、metadata.create_all() メソッドを使用します。このメソッドは、関連付けられたエンジンを通じてデータベースに接続し、存在しないテーブルを作成します。

“`python

定義したテーブルをデータベースに作成

checkfirst=True は、テーブルが既に存在するかどうかを確認してから作成する

metadata.create_all(engine, checkfirst=True)

print(“Tables created successfully.”)

echo=True の場合、CREATE TABLE 文がコンソールに出力されるはずです。

“`

5. データの挿入 (Insert)

Core APIを使ってデータを挿入するには、insert() 関数で挿入文を構築し、エンジンの接続を使って実行します。

insert() 関数は、どのテーブルに挿入するかを指定し、.values() メソッドで挿入するデータの辞書またはリストを指定します。

例：users テーブルにデータを挿入する

“`python
from sqlalchemy import insert

ユーザーデータを挿入するための insert 文を構築

insert_users_stmt = insert(users_table).values(
name=’Alice’,
age=30
)

エンジンから接続を取得し、文を実行

with statementを使うことで、接続が自動的に閉じられる

with engine.connect() as connection:
result = connection.execute(insert_users_stmt)
# 主キーが自動生成される場合、挿入された行の主キーを取得できる (SQLiteの場合)
inserted_id = result.inserted_primary_key[0]
print(f”Inserted user with id: {inserted_id}”)

# 複数の行をまとめて挿入することも可能
insert_multiple_users_stmt = insert(users_table).values([
    {'name': 'Bob', 'age': 25},
    {'name': 'Charlie', 'age': 35},
    {'name': 'Alice', 'age': 31} # 同じ名前のユーザーを追加してみる
])
result_multiple = connection.execute(insert_multiple_users_stmt)
print(f"Inserted {result_multiple.rowcount} more users.")

# 注意: デフォルトでは、Coreの操作は自動コミットされない場合があります。
# 特にトランザクションを意識しない場合は、connection.commit() を明示的に呼び出すか、
# 後述するbegin()を使ってトランザクション内で実行するのが安全です。
# SQLiteではデフォルトで自動コミットが有効なことが多いですが、他のDBでは異なります。
# 明示的にコミット/ロールバックを管理する方が一般的です。

“`

トランザクション内で実行する場合：

“`python
with engine.begin() as connection: # begin()を使うと、withブロック内でトランザクションが開始され、成功すればコミット、例外発生でロールバックされる
insert_users_stmt = insert(users_table).values([
{‘name’: ‘David’, ‘age’: 40},
{‘name’: ‘Eve’, ‘age’: 22}
])
result = connection.execute(insert_users_stmt)
print(f”Inserted {result.rowcount} users within a transaction.”)

with ブロックを抜ける際に自動的にコミットされる

“`

6. データの取得 (Select)

Core APIでデータを取得するには、select() 関数でSELECT文を構築し、実行します。

select() 関数には、取得したいカラムやテーブルを指定します。結果は ResultProxy オブジェクトとして返され、そこから行データをフェッチできます。

例：users テーブルからデータを取得する

“`python
from sqlalchemy import select

usersテーブルから全てのカラムを選択するSELECT文を構築

select_all_users_stmt = select(users_table)

with engine.connect() as connection:
# SELECT文を実行
result = connection.execute(select_all_users_stmt)

# 結果を全て取得 (リストで返される)
users_list = result.fetchall()
print("\n--- All users (fetchall) ---")
for row in users_list:
    # 行はタプルまたは Row オブジェクトとして取得される
    # カラム名でアクセスすることもできる
    print(f"ID: {row.id}, Name: {row.name}, Age: {row.age}")
    # またはインデックスでアクセス
    # print(f"ID: {row[0]}, Name: {row[1]}, Age: {row[2]}")

# 結果を一つずつ取得 (イテレーターとして扱える)
print("\n--- All users (iteration) ---")
result = connection.execute(select_all_users_stmt) # 結果は一度フェッチすると空になるので再度実行
for row in result:
    print(f"ID: {row.id}, Name: {row.name}, Age: {row.age}")

# 最初の1行だけを取得
print("\n--- First user ---")
select_first_user_stmt = select(users_table).limit(1)
result_first = connection.execute(select_first_user_stmt)
first_user = result_first.fetchone()
if first_user:
    print(f"ID: {first_user.id}, Name: {first_user.name}, Age: {first_user.age}")

# 結果を0行か1行として取得 (行数が1行でない場合はエラーまたはNone)
# one(), one_or_none()
print("\n--- User with id 2 (one_or_none) ---")
select_user_by_id_stmt = select(users_table).where(users_table.c.id == 2) # users_table.c でカラムにアクセス
user_id_2 = connection.execute(select_user_by_id_stmt).one_or_none()
if user_id_2:
    print(f"ID: {user_id_2.id}, Name: {user_id_2.name}, Age: {user_id_2.age}")
else:
    print("User with id 2 not found.")

# 特定のカラムだけを選択
print("\n--- Only names and ages ---")
select_names_ages_stmt = select(users_table.c.name, users_table.c.age)
for row in connection.execute(select_names_ages_stmt):
    print(f"Name: {row.name}, Age: {row.age}") # row.name, row[0] のどちらでもアクセス可能

“`

条件指定 (WHERE)

select() に .where() メソッドをチェーンすることで、SELECT文にWHERE句を追加できます。条件式は、テーブルのカラムオブジェクトに対して比較演算子（==, !=, >, <, >=, <=）やSQL関数（like(), in_(), is_(), is_not()）を使って構築します。複数の条件は and_(), or_(), not_() 関数（sqlalchemy モジュールからインポート）で組み合わせます。

“`python
from sqlalchemy import and_, or_, not_

with engine.connect() as connection:
# 年齢が30歳以上のユーザーを選択
select_age_ge_30_stmt = select(users_table).where(users_table.c.age >= 30)
print(“\n— Users with age >= 30 —“)
for row in connection.execute(select_age_ge_30_stmt):
print(f”ID: {row.id}, Name: {row.name}, Age: {row.age}”)

# 名前に 'li' を含み、かつ年齢が30歳以上のユーザーを選択
select_filtered_stmt = select(users_table).where(
    and_(
        users_table.c.name.like('%li%'),
        users_table.c.age >= 30
    )
)
print("\n--- Users with name like '%li%' AND age >= 30 ---")
for row in connection.execute(select_filtered_stmt):
    print(f"ID: {row.id}, Name: {row.name}, Age: {row.age}")

# IDが 1 または 3 のユーザーを選択
select_in_stmt = select(users_table).where(users_table.c.id.in_([1, 3]))
print("\n--- Users with id 1 or 3 ---")
for row in connection.execute(select_in_stmt):
    print(f"ID: {row.id}, Name: {row.name}, Age: {row.age}")

# 年齢がNULLのユーザーを選択 (SQLAlchemyでは None がNULLに変換される)
select_age_is_null_stmt = select(users_table).where(users_table.c.age.is_(None))
print("\n--- Users with age IS NULL ---")
for row in connection.execute(select_age_is_null_stmt):
     print(f"ID: {row.id}, Name: {row.name}, Age: {row.age}")

“`

並べ替え (ORDER BY)

.order_by() メソッドで結果の並べ替えを指定します。昇順はカラムオブジェクトを直接指定、降順は desc() 関数（sqlalchemy モジュールからインポート）を使います。

“`python
from sqlalchemy import desc

with engine.connect() as connection:
# 年齢で降順に並べ替え
select_ordered_stmt = select(users_table).order_by(desc(users_table.c.age))
print(“\n— Users ordered by age (desc) —“)
for row in connection.execute(select_ordered_stmt):
print(f”ID: {row.id}, Name: {row.name}, Age: {row.age}”)

# 名前で昇順、その後年齢で降順に並べ替え
select_multi_ordered_stmt = select(users_table).order_by(users_table.c.name, desc(users_table.c.age))
print("\n--- Users ordered by name (asc), then age (desc) ---")
for row in connection.execute(select_multi_ordered_stmt):
    print(f"ID: {row.id}, Name: {row.name}, Age: {row.age}")

“`

制限とオフセット (LIMIT, OFFSET)

.limit() と .offset() メソッドで行数制限とオフセットを指定します。

“`python
with engine.connect() as connection:
# 最初の2人のユーザーを取得
select_limited_stmt = select(users_table).limit(2)
print(“\n— First 2 users —“)
for row in connection.execute(select_limited_stmt):
print(f”ID: {row.id}, Name: {row.name}, Age: {row.age}”)

# 最初の2人をスキップして、次の2人のユーザーを取得
select_offset_limited_stmt = select(users_table).offset(2).limit(2)
print("\n--- Next 2 users (offset 2, limit 2) ---")
for row in connection.execute(select_offset_limited_stmt):
    print(f"ID: {row.id}, Name: {row.name}, Age: {row.age}")

“`

結合 (JOIN)

select() に .join() メソッドをチェーンすることで、複数のテーブルを結合できます。SQLAlchemyは外部キー制約が定義されていれば自動的に結合条件を推測できますが、明示的に結合条件を指定することも可能です。

先に products テーブルと orders テーブルにダミーデータを挿入しておきましょう。

“`python
from sqlalchemy import insert, DateTime
import datetime

with engine.begin() as connection:
# products テーブルにデータを挿入
insert_products_stmt = insert(products_table).values([
{‘name’: ‘Laptop’, ‘price’: 1200.00},
{‘name’: ‘Keyboard’, ‘price’: 75.50},
{‘name’: ‘Mouse’, ‘price’: 25.00}
])
connection.execute(insert_products_stmt)
print(f”Inserted {insert_products_stmt.rowcount} products.”)

# orders テーブルにデータを挿入
# ユーザーID 1 (Alice) と 2 (Bob) が製品を購入したとする
# ユーザーIDは、usersテーブルに挿入された順に1, 2, 3, 4, 5 と仮定
insert_orders_stmt = insert(orders_table).values([
    {'user_id': 1, 'product_id': 1, 'quantity': 1, 'order_date': datetime.datetime.now()}, # Alice ordered Laptop
    {'user_id': 1, 'product_id': 2, 'quantity': 2, 'order_date': datetime.datetime.now()}, # Alice ordered Keyboards
    {'user_id': 2, 'product_id': 3, 'quantity': 1, 'order_date': datetime.datetime.now()}, # Bob ordered Mouse
    {'user_id': 3, 'product_id': 1, 'quantity': 1, 'order_date': datetime.datetime.now()}  # Charlie ordered Laptop
])
connection.execute(insert_orders_stmt)
print(f"Inserted {insert_orders_stmt.rowcount} orders.")

“`

結合クエリの例：ユーザー名と注文した製品名を取得する

“`python
with engine.connect() as connection:
# usersテーブルとordersテーブルを結合し、さらにproductsテーブルとordersテーブルを結合
# 取得したいカラムを指定（users.name, products.name, orders.quantity）
select_joined_stmt = select(
users_table.c.name.label(‘user_name’), # カラムに別名を付ける .label()
products_table.c.name.label(‘product_name’),
orders_table.c.quantity
).select_from(users_table.join(orders_table).join(products_table)) # 結合の順序と方法を指定

print("\n--- Orders with user and product names (JOIN) ---")
for row in connection.execute(select_joined_stmt):
    print(f"User: {row.user_name}, Product: {row.product_name}, Quantity: {row.quantity}")
    # 別名でアクセスできます

“`

7. データの更新 (Update)

Core APIでデータを更新するには、update() 関数で更新文を構築し、実行します。

update() 関数は、どのテーブルを更新するかを指定し、.values() メソッドで更新するデータの辞書を指定し、.where() メソッドで更新対象の行を絞り込みます。

例：ユーザーの年齢を更新する

“`python
from sqlalchemy import update

IDが1のユーザーの年齢を32に更新する update 文を構築

update_stmt = update(users_table).where(users_table.c.id == 1).values(age=32)

with engine.connect() as connection:
result = connection.execute(update_stmt)
print(f”\nUpdated {result.rowcount} user(s).”) # 更新された行数を確認

# 更新後のデータを確認
updated_user = connection.execute(select(users_table).where(users_table.c.id == 1)).one()
print(f"Updated user 1: ID: {updated_user.id}, Name: {updated_user.name}, Age: {updated_user.age}")

# 複数のユーザーの年齢を一括で増やす
update_multiple_stmt = update(users_table).values(age=users_table.c.age + 1).where(users_table.c.age < 35)
result_multiple = connection.execute(update_multiple_stmt)
print(f"Incremented age for {result_multiple.rowcount} user(s) under 35.")

# 確認
print("\n--- Users after batch update ---")
for row in connection.execute(select(users_table).order_by(users_table.c.id)):
     print(f"ID: {row.id}, Name: {row.name}, Age: {row.age}")

“`

8. データの削除 (Delete)

Core APIでデータを削除するには、delete() 関数で削除文を構築し、実行します。

delete() 関数は、どのテーブルから削除するかを指定し、.where() メソッドで削除対象の行を絞り込みます。

例：IDが5のユーザーを削除する

“`python
from sqlalchemy import delete

IDが5のユーザーを削除する delete 文を構築

delete_stmt = delete(users_table).where(users_table.c.id == 5)

with engine.connect() as connection:
result = connection.execute(delete_stmt)
print(f”\nDeleted {result.rowcount} user(s).”) # 削除された行数を確認

# 削除後のデータを確認 (ID 5 のユーザーは存在しないはず)
remaining_users = connection.execute(select(users_table).order_by(users_table.c.id)).fetchall()
print("\n--- Remaining users ---")
for row in remaining_users:
    print(f"ID: {row.id}, Name: {row.name}, Age: {row.age}")

# 年齢が35歳以上の全てのユーザーを削除
delete_age_ge_35_stmt = delete(users_table).where(users_table.c.age >= 35)
result_multiple = connection.execute(delete_age_ge_35_stmt)
print(f"Deleted {result_multiple.rowcount} user(s) with age >= 35.")

# 最終確認
print("\n--- Users after age >= 35 deletion ---")
for row in connection.execute(select(users_table).order_by(users_table.c.id)):
     print(f"ID: {row.id}, Name: {row.name}, Age: {row.age}")

“`

9. テーブルの削除 (DDL操作)

MetaData オブジェクトに定義されたテーブルをデータベースから削除するには、metadata.drop_all() メソッドを使用します。

“`python

定義したテーブルをデータベースから削除

metadata.drop_all(engine)

print(“\nTables dropped successfully.”)

echo=True の場合、DROP TABLE 文がコンソールに出力されるはずです。

“`

SQLAlchemy Coreは、このようにSQLの各操作（DDL, DML）をPythonのオブジェクトと関数で表現し、データベースに依存しない形で実行できる強力なツールです。生のSQLを文字列として扱うよりも安全で、コードの構造化にも役立ちます。

しかし、Core APIは基本的にリレーショナルな観点（テーブル、行、カラム）でデータベースを扱います。プログラムがオブジェクト指向で設計されている場合、データベースから取得した行データをオブジェクトに手動でマッピングしたり、オブジェクトの変更をデータベース操作に変換したりするのは手間がかかります。ここでSQLAlchemy ORMの出番となります。

SQLAlchemy ORMの基本

SQLAlchemy ORMは、Pythonクラスのインスタンスをデータベースの行に、クラスの属性をテーブルのカラムにマッピングします。これにより、データベース操作をPythonオブジェクトの操作として行うことができます。Core APIの上に構築されており、Coreの機能を活用しながらオブジェクト指向的なインターフェースを提供します。

1. ORMの構成要素

  • Engine: Coreと同様、データベースへの接続を管理します。
  • Declarative Base: ORMモデルクラスの基底クラスとなります。これにより、テーブル定義とクラス定義を同時に行うことができます。
  • Model Class: データベーステーブルに対応するPythonクラスです。Declarative Baseを継承し、クラス属性としてテーブルのカラムやリレーションシップを定義します。
  • Session: ORM操作の中心となるオブジェクトです。データベースとの「対話」や「トランザクション」を表します。オブジェクトのロード、変更追跡、コミット、ロールバックなどを管理します。

2. Declarative Base の設定

ORMモデルクラスを定義するために、まずはDeclarative Baseクラスを作成します。

“`python
from sqlalchemy.ext.declarative import declarative_base

Declarative Base クラスを作成

Base = declarative_base()
``
**注:** SQLAlchemy 2.0以降ではdeclarative_baseは非推奨となり、SQLAlchemy.orm.declarative_baseを使うか、またはMappedAsDataclassを使用するのが推奨されています。ここでは入門として広く使われてきたsqlalchemy.ext.declarative` の方法を先に解説し、後で新しいスタイルにも触れます。

3. モデルクラスの定義

データベーステーブルに対応するPythonクラスを定義します。これらのクラスは上で作成した Base を継承します。

クラス属性として __tablename__ で対応するテーブル名を指定します。
カラムは Column オブジェクトとして定義します。Coreで使った Column と同じですが、ORMではモデルクラスの属性として定義することで、その属性がデータベースのカラムにマッピングされることを示します。

例：User モデルクラスの定義

“`python
from sqlalchemy import Column, Integer, String

Coreの Column をそのまま使う

Base が既に定義されているとする (Declarative Base の設定参照)

class User(Base):
tablename = ‘users’ # 対応するテーブル名

id = Column(Integer, primary_key=True) # id属性 -> idカラム (Integer型, 主キー)
name = Column(String(50), nullable=False) # name属性 -> nameカラム (String(50)型, NOT NULL)
age = Column(Integer) # age属性 -> ageカラム (Integer型)

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

“`

Product および Order モデルクラスの定義（外部キーを含む）

“`python
from sqlalchemy import ForeignKey, Numeric, DateTime
from sqlalchemy.orm import relationship # リレーションシップ定義に使う

class Product(Base):
tablename = ‘products’

id = Column(Integer, primary_key=True)
name = Column(String(100), nullable=False)
price = Column(Numeric(10, 2))

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

class Order(Base):
tablename = ‘orders’

id = Column(Integer, primary_key=True)
user_id = Column(Integer, ForeignKey('users.id'), nullable=False) # 外部キーとして定義
product_id = Column(Integer, ForeignKey('products.id'), nullable=False)
quantity = Column(Integer, default=1)
order_date = Column(DateTime)

# ORMのリレーションシップを定義
# Order オブジェクトから関連する User オブジェクトにアクセスできるようにする
user = relationship("User") # "User" は関連するモデルクラス名

# Order オブジェクトから関連する Product オブジェクトにアクセスできるようにする
product = relationship("Product") # "Product" は関連するモデルクラス名


def __repr__(self):
     # user と product の情報も含める
     return f"<Order(id={self.id}, user_id={self.user_id}, product_id={self.product_id}, quantity={self.quantity}, order_date={self.order_date})>"

``
ここで注目すべきはuser = relationship(“User”)とproduct = relationship(“Product”)の定義です。これらはデータベースレベルの外部キーではなく、**ORMレベルのリレーションシップ** を定義しています。これにより、例えばorderというOrderオブジェクトがあれば、order.userとしてその注文を行ったUserオブジェクトに、order.productとして注文されたProduct` オブジェクトにアクセスできるようになります。SQLAlchemyが裏側で適切なJOINクエリなどを生成してくれます。

4. エンジンとテーブルの作成 (ORMの場合)

Coreと同様にエンジンを作成し、Declarative Baseのメタデータを使ってテーブルを作成します。Declarative Baseは内部的にCoreの MetaData オブジェクトを持っており、それにモデルクラスで定義したテーブル情報を登録します。

“`python

Coreの時と同じようにエンジンを作成

echo=True は引き続きデバッグに便利

engine = create_engine(‘sqlite:///:memory:’, echo=True)

Base クラスのメタデータオブジェクトを使ってテーブルを作成

Base.metadata.create_all(engine)

print(“ORM tables created successfully.”)

echo=True なら CREATE TABLE 文が出力される

“`

5. セッション (Session)

ORMのほとんどの操作は「セッション」を通じて行われます。セッションは、データベースとの対話やトランザクションの状態を管理します。オブジェクトのロード、変更、追加、削除はまずセッション上で行われ、session.commit() を呼び出したときに実際のデータベース操作として反映されます。

セッションを作成するには、まず sessionmaker を使ってセッションクラスを生成します。このセッションクラスは、どのエンジンを使用するかなどの設定を持ちます。

“`python
from sqlalchemy.orm import sessionmaker

セッションクラスを作成 (bind=engine でどのエンジンを使うか指定)

Session = sessionmaker(bind=engine)

セッションインスタンスを作成

session = Session()
“`

セッションはスレッドセーフではないため、通常はリクエストごとや処理ごとに新しいセッションインスタンスを作成し、使い終わったら必ず session.close() するか、 with 文を使って管理します。

“`python

with 文を使ったセッション管理 (推奨)

with Session() as session:
# このブロック内でセッションを使ったデータベース操作を行う
pass # 処理が終わると自動的にセッションが閉じられる
``
さらに、トランザクション管理を同時に行うにはsession.begin()` を使います。

“`python
with Session() as session:
with session.begin(): # トランザクション開始
# このブロック内の操作は一つのトランザクションとして扱われる
# 例外が発生しなければコミットされ、発生すればロールバックされる
pass
# begin() ブロックを抜けるとトランザクションが完了 (コミットまたはロールバック)

session ブロックを抜けるとセッションが閉じられる

``with session.begin():はsession.add(…); session.commit()とsession.rollback()` を適切に組み合わせた便利なイディオムです。特に理由がなければこれを使用するのが良いでしょう。

6. データの挿入 (Create)

ORMでデータを挿入するには、対応するモデルクラスのインスタンスを作成し、セッションに add() します。その後、session.commit() を呼び出すことでデータベースに永続化されます。

“`python
with Session() as session:
# ユーザーオブジェクトを作成
new_user = User(name=’Alice ORM’, age=30)
another_user = User(name=’Bob ORM’, age=25)

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

# 複数のオブジェクトをまとめて追加
more_users = [
    User(name='Charlie ORM', age=35),
    User(name='David ORM', age=40)
]
session.add_all(more_users)

# コミットしてデータベースに保存
session.commit()

print("\n--- ORM Users added ---")
# オブジェクトはコミット後に id が設定される
print(new_user)
print(another_user)
print(more_users)

# コミット後にオブジェクトはセッションに紐づいたままになります
# 必要であれば session.refresh(obj) や session.expunge(obj) を使うこともあります

``session.add()は、実際にはすぐにINSERT文を実行するわけではありません。追加されたオブジェクトはセッションによって「保留中 (pending)」の状態として追跡され、session.commit()またはsession.flush()が呼び出されたときに初めてデータベース操作が行われます (flush` はコミットせずにデータベースと同期します)。

7. データの取得 (Read / Query)

ORMでデータを取得するには、セッションの query() メソッドを使用します。query() には取得したいモデルクラスを指定します。その後、Core APIと同様に .filter(), .order_by(), .limit() などのメソッドをチェーンしてクエリを構築します。

“`python
with Session() as session:
# 全てのユーザーを取得
users = session.query(User).all()
print(“\n— All ORM users (query().all()) —“)
for user in users:
print(user) # repr メソッドで定義した形式で表示される

# IDが1のユーザーを取得 (primary key で取得する場合は get() が便利)
user_by_id = session.query(User).get(1) # get() は主キーを指定
# user_by_id = session.query(User).filter(User.id == 1).first() # filter() + first() でも同じ
print(f"\n--- ORM user with id 1 (query().get(1)) ---")
if user_by_id:
    print(user_by_id)

# 年齢が30歳以上のユーザーを取得
users_age_ge_30 = session.query(User).filter(User.age >= 30).all()
print("\n--- ORM users with age >= 30 (query().filter()) ---")
for user in users_age_ge_30:
    print(user)

# 名前に 'Alice' を含むユーザーを取得
users_named_alice = session.query(User).filter(User.name.like('%Alice%')).all()
print("\n--- ORM users with name like '%Alice%' (query().filter()) ---")
for user in users_named_alice:
    print(user)

# 年齢で降順に並べ替えて、最初の2人を取得
users_ordered_limited = session.query(User).order_by(User.age.desc()).limit(2).all()
print("\n--- Top 2 oldest ORM users (order_by().limit()) ---")
for user in users_ordered_limited:
    print(user)

# 複数の条件を指定 (AND)
users_filtered_and = session.query(User).filter(User.name.like('%M%'), User.age > 20).all()
print("\n--- ORM users with name like '%M%' AND age > 20 ---")
for user in users_filtered_and:
    print(user)

# 複数の条件を指定 (OR) - or_() 関数を使用
from sqlalchemy import or_
users_filtered_or = session.query(User).filter(or_(User.name == 'Alice ORM', User.name == 'Bob ORM')).all()
print("\n--- ORM users named Alice ORM or Bob ORM (or_()) ---")
for user in users_filtered_or:
    print(user)

``
ORMのquery()は、Coreのselect()と似たようなメソッド (filter,order_by,limit,offset) を提供しますが、引数や使い方に違いがあります。例えば、filter()にはCoreのようにテーブルのカラムオブジェクトではなく、モデルクラスの属性（例:User.age）を使います。比較演算子も直接使用します（例:User.age >= 30`）。

リレーションシップを使ったクエリとアクセス

ORMの大きな利点は、リレーションシップを簡単に扱えることです。

まず、ProductとOrderのダミーデータをORMを使って追加してみましょう。

“`python
import datetime

with Session() as session:
with session.begin():
# 製品データを追加
products = [
Product(name=’Laptop ORM’, price=1500.00),
Product(name=’Tablet ORM’, price=500.00),
Product(name=’Phone ORM’, price=800.00)
]
session.add_all(products)

    # ユーザーを取得 (先ほど追加したユーザーを再利用)
    # Alice ORM の ID が 1 だと仮定 (もし変わるならクエリで取得し直す)
    alice_orm = session.query(User).filter_by(name='Alice ORM').first()
    bob_orm = session.query(User).filter_by(name='Bob ORM').first()
    charlie_orm = session.query(User).filter_by(name='Charlie ORM').first()

    if alice_orm and products:
        # 注文を追加 (リレーションシップを使ってオブジェクトで指定可能)
        order1 = Order(user=alice_orm, product=products[0], quantity=1, order_date=datetime.datetime.now()) # Alice -> Laptop
        order2 = Order(user=alice_orm, product=products[1], quantity=2, order_date=datetime.datetime.now()) # Alice -> Tablets
        session.add_all([order1, order2])

    if bob_orm and products:
        order3 = Order(user=bob_orm, product=products[2], quantity=1, order_date=datetime.datetime.now()) # Bob -> Phone
        session.add(order3)

    if charlie_orm and products:
        order4 = Order(user=charlie_orm, product=products[0], quantity=1, order_date=datetime.datetime.now()) # Charlie -> Laptop
        session.add(order4)

 # begin() ブロックを抜けるとコミットされる

print(“\n— ORM Products and Orders added —“)
“`

リレーションシップを使ったデータアクセス：

“`python
with Session() as session:
# 全ての注文を取得し、関連するユーザーと製品にアクセス
orders = session.query(Order).all()
print(“\n— All ORM orders with related objects —“)
for order in orders:
# Order オブジェクトから user と product オブジェクトにアクセス
# 最初のアクセス時にデータベースからロードされる (Lazy Loading)
print(f”Order ID: {order.id}, User: {order.user.name}, Product: {order.product.name}, Quantity: {order.quantity}”)

# 特定のユーザーの注文を取得
alice_orm = session.query(User).filter_by(name='Alice ORM').first()
if alice_orm:
    # User モデルにも Order とのリレーションシップを定義すると、より簡単にアクセスできる
    # (これはまだ定義していないが、後述)
    # 今は Order からユーザーをフィルタリングする
    alice_orders = session.query(Order).filter_by(user=alice_orm).all() # filter_by にオブジェクトを指定できる
    print(f"\n--- Orders by {alice_orm.name} ---")
    for order in alice_orders:
         print(f"  Product: {order.product.name}, Quantity: {order.quantity}")

``
上記の例でorder.user.nameやorder.product.name` にアクセスした際に、SQLAlchemyは必要に応じて（デフォルトでは遅延ロード – Lazy Loading）データベースにクエリを発行して関連オブジェクトを取得します。

8. リレーションシップ (Relationships) の定義

上記の例で、Order モデルから User や Product へアクセスできましたが、User モデルからそのユーザーが行った Order のリストにアクセスするには、User モデル側にもリレーションシップを定義する必要があります。relationship() 関数は、関連するクラスの名前と、オプションでバックリファレンス（反対方向のリレーションシップ）を指定できます。

User モデルに orders リレーションシップを追加します。

“`python
from sqlalchemy.orm import relationship

class User(Base):
tablename = ‘users’

id = Column(Integer, primary_key=True)
name = Column(String(50), nullable=False)
age = Column(Integer)

# この User に紐づく Order オブジェクトのリスト
# uselist=True がデフォルトなのでリストになる
# back_populates='user' は Order モデルの 'user' リレーションシップと双方向であることを示す
orders = relationship("Order", back_populates="user")

def __repr__(self):
    # リレーションシップを表示に含める場合は、ロードされているか確認しないと Lazy Loading でクエリが増える可能性
    # または、repr でリレーションシップを含めない方が安全
    return f"<User(id={self.id}, name='{self.name}', age={self.age})>"

Order モデルの user リレーションシップも back_populates を追加して双方向にする

class Order(Base):
tablename = ‘orders’

id = Column(Integer, primary_key=True)
user_id = Column(Integer, ForeignKey('users.id'), nullable=False)
product_id = Column(Integer, ForeignKey('products.id'), nullable=False)
quantity = Column(Integer, default=1)
order_date = Column(DateTime)

# back_populates='orders' は User モデルの 'orders' リレーションシップと双方向であることを示す
user = relationship("User", back_populates="orders")
product = relationship("Product") # Product 側には Order リレーションシップは不要なら片方向でも良い

def __repr__(self):
     # user や product を repr に含めると Lazy Loading でクエリが発生するので注意
     return f"<Order(id={self.id}, user_id={self.user_id}, product_id={self.product_id}, quantity={self.quantity})>"

ORMモデルを再定義したので、もう一度テーブルを作成し直す (インメモリDBなので可能)

実際のアプリケーションではマイグレーションツールを使います

Base.metadata.drop_all(engine) # 必要なら一度削除

Base.metadata.create_all(engine)

ダミーデータの挿入も必要になる

``
**注意:** ここでモデルクラスを再定義しましたが、create_allはテーブルが存在しない場合にのみ作成するため、インメモリDBをクリアするかdrop_all` をしてから実行する必要があります。実際のアプリケーションでは Alembic のようなマイグレーションツールを使用してスキーマ変更を管理します。

リレーションシップを使ったクエリとアクセス（User -> Orders）：

“`python

(上記で User と Order モデルに back_populates を追加し、必要ならテーブル作成とデータ挿入をやり直した後)

with Session() as session:
# Alice ORM ユーザーを取得
alice_orm = session.query(User).filter_by(name=’Alice ORM’).first()

if alice_orm:
    print(f"\n--- Orders placed by {alice_orm.name} (ORM relationship access) ---")
    # alice_orm.orders にアクセスすると、関連する Order オブジェクトのリストがロードされる
    # これもデフォルトでは Lazy Loading
    if alice_orm.orders:
        for order in alice_orm.orders:
             print(f"  Order ID: {order.id}, Product: {order.product.name}, Quantity: {order.quantity}") # order.product も Lazy Loading
    else:
        print("  No orders found.")

``
このように、relationship` を定義することで、データベースの結合を意識せずにオブジェクト間の関係をたどることができます。

ロード戦略 (Loading Strategies)

デフォルトのLazy Loadingは、関連オブジェクトにアクセスしたときに初めてデータベースにクエリを実行します。これは必要なときだけデータをロードするため効率的ですが、多くのオブジェクトの関連オブジェクトにアクセスする場合、N+1問題（親オブジェクトを1つ取得するクエリ + それぞれの子オブジェクトを取得するN個のクエリ）を引き起こし、パフォーマンスが悪化することがあります。

これを回避するために、SQLAlchemy ORMは様々なロード戦略を提供しています。代表的なものに Eager Loading があります。

  • Joined Loading (joinedload): 親オブジェクトを取得するクエリで、関連する子オブジェクトもJOINを使って同時にロードします。ほとんどの場合、これが最も効率的なEager Loading方法です。
  • Selectin Loading (selectinload): 親オブジェクトを取得した後、別のクエリで、取得した全ての親オブジェクトに関連する子オブジェクトをまとめてロードします（WHERE child.parent_id IN (...) のようなクエリ）。一対多の関係で、多数の親オブジェクトをロードする場合にJoined Loadingより効率的なことがあります。

例：Joined Loading を使ってユーザーとその注文を一度にロードする

“`python
from sqlalchemy.orm import joinedload

with Session() as session:
# ユーザーとその関連注文を Joined Loading で一度にロード
users_with_orders = session.query(User).options(joinedload(User.orders)).all() # options() と joinedload() を使う

print("\n--- Users with orders (Joined Loading) ---")
for user in users_with_orders:
    print(f"User: {user.name}")
    # ここで user.orders にアクセスしても、既にロードされているので追加のクエリは発生しない
    if user.orders:
        for order in user.orders:
            # 注意: order.product はまだ Lazy Loading です
            print(f"  Order ID: {order.id}, Product ID: {order.product_id}, Quantity: {order.quantity}")
            # order.product.name にアクセスすると、ここで初めて Product がロードされる
            # print(f"  Order ID: {order.id}, Product: {order.product.name}, Quantity: {order.quantity}")
    else:
        print("  No orders.")

# Order から User と Product を同時に Joined Loading でロードする例
orders_with_user_product = session.query(Order).options(joinedload(Order.user), joinedload(Order.product)).all()
print("\n--- Orders with user and product (Multi Joined Loading) ---")
for order in orders_with_user_product:
    # ここで order.user と order.product にアクセスしても追加のクエリは発生しない
    print(f"Order ID: {order.id}, User: {order.user.name}, Product: {order.product.name}, Quantity: {order.quantity}")

``echo=True` を設定している場合、Joined Loadingを使うと、Lazy Loadingの場合よりも少ないクエリで同じデータを取得できることがわかるはずです。パフォーマンスが重要な場面では、適切なロード戦略を選択することが重要です。

9. データの更新 (Update)

ORMでデータを更新するには、まずセッションを使って対象のオブジェクトを取得し、そのオブジェクトの属性を変更します。その後、session.commit() を呼び出すことで変更がデータベースに反映されます。SQLAlchemyはセッションに追加されたオブジェクトの変更を自動的に追跡します。

“`python
with Session() as session:
# IDが1のユーザーを取得
user_to_update = session.query(User).get(1)

if user_to_update:
    print(f"\n--- Updating user {user_to_update.id}: {user_to_update.name} ---")
    # オブジェクトの属性を変更
    user_to_update.age = user_to_update.age + 1 # 年齢を1歳増やす
    user_to_update.name = 'Alice ORM Updated'

    # commit() を呼び出すと、変更がデータベースに反映される (UPDATE文が実行される)
    session.commit()
    print(f"Updated user: {user_to_update}")
else:
    print("\nUser with id 1 not found for update.")

複数のユーザーをまとめて更新する場合も、オブジェクトをループして変更し、最後にコミットする

with Session() as session:
users_to_update_batch = session.query(User).filter(User.age < 30).all()
print(f”\n— Updating {len(users_to_update_batch)} users under 30 —“)
for user in users_to_update_batch:
user.age += 5 # 年齢を5歳増やす
print(f” Updating {user.name} to age {user.age}”)

session.commit()
print("Batch update committed.")

# 確認
print("\n--- Users after batch ORM update ---")
for user in session.query(User).order_by(User.id).all():
    print(user)

“`

10. データの削除 (Delete)

ORMでデータを削除するには、まずセッションを使って対象のオブジェクトを取得し、セッションの delete() メソッドにそのオブジェクトを渡します。その後、session.commit() を呼び出すことでデータベースから削除されます。

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

if user_to_delete:
    print(f"\n--- Deleting user {user_to_delete.id}: {user_to_delete.name} ---")
    # セッションからオブジェクトを削除対象としてマーク
    session.delete(user_to_delete)

    # commit() を呼び出すと、削除がデータベースに反映される (DELETE文が実行される)
    session.commit()
    print("User deleted.")
else:
    print("\nUser with id 2 not found for deletion.")

# 複数のユーザーをまとめて削除
users_to_delete_batch = session.query(User).filter(User.name.like('%Bob%')).all()
print(f"\n--- Deleting {len(users_to_delete_batch)} users with 'Bob' in name ---")
for user in users_to_delete_batch:
    session.delete(user)
    print(f"  Marked {user.name} for deletion")

session.commit()
print("Batch deletion committed.")

# 確認
print("\n--- Remaining ORM users ---")
for user in session.query(User).order_by(User.id).all():
    print(user)

“`

11. トランザクション管理 (ORM)

Core APIの connection.begin() と同様に、ORMのセッションもトランザクションを管理します。 session.commit() は現在のトランザクションをコミットし、session.rollback() は現在のトランザクションをロールバックします。

最も推奨される方法は、with session.begin(): を使うことです。これにより、ブロック内で発生した操作が自動的にトランザクションとして扱われ、成功すればコミット、例外発生時はロールバックが行われます。

“`python

例外が発生した場合のロールバック

try:
with Session() as session:
with session.begin():
# ユーザーを一人追加
risky_user = User(name=’Risky User’, age=99)
session.add(risky_user)
print(f”\n— Adding Risky User —“)

        # エラーを発生させる (例: 存在しないカラムにアクセスするなど、ここでは意図的に例外を発生させる)
        # risky_user.non_existent_attribute = 'error' # これは SQLAlchemy のエラーになる
        # または Python レベルのエラー
        1 / 0 # ZeroDivisionError

        # ここには到達しない

# ここにも到達しない

except Exception as e:
print(f”\n— An error occurred: {e} —“)
print(“Transaction will be rolled back.”)
# with session.begin() がロールバックを処理してくれる

ユーザーが追加されていないことを確認

with Session() as session:
print(“\n— Checking users after rollback attempt —“)
users_after_rollback = session.query(User).filter_by(name=’Risky User’).all()
if not users_after_rollback:
print(“Risky User was NOT added (rollback successful).”)
else:
print(“Risky User was added (rollback failed or unexpected behavior).”)

``with session.begin():` イディオムを使うと、開発者はトランザクションの開始、コミット、ロールバックを明示的に書く手間が省け、安全なリソース管理が実現できます。

12. SQLAlchemy 2.0 スタイル (Declarative Mapping)

SQLAlchemy 2.0 以降では、declarative_base の代わりに、よりモダンな DeclarativeBase クラスを作成する方法や、Pythonの型ヒントを活用した書き方が推奨されています。また、クエリの実行方法も session.query(...) ではなく、Coreと同様の select() 構文を使い、 session.execute(select(...)) または ORM用に拡張された session.scalars(select(...)) を使うスタイルが推奨されています。

ここでは簡単な例のみ示しますが、今後の学習ではこの新しいスタイルに慣れることをお勧めします。

新しい DeclarativeBase と型ヒントを使ったモデル定義:

“`python
from sqlalchemy import create_engine, Column, Integer, String
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, sessionmaker, relationship

新しいスタイルでの Declarative Base

class Base(DeclarativeBase):
pass

型ヒントを使ったモデル定義

class UserNew(Base):
tablename = ‘users_new’

# mapped_column を使うことで、型ヒントとカラム定義を同時に行える
id: Mapped[int] = mapped_column(primary_key=True)
name: Mapped[str] = mapped_column(String(50), nullable=False)
age: Mapped[int | None] = mapped_column(Integer, nullable=True) # Union[int, None] または int | None でNullableを表現

# リレーションシップも Mapped を使う
orders: Mapped[list["OrderNew"]] = relationship(back_populates="user") # 文字列で前方参照

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

class OrderNew(Base):
tablename = ‘orders_new’

id: Mapped[int] = mapped_column(primary_key=True)
user_id: Mapped[int] = mapped_column(ForeignKey('users_new.id'))
product_id: Mapped[int] = mapped_column(ForeignKey('products.id')) # products テーブルはそのまま使うと仮定
quantity: Mapped[int] = mapped_column(default=1)
order_date: Mapped[datetime.datetime | None] # Nullable DateTime

user: Mapped["UserNew"] = relationship(back_populates="orders")
# product: Mapped["Product"] = relationship() # Product も新スタイルにする場合は対応が必要

def __repr__(self):
     return f"<OrderNew(id={self.id}, user_id={self.user_id}, product_id={self.product_id}, quantity={self.quantity})>"

新しいテーブルを作成 (必要なら drop_all を先に)

Base.metadata.create_all(engine)

新しいスタイルのクエリ実行

from sqlalchemy import select

SessionNew = sessionmaker(bind=engine)

with SessionNew() as session:

# 新しいオブジェクトの作成・追加は同じ
# new_user = UserNew(name='New Alice', age=28)
# session.add(new_user)
# session.commit()

# クエリ実行スタイル (SELECT)
# stmt = select(UserNew).where(UserNew.name == 'New Alice')
# user = session.scalars(stmt).first() # scalar 結果 (単一カラムまたはオブジェクト) を取得
# print(f"\n--- New style query result: {user} ---")

# 全ユーザーを取得
# stmt_all = select(UserNew).order_by(UserNew.id)
# for user in session.scalars(stmt_all):
#     print(user)

# Joined Loading も select に options をつける
# stmt_joined = select(UserNew).options(joinedload(UserNew.orders))
# for user in session.scalars(stmt_joined):
#     print(user)
#     if user.orders:
#          print(f"  Orders: {[order.id for order in user.orders]}")

“`
新しいスタイルは、Pythonの最新の機能（型ヒント）を活用し、クエリの書き方もCoreとORMで一貫性が高まっています。特にこれからSQLAlchemyを始める場合は、新しいスタイルで学ぶことを検討してください。

まとめ：Core vs ORM

SQLAlchemy CoreとORMは、それぞれ異なるレベルの抽象化を提供します。

  • SQLAlchemy Core:

    • SQL文をPythonのオブジェクトで構築・実行するためのツール。
    • テーブル、カラム、制約などのデータベーススキーマを低レベルで操作。
    • オブジェクトマッピングは行わない。行データはタプルやRowオブジェクトとして扱う。
    • 複雑なクエリ、パフォーマンスが重要な部分、バルク操作、DDL操作などに適している。
    • SQLの詳細な制御が可能。

  • SQLAlchemy ORM:

    • Pythonクラスとデータベーステーブルをマッピング。
    • データベース操作をオブジェクト操作として行える。
    • リレーションシップをオブジェクト間の関連として管理。
    • 変更追跡、セッション管理、アイデンティティマップなどの機能を提供。
    • 一般的なアプリケーションロジック、ビジネスロジックの記述に適している。
    • 抽象度が高いため、Coreに比べてオーバーヘッドがある場合がある。

どちらを使うべきかは、ユースケースによります。

  • ほとんどの典型的なCRUD操作を含むアプリケーション開発では、ORM を使うことで開発効率とコードの可読性が大幅に向上します。
  • 非常に複雑なSQLクエリを組み立てる必要がある場合、データベース固有の高度な機能を利用する場合、またはパフォーマンスが極めて重要な一括データ処理を行う場合は、Core を使う方が適していることがあります。
  • 実際には、一つのアプリケーション内でORMとCoreを組み合わせて使用することも可能です。ORMで取得したセッションからCoreの接続を取得したり、Coreで構築したクエリをORMのセッションで実行したりできます。

結論と次のステップ

この記事では、SQLAlchemyの基本的な概念（Engine, Metadata, Table, Column, Sessionなど）と、CoreおよびORMを使った基本的なデータベース操作（Create, Read, Update, Delete）について解説しました。また、ORMにおけるモデル定義、リレーションシップ、そしてロード戦略にも触れました。

SQLAlchemyは非常に奥が深く、この記事で紹介できたのはその表面的な部分に過ぎません。しかし、ここまでの内容を理解できれば、多くのアプリケーションでSQLAlchemyを使い始めるための基礎は十分に身についているはずです。

次のステップとして、以下の内容を学習することをおすすめします。

  • 高度なクエリ: 集計関数 (SUM, COUNTなど)、グループ化 (GROUP BY)、サブクエリ、CTE (Common Table Expressions)。
  • リレーションシップの発展: 多対多の関係の定義、結合テーブル、より複雑なリレーションシップ設定。
  • データ型: カスタムデータ型の使用、JSON型などのデータベース固有の型の扱い。
  • イベントシステム: データベースイベントに対するフック。
  • 接続プールとトランザクション管理: より詳細な設定とベストプラクティス。
  • マイグレーション: データベーススキーマの変更を管理するためのAlembicなどのツール。
  • テスト: SQLAlchemyを使ったコードのテスト方法。
  • SQLAlchemy 2.0の新機能と推奨スタイル: 今後の開発で主流となる新しい書き方。

SQLAlchemyの公式ドキュメントは非常に詳細で網羅的ですが、初心者には少し難しいかもしれません。しかし、入門記事やチュートリアルで基本的な流れを掴んだ後には、強力なリファレンスとして活用できるでしょう。

データベース操作はアプリケーション開発の重要な部分です。SQLAlchemyのような強力なORMツールを使いこなすことで、より堅牢で保守しやすいコードを書けるようになります。

この入門記事が、あなたのSQLAlchemy学習の助けとなれば幸いです。 Happy Coding!

コメントする

メールアドレスが公開されることはありません。 ※ が付いている欄は必須項目です

上部へスクロール