はじめに

こんにちは、@masa-asa です。以前、マイグレーションに関する記事を書きました。その記事のシリーズとして、今回は ORMapper について記事にしたいと思います。
前回の記事リンクは以下になります。

ORMapper とは何か？

Wikipedia の記事には次のようにあります。

オブジェクト関係マッピング（英: Object-relational mapping, O/RM, ORM）とは、データベースとオブジェクト指向プログラミング言語の間の非互換なデータを変換するプログラミング技法である。

これは、データベースが取り扱うデータとプログラミング言語（特にオブジェクト指向なもの）が扱うデータの関係が異なることで起こる扱いにくさを、何かで解決しようとする試みがあり、それが ORMapper という形になっていると考えるのがわかりやすいかもしれません。

ORMapper は、

データベースの 1 行を、アプリケーション上のオブジェクトに変換する
オブジェクトの変更を、対応する SQL に変換してデータベースに反映する

といった「橋渡し役」を担うことができます。

Python で ORMapper を使う準備をする

Python の ORMapper で著名なものに SQLAlchemy があります。今回はこのツールを使って ORMapper を体験してみます。

SQLAlchemy のインストール

以下のコマンドで SQLAlchemy をインストールします。

uv add "sqlalchemy[asyncio]"
uv add asyncpg

データベースに接続するための設定を作成する

SQLAlchemy でデータベースに接続をするための方法について説明していきます。

エンジンの作成

SQLAlchemy のエンジンは、データベースへの接続の土台です。データベースへの接続情報を記載してエンジンのオブジェクトを作成することで、そのデータベースへの接続が可能になります。
実際にエンジンを作成するコードは次のようになります。

同期処理のエンジン作成

from sqlalchemy import create_engine

engine = create_engine('postgresql+psycopg2://user:password@localhost:5432/mydb', echo=True)

非同期処理のエンジン作成

from sqlalchemy.ext.asyncio import (
    AsyncEngine,
    create_async_engine,
)

engine: AsyncEngine = create_async_engine("postgresql+asyncpg://user:password@localhost:5432/mydb", echo=True)

echo は実行される SQL ステートメントを標準出力に出力するかどうかを決めるフラグです。

本記事の例では、原則非同期の場合を想定して進行します。

セッションの作成

エンジンを作ったら、次はセッションを準備します。
セッションはデータベースに接続してからの一連の操作（作業）を 1 つの単位としてみたものです。
データの追加・更新・削除などの処理をしてコミットしてデータベースに反映するまでの一連の作業を 1 つのセッションとして管理します。

セッションを作成するには、async_sessionmaker を用います。

from sqlalchemy.ext.asyncio import (
    AsyncEngine,
    AsyncSession,
    async_sessionmaker,
    create_async_engine,
)

engine: AsyncEngine = create_async_engine(
    "postgresql+asyncpg://user:password@localhost:5432/mydb",
    echo=True,
)

# セッションの作成部分
async_session = async_sessionmaker(
    bind=engine,
    class_=AsyncSession,
    expire_on_commit=False,
)

非同期処理の場合は、expire_on_commit=False とすることが推奨されます。False にすることで、コミット後も利用していたオブジェクトが無効化されず、オブジェクトの属性にアクセスができます。
以前書いた以下の記事にあるような状態を回避できるというわけです。

SQLAlchemy のセッションの始まりから終わりまでを概念的な処理の流れとして表すと以下の図のようになります。
1 つのセッションの開始〜終了の中に CRUD の操作などのデータ操作が入り、それら一連の操作をコミットしてセッションを閉じるというものが「セッション」です。

「モデル」の定義について

ORMapper では、データベースのテーブルに対応する「モデル」を活用してデータベースへの操作を行います。
SQLAlchemy（Python）の場合は、クラスとしてモデルを定義します。

SQLAlchemy 2.x では、DeclarativeBase を継承して Base クラスを作成します。すべてのモデルクラスは、この Base を継承して定義します。

基底モデルの定義は以下です。

from sqlalchemy.orm import DeclarativeBase


class Base(DeclarativeBase):
    pass

これを継承して、User テーブルに相当するモデルを記述してみます。

from sqlalchemy import String
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column


class Base(DeclarativeBase):
    pass


class User(Base):
    __tablename__ = "users"

    id: Mapped[int] = mapped_column(primary_key=True, autoincrement=True)
    name: Mapped[str] = mapped_column(String(50))
    email: Mapped[str] = mapped_column(String(100), unique=True)

モデル定義について、ポイントになりそうな部分をまとめてみます。

__tablename__：テーブル名を指定します
mapped_column：カラムの制約（主キー、ユニーク、文字列長など）を設定します
Mapped[<型>]：型には int や str などを指定します。Python の型ヒントと統合されているため、IDE の補完や型チェッカー（mypy など）が機能し、開発時にエラーを早期に発見しやすくなります

データベースへの操作を行う

CRUD 操作

ここからは、セッションを使って CRUD 操作の例を見ていきます。
CRUD とは Create（作成）、Read（読み取り）、Update（更新）、Delete（削除）でデータに対する基本操作を意味します。

また、前項までで、セッションファクトリの定義、ユーザーモデルの定義をしました。
これからのコードでは、それらを import して使用することとします。
※実際には、コードを記述したファイルのディレクトリ構造に従って import のパスを指定する必要があります。

from your_project.database import async_session  # セッションファクトリ（エンジンの作成 で定義）
from your_project.models import User              # モデル（モデルの定義 で定義）

Create（作成）

async with async_session() as session:
    user = User(name="田中太郎", email="tanaka@example.com")
    session.add(user)
    await session.commit()
    print(user.id)

前項で定義した User クラスのインスタンスを作成し、session.add() でセッションに登録します。session.commit() を呼ぶことで、データベースに反映されます。

echo=True を設定している場合、コミット時に以下のような SQL が出力されます。

INSERT INTO users (name, email) VALUES ($1::VARCHAR, $2::VARCHAR) RETURNING users.id

Read（取得）

単一レコードの取得

from sqlalchemy import select

async with async_session() as session:
    stmt = select(User).where(User.id == 1)
    result = await session.execute(stmt)
    user = result.scalar_one_or_none()
    print(user.name)

select() で SELECT 文を組み立て、session.execute() で実行します。scalar_one_or_none() は結果が 1 件ならそのオブジェクトを、0 件なら None を返します。

全件取得

from sqlalchemy import select

async with async_session() as session:
    stmt = select(User)
    result = await session.execute(stmt)
    users = list(result.scalars().all())
    for user in users:
        print(user.name)

scalars().all() で、取得した全行を User オブジェクトのリストとして受け取ります。

補足として、本記事では scalar_one_or_none() と scalars().all() の場合を例にしましたが、session.execute() の戻り値である Result オブジェクトには他にも取得メソッドがあります。以下は主なものの一覧です（すべてではありません）。

メソッド	用途	結果なし	複数件
`scalar_one()`	1 件のみ取得（厳密）	例外	例外
`scalar_one_or_none()`	0〜1 件取得	`None`	例外
`scalars().all()`	全件をリストで取得	空リスト	全件返す
`scalars().first()`	先頭 1 件取得	`None`	先頭を返す

SQLAlchemy で select(User) のように単一エンティティを取得する場合は、Row（行オブジェクト）ではなくモデルのインスタンスを直接返す scalar* 系のメソッドを使うのが一般的です。

詳細については、公式ドキュメントをご参照ください。

Update（更新）

from sqlalchemy import select

async with async_session() as session:
    stmt = select(User).where(User.id == 1)
    result = await session.execute(stmt)
    user = result.scalar_one_or_none()

    if user:
        user.email = "new_email@example.com"
        await session.commit()

更新の例では、ユーザー ID を条件に取得してきたユーザーオブジェクトに対して、そのオブジェクトの属性（ここではメールアドレス）を書き換えています。
このように SQLAlchemy では、取得したオブジェクトの属性を変更してコミットするだけで、UPDATE 文が自動的に発行されます。
これは、セッションが管理しているオブジェクトの変更を自動的に検知する仕組み（ダーティチェック）によるものです。

echo=True を設定している場合、コミット時に以下のような SQL が出力されます。
ダーティチェックが機能し、UPDATE 文が発行されていることがわかります。

UPDATE users SET email=$1::VARCHAR WHERE users.id = $2::INTEGER

Delete（削除）

from sqlalchemy import select

async with async_session() as session:
    stmt = select(User).where(User.id == 1)
    result = await session.execute(stmt)
    user = result.scalar_one_or_none()

    if user:
        await session.delete(user)
        await session.commit()

この例では、削除対象のレコードをユーザー ID で検索し、オブジェクトとして取得しています。そしてsession.delete() でオブジェクトを削除対象としてマークし、コミットすることで DELETE 文が実行されます。

echo=True を設定している場合、コミット時に以下のような SQL が出力されます。

DELETE FROM users WHERE users.id = $1::INTEGER

まとめ

本記事では、ORMapper の基本的な概念から SQLAlchemy を使った実装までを一通り紹介しました。

ORMapper は、データベースのテーブルと Python のオブジェクトを対応づけ、SQL を直接書かずにデータベース操作を行うための仕組みです
エンジン でデータベースへの接続を管理し、セッション で一連の操作をまとめます
モデル定義 により、テーブル構造を Python のクラスとして表現できます
CRUD 操作 では、オブジェクトの操作がそのまま SQL に変換されることを確認しました