ライブラリ

Tortoise ORM

概要

Tortoise ORMは、Python向けの非同期ORMです。Django ORMにインスパイアされたAPIを持ちながら、asyncio向けにゼロから構築されており、FastAPI、Starlette、Sanic等の非同期フレームワークと最適な相性を提供します。

詳細

Tortoise ORMは、現代のPython非同期アプリケーション開発に特化して設計されたORMです。Django ORMの使いやすさを保ちながら、完全な非同期サポートを実現しており、リアルタイムアプリケーションや高スループットのWebサービス開発において威力を発揮します。PostgreSQL、MySQL、SQLiteをサポートし、型ヒント対応により開発体験も優秀です。

主な特徴

• 完全非同期: asyncio/awaitパターンによる高性能データベース操作
• Django風API: 学習しやすく直感的なインターフェース
• 型ヒント対応: 現代的なPython開発体験
• FastAPI統合: 非同期フレームワークとの最適な連携
• リアルタイム対応: WebSocketやServer-Sent Eventsとの相性抜群

メリット・デメリット

メリット

• 非同期アプリケーションでの圧倒的なパフォーマンス
• Django開発者にとって親しみやすいAPI設計
• FastAPIとの完璧な統合により現代的なWeb開発が可能
• 型ヒントによる優れた開発者体験
• リアルタイムアプリケーション開発に最適

デメリット

• 非同期プログラミングの学習コストが必要
• エコシステムがまだ発展途上
• 複雑なクエリ表現に制限がある場合がある
• 同期的なライブラリとの統合が困難

参考ページ

• Tortoise ORM 公式サイト
• Tortoise ORM ドキュメント
• Tortoise ORM GitHub

書き方の例

インストールと基本セットアップ

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

# データベースドライバー
pip install asyncpg  # PostgreSQL
pip install aiomysql  # MySQL
pip install aiosqlite  # SQLite

# FastAPI統合用
pip install fastapi tortoise-orm[fastapi]

# settings.py
TORTOISE_ORM = {
    "connections": {
        "default": "postgres://user:password@localhost:5432/mydb"
        # または SQLite: "sqlite://db.sqlite3"
        # または MySQL: "mysql://user:password@localhost:3306/mydb"
    },
    "apps": {
        "models": {
            "models": ["app.models", "aerich.models"],
            "default_connection": "default",
        },
    },
    "use_tz": False,
    "timezone": "UTC"
}

# アプリケーション初期化
from tortoise import Tortoise

async def init_db():
    await Tortoise.init(config=TORTOISE_ORM)
    await Tortoise.generate_schemas()

async def close_db():
    await Tortoise.close_connections()

基本的なCRUD操作（モデル定義、作成、読み取り、更新、削除）

# models.py
from tortoise.models import Model
from tortoise import fields
from typing import Optional

class User(Model):
    id = fields.IntField(pk=True)
    name = fields.CharField(max_length=100)
    email = fields.CharField(max_length=255, unique=True)
    is_active = fields.BooleanField(default=True)
    created_at = fields.DatetimeField(auto_now_add=True)
    updated_at = fields.DatetimeField(auto_now=True)
    
    # リレーション
    posts: fields.ReverseRelation["Post"]
    
    class Meta:
        table = "users"
        
    def __str__(self):
        return self.name

class Post(Model):
    id = fields.IntField(pk=True)
    title = fields.CharField(max_length=200)
    content = fields.TextField()
    published_at = fields.DatetimeField(null=True)
    created_at = fields.DatetimeField(auto_now_add=True)
    updated_at = fields.DatetimeField(auto_now=True)
    
    # 外部キー
    author: fields.ForeignKeyRelation[User] = fields.ForeignKeyField(
        "models.User", related_name="posts"
    )
    
    class Meta:
        table = "posts"
        
    def __str__(self):
        return self.title

# マイグレーション実行
# aerich init -t settings.TORTOISE_ORM
# aerich init-db
# aerich migrate

# 基本的なCRUD操作
async def crud_examples():
    # 作成
    user = await User.create(
        name="田中太郎",
        email="[email protected]"
    )
    
    # 読み取り
    all_users = await User.all()
    user_by_id = await User.get(id=1)
    user_by_email = await User.get(email="[email protected]")
    
    # 複数条件での検索
    active_users = await User.filter(is_active=True)
    
    # 更新
    user = await User.get(id=1)
    user.name = "田中次郎"
    await user.save()
    
    # 一括更新
    await User.filter(is_active=False).update(is_active=True)
    
    # 削除
    user = await User.get(id=1)
    await user.delete()
    
    # 一括削除
    await User.filter(is_active=False).delete()
    
    return user

高度なクエリとリレーションシップ

from tortoise.query_utils import Q
from tortoise.functions import Count, Max, Min, Avg
from datetime import datetime, timedelta

async def advanced_queries():
    # 複雑な条件クエリ
    recent_active_users = await User.filter(
        Q(is_active=True) & 
        Q(created_at__gte=datetime.now() - timedelta(days=30)) &
        Q(email__icontains="@company.com")
    ).order_by("-created_at").limit(10)
    
    # リレーションシップクエリ（Prefetch）
    users_with_posts = await User.all().prefetch_related("posts")
    
    for user in users_with_posts:
        print(f"{user.name}: {len(user.posts)} posts")
    
    # 条件付きリレーションクエリ
    users_with_recent_posts = await User.filter(
        posts__created_at__gte=datetime.now() - timedelta(days=7)
    ).distinct().prefetch_related("posts")
    
    # 集約クエリ
    user_stats = await User.annotate(
        post_count=Count("posts")
    ).filter(post_count__gt=0)
    
    # サブクエリとJOIN
    popular_authors = await User.filter(
        posts__published_at__isnull=False
    ).annotate(
        published_posts=Count("posts", _filter=Q(posts__published_at__isnull=False)),
        latest_post=Max("posts__created_at")
    ).filter(published_posts__gte=5)
    
    # 複雑な検索
    search_results = await Post.filter(
        Q(title__icontains="Python") | Q(content__icontains="Python")
    ).select_related("author")
    
    # 集約関数
    stats = await Post.all().aggregate(
        total_posts=Count("id"),
        avg_content_length=Avg("content__length"),
        latest_post=Max("created_at")
    )
    
    return recent_active_users, stats

# カスタムQuerySet
class UserQuerySet:
    def __init__(self, model_class):
        self.model_class = model_class
    
    def active(self):
        return self.model_class.filter(is_active=True)
    
    def with_posts(self):
        return self.model_class.filter(posts__isnull=False).distinct()
    
    def recent(self, days=30):
        cutoff = datetime.now() - timedelta(days=days)
        return self.model_class.filter(created_at__gte=cutoff)

# 使用例
async def custom_queries():
    queryset = UserQuerySet(User)
    active_users_with_posts = await queryset.active().with_posts().recent(7)

マイグレーションとスキーマ管理

# Aerichマイグレーションツール使用
pip install aerich

# 初期化
aerich init -t settings.TORTOISE_ORM

# 初期データベース作成
aerich init-db

# マイグレーション生成
aerich migrate --name "add_user_status"

# マイグレーション適用
aerich upgrade

# ダウングレード
aerich downgrade

# 現在のマイグレーション状態確認
aerich heads

# プログラムによるスキーマ生成
from tortoise import Tortoise

async def create_tables():
    await Tortoise.init(config=TORTOISE_ORM)
    await Tortoise.generate_schemas()

# カスタムマイグレーション
async def custom_migration():
    from tortoise import connections
    
    conn = connections.get("default")
    
    # カスタムSQL実行
    await conn.execute_query("""
        CREATE INDEX IF NOT EXISTS idx_users_email_active 
        ON users (email, is_active)
    """)
    
    await conn.execute_query("""
        CREATE INDEX IF NOT EXISTS idx_posts_author_published 
        ON posts (author_id, published_at) 
        WHERE published_at IS NOT NULL
    """)

# データシーディング
async def seed_data():
    # テストデータ作成
    users_data = [
        {"name": "ユーザー1", "email": "[email protected]"},
        {"name": "ユーザー2", "email": "[email protected]"},
        {"name": "ユーザー3", "email": "[email protected]"},
    ]
    
    users = await User.bulk_create([User(**data) for data in users_data])
    
    # 関連データ作成
    posts_data = []
    for user in users:
        for i in range(3):
            posts_data.append(Post(
                title=f"{user.name}の投稿{i+1}",
                content=f"これは{user.name}による投稿です。",
                author=user,
                published_at=datetime.now() if i % 2 == 0 else None
            ))
    
    await Post.bulk_create(posts_data)

パフォーマンス最適化と高度な機能

from tortoise.transactions import in_transaction, atomic
from tortoise.expressions import F

async def performance_optimization():
    # トランザクション使用
    async with in_transaction() as connection:
        user = await User.create(
            name="山田花子",
            email="[email protected]",
            using_db=connection
        )
        
        post = await Post.create(
            title="最初の投稿",
            content="Tortoise ORMを使った投稿です",
            author=user,
            published_at=datetime.now(),
            using_db=connection
        )
    
    # アトミック操作（デコレータ使用）
    @atomic()
    async def create_user_with_posts(name: str, email: str, post_titles: list):
        user = await User.create(name=name, email=email)
        
        posts = []
        for title in post_titles:
            posts.append(Post(
                title=title,
                content=f"Content for {title}",
                author=user
            ))
        
        await Post.bulk_create(posts)
        return user
    
    # バッチ操作
    await User.bulk_create([
        User(name="ユーザー1", email="[email protected]"),
        User(name="ユーザー2", email="[email protected]"),
        User(name="ユーザー3", email="[email protected]"),
    ])
    
    # バッチ更新
    await User.filter(is_active=False).update(is_active=True)
    
    # F式による原子的更新
    await Post.filter(id=1).update(view_count=F("view_count") + 1)
    
    # 接続プール設定
    TORTOISE_ORM = {
        "connections": {
            "default": {
                "engine": "tortoise.backends.asyncpg",
                "credentials": {
                    "host": "localhost",
                    "port": 5432,
                    "user": "postgres",
                    "password": "password",
                    "database": "mydb",
                    "minsize": 1,
                    "maxsize": 10,
                    "max_queries": 50000,
                    "max_inactive_connection_lifetime": 300,
                }
            }
        },
        "apps": {
            "models": {
                "models": ["app.models"],
                "default_connection": "default",
            }
        }
    }

# カスタムフィールド
class TimestampMixin:
    created_at = fields.DatetimeField(auto_now_add=True)
    updated_at = fields.DatetimeField(auto_now=True)

class User(Model, TimestampMixin):
    id = fields.IntField(pk=True)
    name = fields.CharField(max_length=100)
    email = fields.CharField(max_length=255, unique=True)
    
    # カスタムプロパティ
    @property
    def display_name(self) -> str:
        return f"{self.name} ({self.email})"
    
    # カスタムメソッド
    async def get_recent_posts(self, days: int = 30) -> list:
        cutoff = datetime.now() - timedelta(days=days)
        return await self.posts.filter(created_at__gte=cutoff)

フレームワーク統合と実用例

# FastAPI統合
from fastapi import FastAPI, Depends, HTTPException
from fastapi.responses import JSONResponse
from tortoise.contrib.fastapi import register_tortoise
from pydantic import BaseModel
from typing import List

app = FastAPI()

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

class UserResponse(BaseModel):
    id: int
    name: str
    email: str
    is_active: bool
    created_at: datetime
    
    class Config:
        orm_mode = True

class PostResponse(BaseModel):
    id: int
    title: str
    content: str
    published_at: Optional[datetime]
    author: UserResponse
    
    class Config:
        orm_mode = True

# API エンドポイント
@app.get("/users/", response_model=List[UserResponse])
async def get_users(skip: int = 0, limit: int = 10):
    users = await User.all().offset(skip).limit(limit)
    return users

@app.get("/users/{user_id}", response_model=UserResponse)
async def get_user(user_id: int):
    user = await User.get_or_none(id=user_id)
    if not user:
        raise HTTPException(status_code=404, detail="User not found")
    return user

@app.post("/users/", response_model=UserResponse)
async def create_user(user: UserCreate):
    try:
        db_user = await User.create(**user.dict())
        return db_user
    except Exception as e:
        raise HTTPException(status_code=400, detail=str(e))

@app.get("/users/{user_id}/posts/", response_model=List[PostResponse])
async def get_user_posts(user_id: int):
    user = await User.get_or_none(id=user_id)
    if not user:
        raise HTTPException(status_code=404, detail="User not found")
    
    posts = await Post.filter(author=user).select_related("author")
    return posts

# WebSocket統合
from fastapi import WebSocket, WebSocketDisconnect
import json

class ConnectionManager:
    def __init__(self):
        self.active_connections: List[WebSocket] = []

    async def connect(self, websocket: WebSocket):
        await websocket.accept()
        self.active_connections.append(websocket)

    def disconnect(self, websocket: WebSocket):
        self.active_connections.remove(websocket)

    async def broadcast(self, message: str):
        for connection in self.active_connections:
            await connection.send_text(message)

manager = ConnectionManager()

@app.websocket("/ws")
async def websocket_endpoint(websocket: WebSocket):
    await manager.connect(websocket)
    try:
        while True:
            data = await websocket.receive_text()
            # リアルタイムでデータベース操作
            message_data = json.loads(data)
            
            if message_data["type"] == "create_post":
                post = await Post.create(
                    title=message_data["title"],
                    content=message_data["content"],
                    author_id=message_data["author_id"]
                )
                
                # 全クライアントに新投稿をブロードキャスト
                await manager.broadcast(json.dumps({
                    "type": "new_post",
                    "post": {
                        "id": post.id,
                        "title": post.title,
                        "content": post.content,
                        "created_at": post.created_at.isoformat()
                    }
                }))
                
    except WebSocketDisconnect:
        manager.disconnect(websocket)

# バックグラウンドタスク
from fastapi import BackgroundTasks

async def send_notification_email(user_id: int):
    user = await User.get(id=user_id)
    # メール送信ロジック
    print(f"Sending notification to {user.email}")

@app.post("/users/{user_id}/notify/")
async def notify_user(user_id: int, background_tasks: BackgroundTasks):
    background_tasks.add_task(send_notification_email, user_id)
    return {"message": "Notification scheduled"}

# Tortoise ORM設定
register_tortoise(
    app,
    config=TORTOISE_ORM,
    generate_schemas=True,
    add_exception_handlers=True,
)

# Starlette統合
from starlette.applications import Starlette
from starlette.responses import JSONResponse
from starlette.routing import Route

async def list_users(request):
    users = await User.all()
    return JSONResponse([{
        "id": user.id,
        "name": user.name,
        "email": user.email
    } for user in users])

async def create_user(request):
    data = await request.json()
    user = await User.create(**data)
    return JSONResponse({
        "id": user.id,
        "name": user.name,
        "email": user.email
    })

starlette_app = Starlette(routes=[
    Route("/users", list_users, methods=["GET"]),
    Route("/users", create_user, methods=["POST"]),
])

# アプリケーション起動時の初期化
@app.on_event("startup")
async def startup_event():
    await init_db()
    await seed_data()  # 開発環境でのみ

@app.on_event("shutdown")
async def shutdown_event():
    await close_db()

# テスト統合
import pytest
from tortoise.contrib.test import finalizer, initializer

@pytest.fixture(scope="module")
def event_loop():
    loop = asyncio.get_event_loop_policy().new_event_loop()
    yield loop
    loop.close()

@pytest.fixture(scope="module", autouse=True)
async def initialize_tests(request):
    await initializer(["app.models"])
    request.addfinalizer(finalizer)

@pytest.mark.asyncio
async def test_create_user():
    user = await User.create(name="Test User", email="[email protected]")
    assert user.name == "Test User"
    assert user.email == "[email protected]"
    assert user.is_active is True

@pytest.mark.asyncio
async def test_user_posts_relationship():
    user = await User.create(name="Author", email="[email protected]")
    post = await Post.create(
        title="Test Post",
        content="Test content",
        author=user
    )
    
    await user.fetch_related("posts")
    assert len(user.posts) == 1
    assert user.posts[0].title == "Test Post"