sqlmodel-expert

SQLModel エキスパート

本番データベース向けの高度な SQLModel パターンと包括的な Alembic マイグレーション。

クイックスタート

基本モデルの定義

from sqlmodel import Field, SQLModel
from typing import Optional
from datetime import datetime

class Task(SQLModel, table=True):
    id: Optional[int] = Field(default=None, primary_key=True)
    title: str = Field(index=True)
    description: Optional[str] = None
    completed: bool = Field(default=False)
    created_at: datetime = Field(default_factory=datetime.utcnow)

データベースの初期化

# Using provided script
python scripts/init_db.py --url postgresql://user:pass@localhost/db

# Or manually
from sqlmodel import create_engine
engine = create_engine("postgresql://user:pass@localhost/db")
SQLModel.metadata.create_all(engine)

マイグレーションの作成

# Using provided helper script
./scripts/migrate.sh create "add user table"

# Or directly with Alembic
alembic revision --autogenerate -m "add user table"
alembic upgrade head

主なトピック

1. 高度なモデルパターン

参照先： references/advanced-models.md

• リレーションシップ：一対多、多対多、自己参照
• 継承：単一テーブル、結合テーブル、ポリモーフィズム
• バリデーション：Pydantic バリデーター、カスタム制約
• ミックスイン：タイムスタンプ、論理削除、再利用可能なパターン
• フィールド型：列挙型、JSON、配列、カスタム型
• インデックス：単一、複合、部分インデックス
• 制約：一意制約、チェック制約、外部キーのカスケード

2. 包括的なマイグレーション

参照先： references/migrations.md

• Alembic のセットアップ：SQLModel 向けの設定、env.py
• マイグレーションの作成：自動生成と手動作成
• スキーマの変更：列の追加・削除、名前変更、型変更
• データマイグレーション：複雑なデータ変換
• 本番ワークフロー：ダウンタイムなしのマイグレーション
• ロールバック戦略：安全なダウングレードのパターン
• トラブルシューティング：よくある問題と解決策

3. クエリの最適化

参照先： references/queries-optimization.md

• N+1 問題：イーガーローディングによる解決策
• クエリパターン：結合、集計、サブクエリ
• パフォーマンス：インデックス、バッチ処理、プロファイリング
• 高度なクエリ：ウィンドウ関数、CTE
• 一括操作：大規模な挿入、更新、削除
• テスト：クエリ数の計測、実行計画の分析

よく使われるパターン

一対多リレーションシップ

from typing import List
from sqlmodel import Field, Relationship, SQLModel

class Team(SQLModel, table=True):
    id: Optional[int] = Field(default=None, primary_key=True)
    name: str

    # One team has many heroes
    heroes: List["Hero"] = Relationship(back_populates="team")

class Hero(SQLModel, table=True):
    id: Optional[int] = Field(default=None, primary_key=True)
    name: str
    team_id: Optional[int] = Field(foreign_key="team.id")

    # Many heroes belong to one team
    team: Optional[Team] = Relationship(back_populates="heroes")

リンクテーブルを使った多対多リレーションシップ

class HeroTeamLink(SQLModel, table=True):
    hero_id: int = Field(foreign_key="hero.id", primary_key=True)
    team_id: int = Field(foreign_key="team.id", primary_key=True)
    joined_at: datetime = Field(default_factory=datetime.utcnow)

class Hero(SQLModel, table=True):
    id: Optional[int] = Field(default=None, primary_key=True)
    name: str
    teams: List["Team"] = Relationship(
        back_populates="heroes",
        link_model=HeroTeamLink
    )

class Team(SQLModel, table=True):
    id: Optional[int] = Field(default=None, primary_key=True)
    name: str
    heroes: List[Hero] = Relationship(
        back_populates="teams",
        link_model=HeroTeamLink
    )

N+1 クエリ問題の解決

from sqlalchemy.orm import selectinload

# BAD - N+1 queries
users = session.exec(select(User)).all()
for user in users:
    posts = user.posts  # Each triggers a query!

# GOOD - Eager loading (2 queries total)
statement = select(User).options(selectinload(User.posts))
users = session.exec(statement).all()
for user in users:
    posts = user.posts  # No additional query!

マイグレーションの作成

# 1. Modify your model
class User(SQLModel, table=True):
    id: Optional[int] = Field(default=None, primary_key=True)
    email: str
    phone: str  # New field added

# 2. Generate migration
# alembic revision --autogenerate -m "add phone to user"

# 3. Review generated migration
def upgrade() -> None:
    op.add_column('user', sa.Column('phone', sa.String(), nullable=True))

def downgrade() -> None:
    op.drop_column('user', 'phone')

# 4. Apply migration
# alembic upgrade head

マイグレーション用ヘルパースクリプト

データベースの初期化

python scripts/init_db.py --url postgresql://user:pass@localhost/db

マイグレーション操作

./scripts/migrate.sh init              # Initialize Alembic
./scripts/migrate.sh create "message"  # Create migration
./scripts/migrate.sh upgrade           # Apply migrations
./scripts/migrate.sh downgrade         # Rollback one
./scripts/migrate.sh current           # Show current
./scripts/migrate.sh history           # Show history
./scripts/migrate.sh test              # Test up & down

モデルの例

テンプレートとして assets/example-models.py のモデル例を使用します。

• タイムスタンプミックスインを使った User モデル
• 列挙型とリレーションシップを使った Task モデル
• 多対多リレーションシップを使った Team モデル
• リンクテーブルを使ったタグシステム
• 読み取り、書き込み、更新用の分離モデル

プロジェクトにコピーします。

cp assets/example-models.py your-project/app/models.py

ベストプラクティスのチェックリスト

モデル設計

• すべてのフィールドに型ヒントを使用する
• 読み取り、書き込み、更新用のモデルを分ける
• 共通フィールド（タイムスタンプ、論理削除）にはミックスインを使用する
• 外部キーと、頻繁にクエリされる列にインデックスを定義する
• 選択肢が制限される項目には列挙型を使用する
• Pydantic バリデーターで適切なバリデーションを実装する

リレーションシップ

• 双方向リレーションシップには back_populates を使用する
• 多対多リレーションシップには明示的なリンクテーブルを作成する
• カスケード削除の動作を検討する
• N+1 クエリを防ぐため、イーガーローディングを使用する
• 外部キー列にインデックスを設定する

マイグレーション

• 自動生成されたマイグレーションを必ず確認する
• 1つのマイグレーションにつき、論理的な変更は1つにする
• アップグレードとダウングレードの両方をテストする
• 内容がわかるマイグレーション名を付ける
• 適用済みのマイグレーションは編集しない
• スキーマの変更時にはデータマイグレーションを追加する
• 本番マイグレーションの前にデータベースをバックアップする

クエリの最適化

• リレーションシップにはイーガーローディング（selectinload）を使用する
• 必要な列だけを選択する
• WHERE/ORDER BY の列にインデックスを使用する
• ループの代わりにバッチ処理を使用する
• 遅いクエリをプロファイリングする
• コネクションプーリングを使用する

トラブルシューティングガイド

マイグレーションの問題

問題：Alembic がモデルの変更を検出しない

# Solution: Ensure models are imported in env.py
from app.models import User, Task, Team  # Import all models
target_metadata = SQLModel.metadata

問題：マイグレーションに失敗する

# Check current state
alembic current

# Manually fix issue, then stamp
alembic stamp head

# Or downgrade and retry
alembic downgrade -1
alembic upgrade head

クエリのパフォーマンス

問題：クエリが遅い

# Enable query logging
logging.getLogger('sqlalchemy.engine').setLevel(logging.INFO)

# Use EXPLAIN ANALYZE
explain = session.exec(text("EXPLAIN ANALYZE SELECT ...")).all()

# Profile queries
# See references/queries-optimization.md for detailed patterns

問題：N+1 クエリ

# Use selectinload
statement = select(User).options(selectinload(User.posts))

# Or joinedload
from sqlalchemy.orm import joinedload
statement = select(User).options(joinedload(User.posts))

本番ワークフロー

開発

1. SQLModel モデルを変更します
2. マイグレーションを生成します：./scripts/migrate.sh create "description"
3. 生成されたマイグレーションファイルを確認します
4. マイグレーションをテストします：./scripts/migrate.sh test
5. マイグレーションファイルをコミットします

ステージング

1. アプリケーションコードをデプロイします
2. マイグレーションを実行します：alembic upgrade head
3. データの整合性を確認します
4. アプリケーションをテストします

本番環境

1. データベースをバックアップします：pg_dump mydb > backup.sql
2. メンテナンス時間帯にデプロイします
3. マイグレーションを実行します：alembic upgrade head
4. ログとメトリクスを監視します
5. アプリケーションの機能を確認します

ゼロダウンタイムのマイグレーション戦略

大規模な本番データベースの場合：

# Phase 1: Add new column (nullable)
def upgrade():
    op.add_column('user', sa.Column('new_email', sa.String(), nullable=True))

# Deploy app version that writes to both columns

# Phase 2: Backfill data
def upgrade():
    op.execute("UPDATE user SET new_email = email WHERE new_email IS NULL")

# Phase 3: Make non-nullable
def upgrade():
    op.alter_column('user', 'new_email', nullable=False)

# Deploy app version that reads from new column

# Phase 4: Drop old column
def upgrade():
    op.drop_column('user', 'email')

追加リソース

• 高度なパターン：継承、ポリモーフィズム、複合キーについては references/advanced-models.md を参照してください
• マイグレーションガイド：Alembic の習得については references/migrations.md を参照してください
• クエリの最適化：パフォーマンスのチューニングについては references/queries-optimization.md を参照してください

このスキルには、プロフェッショナルな SQLModel の開発とデータベース管理に必要なすべてが含まれています。