Pythonで始めるTortoise ORM入門：初心者向け解説

PythonのWebアプリケーション開発において、データベース操作を効率的に行うためのORM（Object Relational Mapping）は欠かせません。

本記事では、軽量かつシンプルなORMライブラリである Tortoise ORM の基本から応用までを、初心者の方にもわかりやすく解説します。

Tortoise ORMとは？

Tortoise ORMは、Pythonの**非同期（async/await）**に対応したORMライブラリです。

主要な特徴は以下のとおりです。

  • 完全非同期対応：async/await構文でシームレスなDBアクセス
  • シンプルなAPI：Django ORMライクな直感的なモデル定義
  • 軽量：余計な依存を持たず、導入が容易
  • マルチデータベース対応：SQLite、MySQL、PostgreSQLなどをサポート

これらにより、FastAPIやStarletteなどの非同期フレームワークと組み合わせて使いやすいのが魅力です。

環境構築とインストール

# Python3.8+ が必要
# 仮想環境の作成
python -m venv .venv
source .venv/bin/activate  # Windows: .\.venv\\Scripts\\activate

# Tortoise ORM のインストール
pip install tortoise-orm

依存関係としては、ドライバ（例: asyncpg for PostgreSQL）が必要であれば別途インストールします。

# PostgreSQL用ドライバを追加
pip install asyncpg

モデルの定義

ベースモデル

Tortoise ORMでは、Model クラスを継承してテーブルを定義します。

from tortoise import fields, models

class User(models.Model):
    id = fields.IntField(pk=True)
    username = fields.CharField(max_length=20, unique=True)
    email = fields.CharField(max_length=255)
    is_active = fields.BooleanField(default=True)
    created_at = fields.DatetimeField(auto_now_add=True)

    class Meta:
        table = "users"

    def __str__(self):
        return self.username

データベース接続設定

from tortoise import Tortoise, run_async

async def init_db():
    await Tortoise.init(
        db_url="sqlite://db.sqlite3",
        modules={"models": ["__main__"]}
    )
    # テーブルを自動生成
    await Tortoise.generate_schemas()

if __name__ == "__main__":
    run_async(init_db())

CRUD操作

実際の操作にはすべて非同期関数（async def）を使います。

作成（Create）

async def create_user():
    user = await User.create(
        username="alice",
        email="alice@example.com"
    )
    print(f"作成: {user.id} - {user.username}")

読み取り（Read）

# 単一取得
user = await User.get(id=1)
# 条件検索
users = await User.filter(is_active=True).all()

更新（Update）

user = await User.get(username="alice")
user.email = "alice_new@example.com"
await user.save()

削除（Delete）

user = await User.get(id=1)
await user.delete()

関連（リレーション）の設定

One-to-Many

class Blog(models.Model):
    id = fields.IntField(pk=True)
    title = fields.CharField(max_length=100)
    author = fields.ForeignKeyField("models.User", related_name="blogs")
# ブログ取得時に作者情報を取得（プリフェッチ）
blogs = await Blog.all().prefetch_related("author")

Many-to-Many

class Tag(models.Model):
    id = fields.IntField(pk=True)
    name = fields.CharField(max_length=20, unique=True)

class Post(models.Model):
    id = fields.IntField(pk=True)
    title = fields.CharField(max_length=100)
    tags = fields.ManyToManyField("models.Tag", related_name="posts")
post = await Post.get(id=1)
await post.tags.add(tag1, tag2)

非同期操作とトランザクション

from tortoise.transactions import in_transaction

async def transfer_points(sender_id, receiver_id, amount):
    async with in_transaction() as conn:
        sender = await User.get(id=sender_id).using_db(conn)
        receiver = await User.get(id=receiver_id).using_db(conn)
        sender.points -= amount
        receiver.points += amount
        await sender.save(using_db=conn)
        await receiver.save(using_db=conn)

ベストプラクティスとトラブルシューティング

  • モデル間のリレーションは related_name を明示的に設定すると可読性が向上します。
  • テーブル生成後にモデルを変更した場合、マイグレーションツール（Aerichなど）の導入を推奨します。
  • 非同期実行時は、必ず run_async や asyncio.run でエントリポイントをラップしてください。
  • 大量データの一括操作には bulk_create、bulk_update を活用すると効率的です。

演習問題

問題1

新しいモデル Category を定義し、Post モデルに ForeignKey でカテゴリを追加してください。

問題2

すべての User のうち、作成日 (created_at) が30日以内のユーザーを取得するクエリを作成してください。

問題3

2つのユーザー間でポイントを安全にやり取りするトランザクション関数を実装してください。

解答例

解答1 
class Category(models.Model):
    id = fields.IntField(pk=True)
    name = fields.CharField(max_length=50, unique=True)

class Post(models.Model):
    # 既存フィールド...
    category = fields.ForeignKeyField("models.Category", related_name="posts")
解答2 
from datetime import datetime, timedelta

thirty_days_ago = datetime.utcnow() - timedelta(days=30)
recent_users = await User.filter(created_at__gte=thirty_days_ago).all()
解答3 
from tortoise.transactions import in_transaction

async def transfer_points(sender_id, receiver_id, amount):
    async with in_transaction() as conn:
        sender = await User.get(id=sender_id).using_db(conn)
        receiver = await User.get(id=receiver_id).using_db(conn)
        if sender.points < amount:
            raise ValueError("残高不足です。")
        sender.points -= amount
        receiver.points += amount
        await sender.save(using_db=conn)
        await receiver.save(using_db=conn)

まとめ

本記事では、Pythonで使える軽量ORMライブラリ Tortoise ORM の導入からモデル定義、CRUDや関連設定、非同期操作、ベストプラクティスまでを解説しました。

演習問題に挑戦し、自分のプロジェクトでもぜひ活用してみてください！