Python でデータベースを扱う際、ORM（Object-Relational Mapping）は開発効率を大きく向上させます。しかし、SQLAlchemy、Tortoise ORM、Beanie という主要な ORM の中から、どれを選ぶべきか迷っていませんか？

それぞれの ORM には独自の強みがあり、プロジェクトの特性によって最適な選択肢は変わってきます。この記事では、3 つの ORM を実際のコード例とともに徹底比較し、あなたのプロジェクトに最適な選択ができるようサポートいたします。

背景

Python ORM の重要性

Python のバックエンド開発において、データベース操作は避けて通れません。生の SQL を書くこともできますが、ORM を使うことで以下のメリットが得られるのです。

• コードの可読性向上：Python のオブジェクト指向な書き方でデータベース操作ができます
• データベース移行の容易性：ORM が抽象化レイヤーを提供するため、データベース変更時の影響を最小限に抑えられるでしょう
• セキュリティ強化：SQL インジェクション攻撃への対策が自動的に組み込まれています
• 開発速度の向上：ボイラープレートコードを削減し、ビジネスロジックに集中できますね

3 つの ORM 概要

Python エコシステムには数多くの ORM が存在しますが、今回は特に人気の高い 3 つに焦点を当てます。

以下の図で、各 ORM の主要な特徴と対応データベースを確認しましょう。

mermaidCopy codeflowchart TB
  subgraph SQLAlchemy
    SA["SQLAlchemy<br />成熟度: ★★★★★"]
    SA --> SA_DB["PostgreSQL/MySQL/<br />SQLite/Oracle など"]
    SA --> SA_TYPE["同期・非同期両対応"]
  end

  subgraph Tortoise
    TO["Tortoise ORM<br />成熟度: ★★★☆☆"]
    TO --> TO_DB["PostgreSQL/MySQL/<br />SQLite"]
    TO --> TO_TYPE["非同期専用"]
  end

  subgraph Beanie
    BE["Beanie<br />成熟度: ★★★☆☆"]
    BE --> BE_DB["MongoDB 専用"]
    BE --> BE_TYPE["非同期専用"]
  end

図の要点：

• SQLAlchemy はリレーショナルデータベース全般に対応し、同期・非同期両方をサポート
• Tortoise ORM は非同期専用で主要な RDBMS に対応
• Beanie は MongoDB に特化した非同期 ORM

それぞれの ORM は異なる設計思想を持っており、使用するデータベースやアプリケーションの要件によって選択が変わってきます。

課題

ORM 選択における主要な判断ポイント

プロジェクトに適した ORM を選ぶ際、開発者は以下のような課題に直面します。

データベースの種類による制約

リレーショナルデータベースを使うのか、NoSQL の MongoDB を使うのかによって、選択肢が大きく変わってしまいます。Beanie は MongoDB 専用であるため、RDBMS を使う場合は候補から外れますね。

同期処理 vs 非同期処理

FastAPI や Sanic などの非同期フレームワークを使用する場合、ORM も非同期対応である必要があります。一方、Flask や Django などの従来型フレームワークでは、同期処理が中心となるでしょう。

以下の図で、各 ORM とフレームワークの相性を見ていきましょう。

mermaidCopy codeflowchart LR
  subgraph Frameworks["Webフレームワーク"]
    asyncNode["FastAPI/Sanic<br />(非同期)"]
    syncNode["Flask/Django<br />(同期)"]
  end

  subgraph ORMs["ORM選択"]
    sa["SQLAlchemy<br />(両対応)"]
    tor["Tortoise ORM<br />(非同期のみ)"]
    bean["Beanie<br />(非同期のみ)"]
  end

  asyncNode --|最適|--> tor
  asyncNode --|最適|--> bean
  asyncNode --|対応可|--> sa
  syncNode --|最適|--> sa
  syncNode --|不適合|--> tor
  syncNode --|不適合|--> bean

図で理解できる要点：

• 非同期フレームワークには Tortoise や Beanie が自然にフィット
• 同期フレームワークでは SQLAlchemy が王道
• SQLAlchemy は両方に対応できる柔軟性がある

学習コストと開発効率のバランス

SQLAlchemy は非常に強力ですが、習得には時間がかかります。Tortoise ORM や Beanie は比較的シンプルな API を提供していますが、機能の豊富さでは SQLAlchemy に及びません。

パフォーマンス要件

大量のデータを扱う場合や、高いスループットが求められる場合、ORM のパフォーマンス特性が重要になってきます。特に N+1 問題への対応やクエリの最適化機能が、実運用では大きな差を生むでしょう。

コミュニティとエコシステム

長期的なプロジェクトでは、ORM のコミュニティの活発さや、サードパーティライブラリの充実度も重要な判断材料となります。

解決策

SQLAlchemy：業界標準の万能 ORM

SQLAlchemy は 2006 年から開発されている、Python で最も成熟した ORM です。幅広いデータベースをサポートし、同期・非同期両方の処理に対応しています。

SQLAlchemy の強み

#	項目	詳細
1	成熟度	15 年以上の開発実績と豊富なドキュメント
2	柔軟性	Core API と ORM API の 2 層構造で低レベル操作も可能
3	データベース対応	PostgreSQL、MySQL、SQLite、Oracle など主要 RDBMS すべてに対応
4	エコシステム	Alembic（マイグレーション）など充実したツール群
5	同期・非同期	バージョン 1.4 以降、asyncio を完全サポート

SQLAlchemy の活用シーン

• 既存の大規模プロジェクトへの統合
• 複雑なクエリや高度なデータベース操作が必要な場合
• 同期・非同期両方の処理が混在するアプリケーション
• データベースを将来的に切り替える可能性がある場合

Tortoise ORM：非同期に特化したシンプルな ORM

Tortoise ORM は Django ORM のような直感的な API を持ちながら、非同期処理に完全対応した ORM です。2018 年にリリースされ、FastAPI などの非同期フレームワークとの相性が抜群ですね。

Tortoise ORM の強み

#	項目	詳細
1	シンプルさ	Django 風の直感的な API で学習コストが低い
2	非同期特化	asyncio ネイティブで高いパフォーマンス
3	型ヒント	Pydantic との統合で型安全性が高い
4	マイグレーション	Aerich による自動マイグレーション機能
5	ドキュメント	わかりやすい公式ドキュメントとサンプル

Tortoise ORM の活用シーン

• FastAPI や Sanic などの非同期フレームワークを使用する場合
• Django からの移行で同様の API を求める場合
• シンプルで理解しやすい ORM を求める場合
• スタートアップや中小規模のプロジェクト

Beanie：MongoDB 専用の非同期 ORM

Beanie は MongoDB 専用に設計された非同期 ORM で、Pydantic と緊密に統合されています。2020 年にリリースされた新しい ORM ですが、MongoDB を使うプロジェクトでは非常に強力な選択肢となるでしょう。

Beanie の強み

#	項目	詳細
1	MongoDB 最適化	MongoDB の機能をフル活用できる設計
2	Pydantic 統合	データ検証とシリアライゼーションが統一的
3	非同期ネイティブ	Motor（非同期 MongoDB ドライバ）ベース
4	型安全性	完全な型ヒント対応で IDE サポートが充実
5	シンプル API	最小限の設定で使い始められる

Beanie の活用シーン

• MongoDB を使用するプロジェクト
• FastAPI と Pydantic を既に使用している場合
• ドキュメント指向データベースの柔軟性を活かしたい場合
• リアルタイムアプリケーションやログ収集システム

選択基準のフローチャート

以下の図で、あなたのプロジェクトに最適な ORM を選択するためのフローを確認しましょう。

mermaidCopy codeflowchart TD
    start["ORM選択開始"] --> db_type{"データベースの<br/>種類は?"}

    db_type -->|MongoDB| beanie["Beanie"]
    db_type -->|RDBMS| async_check{"非同期処理が<br/>必要?"}

    async_check -->|はい| complexity{"プロジェクトの<br/>複雑度は?"}
    async_check -->|いいえ| sqlalchemy_sync["SQLAlchemy<br/>(同期モード)"]

    complexity -->|シンプル| tortoise["Tortoise ORM"]
    complexity -->|複雑| sqlalchemy_async["SQLAlchemy<br/>(非同期モード)"]

    beanie --> result_beanie["・MongoDB専用<br/>・Pydantic統合<br/>・型安全"]
    tortoise --> result_tortoise["・学習容易<br/>・FastAPI最適<br/>・中規模向け"]
    sqlalchemy_async --> result_sa_async["・高機能<br/>・柔軟性高<br/>・大規模向け"]
    sqlalchemy_sync --> result_sa_sync["・業界標準<br/>・豊富な実績<br/>・エコシステム充実"]

図で理解できる要点：

• データベースの種類が最初の分岐点
• 非同期の必要性が次の重要な判断基準
• プロジェクトの複雑度によって最適な選択が変わる

具体例

それぞれの ORM を使った実装例を見ていきましょう。同じ機能を 3 つの ORM で実装することで、違いが明確に理解できます。

共通の要件定義

以下のようなブログアプリケーションを想定します。

• ユーザー（User）と記事（Post）の 2 つのモデル
• ユーザーは複数の記事を持つ（1 対多の関係）
• 記事の作成、取得、更新、削除の基本操作
• 非同期 API エンドポイントでの実装

SQLAlchemy 2.0 による実装

モデル定義

まず、SQLAlchemy でデータベースモデルを定義します。SQLAlchemy 2.0 の新しい宣言的スタイルを使用しますね。

pythonCopy code# models.py - SQLAlchemyのモデル定義
from datetime import datetime
from typing import List, Optional
from sqlalchemy import String, Text, ForeignKey, DateTime
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, relationship


# Baseクラスの定義
class Base(DeclarativeBase):
    pass

次に、User モデルを定義します。このモデルはユーザー情報を管理します。

pythonCopy code# User モデルの定義
class User(Base):
    __tablename__ = "users"

    # プライマリキー
    id: Mapped[int] = mapped_column(primary_key=True)

    # ユーザー名（必須、一意）
    username: Mapped[str] = mapped_column(String(50), unique=True, nullable=False)

    # メールアドレス（必須、一意）
    email: Mapped[str] = mapped_column(String(100), unique=True, nullable=False)

    # 作成日時（自動設定）
    created_at: Mapped[datetime] = mapped_column(DateTime, default=datetime.utcnow)

    # リレーション: このユーザーが持つ記事の一覧
    posts: Mapped[List["Post"]] = relationship(back_populates="author", cascade="all, delete-orphan")

続いて、Post モデルを定義します。記事データを管理するモデルですね。

pythonCopy code# Post モデルの定義
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] = mapped_column(Text, nullable=False)

    # 外部キー: ユーザーID
    author_id: Mapped[int] = mapped_column(ForeignKey("users.id"), nullable=False)

    # 作成日時と更新日時
    created_at: Mapped[datetime] = mapped_column(DateTime, default=datetime.utcnow)
    updated_at: Mapped[datetime] = mapped_column(DateTime, default=datetime.utcnow, onupdate=datetime.utcnow)

    # リレーション: この記事の著者
    author: Mapped["User"] = relationship(back_populates="posts")

データベース接続設定

非同期処理に対応したデータベース接続を設定します。

pythonCopy code# database.py - データベース接続の設定
from sqlalchemy.ext.asyncio import create_async_engine, async_sessionmaker, AsyncSession

# データベースURL（PostgreSQLの例）
DATABASE_URL = "postgresql+asyncpg://user:password@localhost/blog_db"

# 非同期エンジンの作成
engine = create_async_engine(
    DATABASE_URL,
    echo=True,  # SQLログを出力（開発時のみ推奨）
    future=True
)

# セッションファクトリーの作成
AsyncSessionLocal = async_sessionmaker(
    engine,
    class_=AsyncSession,
    expire_on_commit=False
)

テーブルを作成する初期化関数を定義しましょう。

pythonCopy code# テーブル作成関数
async def init_db():
    """データベースのテーブルを作成する"""
    async with engine.begin() as conn:
        # すべてのテーブルを作成
        await conn.run_sync(Base.metadata.create_all)

CRUD 操作の実装

ユーザー作成処理を実装します。トランザクション管理にも注目してください。

pythonCopy code# crud.py - CRUD操作の実装
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy.orm import selectinload
from models import User, Post


async def create_user(session: AsyncSession, username: str, email: str) -> User:
    """新しいユーザーを作成する"""
    # Userオブジェクトを作成
    new_user = User(username=username, email=email)

    # セッションに追加
    session.add(new_user)

    # データベースにコミット
    await session.commit()

    # オブジェクトをリフレッシュして最新のデータを取得
    await session.refresh(new_user)

    return new_user

記事作成処理を実装します。外部キーの関連付けに注意しましょう。

pythonCopy codeasync def create_post(session: AsyncSession, title: str, content: str, author_id: int) -> Post:
    """新しい記事を作成する"""
    # Postオブジェクトを作成（author_idで関連付け）
    new_post = Post(title=title, content=content, author_id=author_id)

    session.add(new_post)
    await session.commit()
    await session.refresh(new_post)

    return new_post

ユーザーとその記事を一括取得する処理を実装します。N+1 問題を回避するために selectinload を使用していますね。

pythonCopy codeasync def get_user_with_posts(session: AsyncSession, user_id: int) -> Optional[User]:
    """ユーザーとその記事を取得する（N+1問題を回避）"""
    # selectinloadでpostsを事前読み込み
    stmt = select(User).options(selectinload(User.posts)).where(User.id == user_id)

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

    # 結果を取得（存在しない場合はNone）
    user = result.scalar_one_or_none()

    return user

記事を更新する処理を実装します。部分的な更新にも対応できる柔軟な設計ですね。

pythonCopy codeasync def update_post(session: AsyncSession, post_id: int, title: str = None, content: str = None) -> Optional[Post]:
    """記事を更新する"""
    # 記事を取得
    stmt = select(Post).where(Post.id == post_id)
    result = await session.execute(stmt)
    post = result.scalar_one_or_none()

    if not post:
        return None

    # 提供された値のみを更新
    if title is not None:
        post.title = title
    if content is not None:
        post.content = content

    await session.commit()
    await session.refresh(post)

    return post

FastAPI との統合

FastAPI アプリケーションでの使用例を見ていきましょう。

pythonCopy code# main.py - FastAPIアプリケーション
from fastapi import FastAPI, Depends, HTTPException
from sqlalchemy.ext.asyncio import AsyncSession
from database import AsyncSessionLocal, init_db
from crud import create_user, create_post, get_user_with_posts
from pydantic import BaseModel

app = FastAPI()

# Pydanticモデル（リクエスト/レスポンス用）
class UserCreate(BaseModel):
    username: str
    email: str

class PostCreate(BaseModel):
    title: str
    content: str

依存性注入を使ってデータベースセッションを管理します。

pythonCopy code# 依存性注入: データベースセッションの取得
async def get_db():
    """データベースセッションを取得する依存関数"""
    async with AsyncSessionLocal() as session:
        try:
            yield session
        finally:
            await session.close()

起動時にデータベースを初期化し、API エンドポイントを実装します。

pythonCopy code# 起動時にテーブルを作成
@app.on_event("startup")
async def startup_event():
    await init_db()

# ユーザー作成エンドポイント
@app.post("/users/")
async def create_user_endpoint(user: UserCreate, db: AsyncSession = Depends(get_db)):
    """新しいユーザーを作成するAPIエンドポイント"""
    try:
        new_user = await create_user(db, user.username, user.email)
        return {"id": new_user.id, "username": new_user.username, "email": new_user.email}
    except Exception as e:
        raise HTTPException(status_code=400, detail=str(e))

# ユーザーと記事取得エンドポイント
@app.get("/users/{user_id}")
async def get_user_endpoint(user_id: int, db: AsyncSession = Depends(get_db)):
    """ユーザーとその記事を取得するAPIエンドポイント"""
    user = await get_user_with_posts(db, user_id)
    if not user:
        raise HTTPException(status_code=404, detail="User not found")

    return {
        "id": user.id,
        "username": user.username,
        "posts": [{"id": p.id, "title": p.title} for p in user.posts]
    }

Tortoise ORM による実装

Tortoise ORM は Django ライクなシンプルな API を提供します。同じ機能を実装してみましょう。

モデル定義

Tortoise ORM でのモデル定義は非常に直感的です。

pythonCopy code# models.py - Tortoise ORMのモデル定義
from tortoise import fields
from tortoise.models import Model


class User(Model):
    """ユーザーモデル"""
    # IDは自動的に作成される
    id = fields.IntField(pk=True)

    # ユーザー名（一意制約）
    username = fields.CharField(max_length=50, unique=True)

    # メールアドレス（一意制約）
    email = fields.CharField(max_length=100, unique=True)

    # 作成日時（自動設定）
    created_at = fields.DatetimeField(auto_now_add=True)

    # リレーション: 逆参照でpostsにアクセス可能
    # posts: fields.ReverseRelation["Post"]

    class Meta:
        table = "users"

Post モデルも同様にシンプルに定義できます。

pythonCopy codeclass Post(Model):
    """記事モデル"""
    id = fields.IntField(pk=True)

    # 記事タイトル
    title = fields.CharField(max_length=200)

    # 記事本文
    content = fields.TextField()

    # 外部キー: ユーザーとの関連
    author = fields.ForeignKeyField(
        "models.User",  # モデルの参照
        related_name="posts",  # 逆参照時の名前
        on_delete=fields.CASCADE  # ユーザー削除時に記事も削除
    )

    # 作成日時と更新日時
    created_at = fields.DatetimeField(auto_now_add=True)
    updated_at = fields.DatetimeField(auto_now=True)

    class Meta:
        table = "posts"

データベース接続設定

Tortoise ORM の初期化は設定辞書を使用します。

pythonCopy code# database.py - データベース接続設定
from tortoise import Tortoise

# データベース設定
TORTOISE_ORM = {
    "connections": {
        # 接続名: データベースURL
        "default": "postgres://user:password@localhost:5432/blog_db"
    },
    "apps": {
        "models": {
            # モデルが定義されているモジュール
            "models": ["models"],
            # デフォルト接続
            "default_connection": "default",
        }
    }
}


async def init_db():
    """データベースを初期化する"""
    # Tortoiseを初期化
    await Tortoise.init(config=TORTOISE_ORM)

    # テーブルを作成
    await Tortoise.generate_schemas()


async def close_db():
    """データベース接続を閉じる"""
    await Tortoise.close_connections()

CRUD 操作の実装

Tortoise ORM の CRUD 操作は非常にシンプルです。

pythonCopy code# crud.py - CRUD操作
from models import User, Post
from typing import Optional


async def create_user(username: str, email: str) -> User:
    """新しいユーザーを作成する"""
    # createメソッドで作成と保存を同時に実行
    user = await User.create(username=username, email=email)
    return user


async def create_post(title: str, content: str, author_id: int) -> Post:
    """新しい記事を作成する"""
    # author_idで関連付け
    post = await Post.create(title=title, content=content, author_id=author_id)
    return post

関連データの取得も prefetch_related を使えば簡単です。

pythonCopy codeasync def get_user_with_posts(user_id: int) -> Optional[User]:
    """ユーザーとその記事を取得する"""
    # prefetch_relatedでN+1問題を回避
    user = await User.get_or_none(id=user_id).prefetch_related("posts")
    return user


async def update_post(post_id: int, title: str = None, content: str = None) -> Optional[Post]:
    """記事を更新する"""
    post = await Post.get_or_none(id=post_id)

    if not post:
        return None

    # 値を更新
    if title is not None:
        post.title = title
    if content is not None:
        post.content = content

    # 保存
    await post.save()

    return post

リスト取得やフィルタリングも直感的に書けます。

pythonCopy codeasync def get_all_posts(limit: int = 10) -> list[Post]:
    """すべての記事を取得する（著者情報も含む）"""
    # select_relatedで関連データを同時取得
    posts = await Post.all().select_related("author").limit(limit)
    return posts


async def delete_post(post_id: int) -> bool:
    """記事を削除する"""
    post = await Post.get_or_none(id=post_id)

    if not post:
        return False

    await post.delete()
    return True

FastAPI との統合

Tortoise ORM を FastAPI で使用する例です。

pythonCopy code# main.py - FastAPIアプリケーション
from fastapi import FastAPI, HTTPException
from tortoise.contrib.fastapi import register_tortoise
from pydantic import BaseModel
from crud import create_user, create_post, get_user_with_posts

app = FastAPI()

# Pydanticモデル
class UserCreate(BaseModel):
    username: str
    email: str

class PostCreate(BaseModel):
    title: str
    content: str

Tortoise ORM は FastAPI との統合が簡単に行えます。

pythonCopy code# Tortoiseの登録（起動時に初期化、終了時にクローズ）
register_tortoise(
    app,
    db_url="postgres://user:password@localhost:5432/blog_db",
    modules={"models": ["models"]},
    generate_schemas=True,  # テーブル自動作成
    add_exception_handlers=True,  # 例外ハンドラーを追加
)


# ユーザー作成エンドポイント
@app.post("/users/")
async def create_user_endpoint(user: UserCreate):
    """新しいユーザーを作成する"""
    try:
        new_user = await create_user(user.username, user.email)
        return {"id": new_user.id, "username": new_user.username}
    except Exception as e:
        raise HTTPException(status_code=400, detail=str(e))


# ユーザー取得エンドポイント
@app.get("/users/{user_id}")
async def get_user_endpoint(user_id: int):
    """ユーザーとその記事を取得する"""
    user = await get_user_with_posts(user_id)

    if not user:
        raise HTTPException(status_code=404, detail="User not found")

    return {
        "id": user.id,
        "username": user.username,
        "posts": [{"id": p.id, "title": p.title} for p in user.posts]
    }

Beanie による実装

Beanie は MongoDB 専用の ORM で、Pydantic との統合が特徴です。

モデル定義

Beanie のモデルは Pydantic の BaseModel を継承します。

pythonCopy code# models.py - Beanieのモデル定義
from datetime import datetime
from typing import Optional, List
from beanie import Document, Link
from pydantic import Field, EmailStr


class User(Document):
    """ユーザードキュメント"""
    # MongoDBの_idは自動的に作成される
    username: str = Field(..., unique=True, max_length=50)
    email: EmailStr = Field(..., unique=True)
    created_at: datetime = Field(default_factory=datetime.utcnow)

    class Settings:
        # コレクション名
        name = "users"

        # インデックス設定
        indexes = [
            "username",
            "email",
        ]

Post モデルは Link を使ってユーザーと関連付けます。

pythonCopy codeclass Post(Document):
    """記事ドキュメント"""
    title: str = Field(..., max_length=200)
    content: str

    # Linkで他のドキュメントへの参照を作成
    author: Link[User]

    created_at: datetime = Field(default_factory=datetime.utcnow)
    updated_at: datetime = Field(default_factory=datetime.utcnow)

    class Settings:
        name = "posts"

        # インデックス設定
        indexes = [
            "author",
            "created_at",
        ]

データベース接続設定

Beanie の初期化は Motor クライアントを使用します。

pythonCopy code# database.py - データベース接続設定
from motor.motor_asyncio import AsyncIOMotorClient
from beanie import init_beanie
from models import User, Post

# MongoDBクライアント
client = None


async def init_db():
    """データベースを初期化する"""
    global client

    # Motorクライアントを作成
    client = AsyncIOMotorClient("mongodb://localhost:27017")

    # Beanieを初期化（使用するドキュメントクラスを登録）
    await init_beanie(
        database=client.blog_db,  # データベース名
        document_models=[User, Post]  # ドキュメントモデル
    )


async def close_db():
    """データベース接続を閉じる"""
    if client:
        client.close()

CRUD 操作の実装

Beanie の CRUD 操作は Pydantic ベースで型安全です。

pythonCopy code# crud.py - CRUD操作
from models import User, Post
from typing import Optional
from beanie import PydanticObjectId


async def create_user(username: str, email: str) -> User:
    """新しいユーザーを作成する"""
    # Userオブジェクトを作成
    user = User(username=username, email=email)

    # 保存
    await user.insert()

    return user

Post の作成では、Link を使って User を関連付けます。

pythonCopy codeasync def create_post(title: str, content: str, author_id: PydanticObjectId) -> Post:
    """新しい記事を作成する"""
    # ユーザーを取得
    author = await User.get(author_id)

    if not author:
        raise ValueError("Author not found")

    # Postオブジェクトを作成（Linkで関連付け）
    post = Post(title=title, content=content, author=author)

    await post.insert()

    return post

関連データの取得は fetch_links を使います。

pythonCopy codeasync def get_user_by_id(user_id: PydanticObjectId) -> Optional[User]:
    """IDでユーザーを取得する"""
    user = await User.get(user_id)
    return user


async def get_posts_by_user(user_id: PydanticObjectId) -> List[Post]:
    """ユーザーの記事を取得する（著者情報も含む）"""
    user = await User.get(user_id)

    if not user:
        return []

    # findで検索し、fetch_linksで関連データを取得
    posts = await Post.find(Post.author.id == user.id).to_list()

    return posts

更新と削除の操作も直感的です。

pythonCopy codeasync def update_post(post_id: PydanticObjectId, title: str = None, content: str = None) -> Optional[Post]:
    """記事を更新する"""
    post = await Post.get(post_id)

    if not post:
        return None

    # 値を更新
    if title is not None:
        post.title = title
    if content is not None:
        post.content = content

    post.updated_at = datetime.utcnow()

    # 保存
    await post.save()

    return post


async def delete_post(post_id: PydanticObjectId) -> bool:
    """記事を削除する"""
    post = await Post.get(post_id)

    if not post:
        return False

    await post.delete()
    return True

FastAPI との統合

Beanie を FastAPI で使用する例です。Pydantic との統合が美しいですね。

pythonCopy code# main.py - FastAPIアプリケーション
from fastapi import FastAPI, HTTPException
from contextlib import asynccontextmanager
from pydantic import BaseModel, EmailStr
from beanie import PydanticObjectId
from database import init_db, close_db
from crud import create_user, create_post, get_user_by_id, get_posts_by_user


# ライフスパン管理（起動・終了時の処理）
@asynccontextmanager
async def lifespan(app: FastAPI):
    # 起動時
    await init_db()
    yield
    # 終了時
    await close_db()


app = FastAPI(lifespan=lifespan)


# Pydanticモデル
class UserCreate(BaseModel):
    username: str
    email: EmailStr


class PostCreate(BaseModel):
    title: str
    content: str

API エンドポイントの実装です。型安全性が保たれていることに注目しましょう。

pythonCopy code# ユーザー作成エンドポイント
@app.post("/users/")
async def create_user_endpoint(user: UserCreate):
    """新しいユーザーを作成する"""
    try:
        new_user = await create_user(user.username, user.email)
        # Beanieのドキュメントは自動的にJSONシリアライズ可能
        return {"id": str(new_user.id), "username": new_user.username}
    except Exception as e:
        raise HTTPException(status_code=400, detail=str(e))


# 記事作成エンドポイント
@app.post("/users/{user_id}/posts/")
async def create_post_endpoint(user_id: PydanticObjectId, post: PostCreate):
    """ユーザーの記事を作成する"""
    try:
        new_post = await create_post(post.title, post.content, user_id)
        return {"id": str(new_post.id), "title": new_post.title}
    except ValueError as e:
        raise HTTPException(status_code=404, detail=str(e))
    except Exception as e:
        raise HTTPException(status_code=400, detail=str(e))


# ユーザーと記事取得エンドポイント
@app.get("/users/{user_id}")
async def get_user_endpoint(user_id: PydanticObjectId):
    """ユーザーとその記事を取得する"""
    user = await get_user_by_id(user_id)

    if not user:
        raise HTTPException(status_code=404, detail="User not found")

    posts = await get_posts_by_user(user_id)

    return {
        "id": str(user.id),
        "username": user.username,
        "posts": [{"id": str(p.id), "title": p.title} for p in posts]
    }

3 つの ORM の実装比較

以下の図で、各 ORM の実装フローを比較してみましょう。

mermaidCopy codesequenceDiagram
    participant Client as クライアント
    participant API as FastAPI
    participant ORM as ORM層
    participant DB as データベース

    Note over Client,DB: SQLAlchemy: セッション管理が明示的
    Client->>API: POST /users/
    API->>ORM: get_db() でセッション取得
    ORM->>DB: BEGIN TRANSACTION
    ORM->>DB: INSERT INTO users
    DB-->>ORM: ユーザーID返却
    ORM->>DB: COMMIT
    ORM-->>API: Userオブジェクト
    API-->>Client: JSON レスポンス

    Note over Client,DB: Tortoise/Beanie: 自動管理でシンプル
    Client->>API: POST /users/
    API->>ORM: create() 呼び出し
    ORM->>DB: INSERT（自動トランザクション）
    DB-->>ORM: ドキュメント/レコード
    ORM-->>API: オブジェクト
    API-->>Client: JSON レスポンス

図で理解できる要点：

• SQLAlchemy はセッション管理が明示的で制御が細かい
• Tortoise ORM と Beanie はトランザクション管理が自動化されシンプル
• すべての ORM で基本的なフローは同じだが、抽象化レベルが異なる

パフォーマンス比較

各 ORM のパフォーマンス特性を理解することは重要です。

#	比較項目	SQLAlchemy	Tortoise ORM	Beanie
1	クエリ速度	★★★★☆	★★★★★	★★★★☆
2	メモリ使用量	★★★☆☆	★★★★☆	★★★★★
3	接続プール	完全サポート	サポート	Motor 経由でサポート
4	N+1 問題対策	selectinload/joinedload	prefetch_related/select_related	fetch_links
5	バルク操作	優れている	標準的	優れている（MongoDB の特性）

学習曲線の比較

開発者の学習コストも重要な判断材料となります。

#	学習項目	SQLAlchemy	Tortoise ORM	Beanie
1	初期学習時間	2-3 週間	3-5 日	2-4 日
2	公式ドキュメント	★★★★★	★★★★☆	★★★☆☆
3	コミュニティ	非常に大きい	中規模	小〜中規模
4	サンプルコード	豊富	十分	やや少ない
5	エラーメッセージ	詳細だが複雑	わかりやすい	わかりやすい

まとめ

Python の主要な ORM である SQLAlchemy、Tortoise ORM、Beanie の特徴と使い分けについて詳しく見てきました。それぞれに明確な強みがあり、プロジェクトの要件によって最適な選択肢は変わってきます。

各 ORM の推奨ケース

SQLAlchemy を選ぶべき場合：

• 大規模で複雑なプロジェクト
• 高度なクエリ最適化が必要な場合
• 同期・非同期両方の処理が必要な場合
• 長期的な保守性を重視する場合
• 豊富なエコシステムを活用したい場合

Tortoise ORM を選ぶべき場合：

• FastAPI や Sanic などの非同期フレームワークを使用する場合
• Django ORM に慣れている開発者がいる場合
• 学習コストを抑えたい場合
• 中小規模のプロジェクト
• シンプルで理解しやすいコードを重視する場合

Beanie を選ぶべき場合：

• MongoDB を使用するプロジェクト
• Pydantic を既に使用している場合
• ドキュメント指向データベースの柔軟性が必要な場合
• スキーマレスなデータ構造を扱う場合
• FastAPI との完全な型安全性を求める場合

最終的な選択のポイント

ORM の選択は、技術的な要件だけでなく、チームのスキルセットやプロジェクトの将来性も考慮する必要があります。SQLAlchemy は学習コストが高いものの、その投資は長期的に報われるでしょう。

一方、Tortoise ORM や Beanie は、素早く開発を始めたい場合や、シンプルなアプリケーションには最適な選択となります。特に FastAPI との組み合わせでは、これらの新しい ORM の恩恵を最大限に受けられますね。

どの ORM を選んでも、適切に使用すれば高品質なアプリケーションを構築できます。この記事で紹介したコード例を参考に、ぜひあなたのプロジェクトに最適な ORM を選択してください。

関連リンク

以下の公式ドキュメントとリソースが、さらなる学習に役立つでしょう。

SQLAlchemy：

• SQLAlchemy 公式ドキュメント
• SQLAlchemy 2.0 チュートリアル
• Alembic（マイグレーションツール）

Tortoise ORM：

• Tortoise ORM 公式ドキュメント
• Aerich（マイグレーションツール）
• Tortoise ORM Examples

Beanie：

• Beanie 公式ドキュメント
• Motor（非同期 MongoDB ドライバ）
• MongoDB 公式ドキュメント

FastAPI との統合：

• FastAPI 公式ドキュメント
• FastAPI with SQLAlchemy
• FastAPI with MongoDB