sqlalchemy

SQLAlchemy Core ナレッジ

詳細なナレッジ：包括的なドキュメントについては、technology: sqlalchemy を指定して mcp__documentation__fetch_docs を使用してください。

このスキルを使用しない場合

• TypeScript／Node.js プロジェクト：prisma、drizzle、または typeorm のスキルを使用してください
• Django アプリケーション：Django ORM のドキュメントを使用してください（ここでは扱いません）
• 生の SQL クエリ：SQL を直接実行するには database-query MCP サーバーを使用してください
• NoSQL データベース：MongoDB の操作には mongodb スキルを使用してください
• FastAPI 固有の場合：統合パターンについては fastapi-expert が必要になる場合があります
• データベース設計：スキーマアーキテクチャについては sql-expert または architect-expert に相談してください
• その他の Python ORM：Peewee、Pony ORM、Tortoise はこのスキルではサポートされていません

モデル定義

from sqlalchemy import Column, Integer, String, Boolean, DateTime, ForeignKey
from sqlalchemy.orm import relationship, DeclarativeBase
from datetime import datetime

class Base(DeclarativeBase):
    pass

class User(Base):
    __tablename__ = 'users'

    id = Column(Integer, primary_key=True)
    name = Column(String(100), nullable=False)
    email = Column(String(255), unique=True, nullable=False)
    is_active = Column(Boolean, default=True)
    created_at = Column(DateTime, default=datetime.utcnow)

    posts = relationship('Post', back_populates='author')

class Post(Base):
    __tablename__ = 'posts'

    id = Column(Integer, primary_key=True)
    title = Column(String(255), nullable=False)
    content = Column(String)
    author_id = Column(Integer, ForeignKey('users.id'))

    author = relationship('User', back_populates='posts')

セッション操作

from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker

engine = create_engine('postgresql://user:pass@localhost/db')
Session = sessionmaker(bind=engine)

# Create
with Session() as session:
    user = User(name='John', email='[email protected]')
    session.add(user)
    session.commit()

# Read
with Session() as session:
    users = session.query(User).all()
    user = session.query(User).filter_by(id=1).first()
    active_users = session.query(User).filter(User.is_active == True).all()

# Update
with Session() as session:
    user = session.query(User).filter_by(id=1).first()
    user.name = 'Jane'
    session.commit()

# Delete
with Session() as session:
    session.query(User).filter_by(id=1).delete()
    session.commit()

非同期処理のサポート

from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession
from sqlalchemy.orm import sessionmaker

engine = create_async_engine('postgresql+asyncpg://user:pass@localhost/db')
async_session = sessionmaker(engine, class_=AsyncSession, expire_on_commit=False)

async def get_users():
    async with async_session() as session:
        result = await session.execute(select(User))
        return result.scalars().all()

アンチパターン

アンチパターン	問題点	より良い方法
セッションを閉じない	接続リーク、コネクションプールの枯渇	コンテキストマネージャー（with Session()）を使用する
SQLAlchemy 2.0 以降で session.query() を使用する	非推奨のレガシー API	session.execute() と select() を使用する
リレーションの遅延読み込みを行う	N+1 クエリ問題	selectinload() または joinedload() を使用する
接続文字列をハードコードする	セキュリティリスク	環境変数を使用する
コネクションプールを設定しない	接続の問題、パフォーマンスの低下	pool_size、max_overflow、pool_recycle を設定する
理解せずに session.flush() を使用する	部分的なコミット、混乱	トランザクションには commit() を使用する
本番環境で pool_pre_ping を設定しない	DB の再起動後に古い接続が残る	pool_pre_ping=True を有効にする
外部キーにインデックスがない	JOIN が遅くなる	頻繁に JOIN されるカラムに Index() を追加する
一括操作に ORM を使用する	大規模なデータセットでは非常に遅い	bulk_insert_mappings() または Core を使用する
IntegrityError を処理しない	ユーザーに分かりにくいエラーが表示される	捕捉して、分かりやすいメッセージを返す

トラブルシューティング（クイックリファレンス）

問題	考えられる原因	解決策
"No module named 'sqlalchemy'"	SQLAlchemy がインストールされていない	pip install sqlalchemy を実行する
"Can't connect to server"	DATABASE_URL が正しくない、または DB が停止している	接続文字列を確認し、DB の状態を確認する
"Table doesn't exist"	マイグレーションが実行されていない	alembic upgrade head を実行する
"DetachedInstanceError"	セッションの外でリレーションにアクセスしている	joinedload() を使用するか、セッションを開いたままにする
"IntegrityError: duplicate key"	一意制約違反	既存のレコードを確認し、エラーを処理する
"InvalidRequestError: SQL expression"	2.0 スタイルの select() が不足している	session.query() ではなく select(Model) を使用する
クエリが遅い	N+1 問題、インデックスの不足	イーガーローディングを追加し、インデックスを作成する
コネクションプールのタイムアウト	接続数が多すぎる	pool_size を増やし、接続リークを修正する
"Stale data"	セッションが古いデータをキャッシュしている	session.expire_all() を呼び出すか、オブジェクトを更新する
Alembic の競合	複数のマイグレーションヘッドがある	alembic merge でブランチをマージする

本番環境への対応

エンジン設定

# database.py
from sqlalchemy import create_engine, event
from sqlalchemy.orm import sessionmaker, scoped_session
from sqlalchemy.pool import QueuePool
import os

DATABASE_URL = os.environ['DATABASE_URL']

engine = create_engine(
    DATABASE_URL,
    poolclass=QueuePool,
    pool_size=20,
    max_overflow=10,
    pool_timeout=30,
    pool_recycle=1800,  # Recycle connections after 30 minutes
    pool_pre_ping=True,  # Check connection validity
    echo=os.environ.get('DEBUG') == 'true',
    connect_args={
        'sslmode': 'require' if os.environ.get('ENV') == 'production' else 'prefer',
        'connect_timeout': 10,
    }
)

# Event listeners for monitoring
@event.listens_for(engine, 'before_cursor_execute')
def before_cursor_execute(conn, cursor, statement, parameters, context, executemany):
    conn.info.setdefault('query_start_time', []).append(time.time())

@event.listens_for(engine, 'after_cursor_execute')
def after_cursor_execute(conn, cursor, statement, parameters, context, executemany):
    total = time.time() - conn.info['query_start_time'].pop()
    if total > 0.5:  # Log slow queries
        logger.warning(f'Slow query ({total:.2f}s): {statement[:100]}')

SessionLocal = sessionmaker(bind=engine)
ScopedSession = scoped_session(SessionLocal)

コンテキストマネージャーパターン

from contextlib import contextmanager
from typing import Generator

@contextmanager
def get_db() -> Generator[Session, None, None]:
    """Database session context manager with automatic cleanup."""
    session = SessionLocal()
    try:
        yield session
        session.commit()
    except Exception:
        session.rollback()
        raise
    finally:
        session.close()

# Usage
def create_user(name: str, email: str) -> User:
    with get_db() as db:
        user = User(name=name, email=email)
        db.add(user)
        db.flush()  # Get ID without committing
        return user

非同期設定

from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession, async_sessionmaker

async_engine = create_async_engine(
    DATABASE_URL.replace('postgresql://', 'postgresql+asyncpg://'),
    pool_size=20,
    max_overflow=10,
    pool_recycle=1800,
    echo=False,
)

async_session = async_sessionmaker(
    async_engine,
    class_=AsyncSession,
    expire_on_commit=False,
)

async def get_async_db() -> AsyncGenerator[AsyncSession, None]:
    async with async_session() as session:
        try:
            yield session
            await session.commit()
        except Exception:
            await session.rollback()
            raise

クエリの最適化

from sqlalchemy.orm import joinedload, selectinload
from sqlalchemy import select

# Eager loading to avoid N+1
async def get_users_with_posts(db: AsyncSession):
    stmt = (
        select(User)
        .options(selectinload(User.posts))
        .where(User.is_active == True)
    )
    result = await db.execute(stmt)
    return result.scalars().all()

# Pagination
async def paginate(
    db: AsyncSession,
    page: int,
    per_page: int
) -> dict:
    offset = (page - 1) * per_page

    # Count
    count_stmt = select(func.count(User.id))
    total = await db.scalar(count_stmt)

    # Data
    stmt = (
        select(User)
        .order_by(User.created_at.desc())
        .offset(offset)
        .limit(per_page)
    )
    result = await db.execute(stmt)

    return {
        'data': result.scalars().all(),
        'total': total,
        'page': page,
        'pages': math.ceil(total / per_page),
    }

# Bulk insert
def bulk_insert_users(db: Session, users_data: list[dict]):
    db.execute(insert(User), users_data)
    db.commit()

Alembic マイグレーション

# alembic/env.py
from sqlalchemy import engine_from_config, pool
from alembic import context
from models import Base

target_metadata = Base.metadata

def run_migrations_online():
    connectable = engine_from_config(
        config.get_section(config.config_ini_section),
        prefix='sqlalchemy.',
        poolclass=pool.NullPool,
    )

    with connectable.connect() as connection:
        context.configure(
            connection=connection,
            target_metadata=target_metadata,
            compare_type=True,
            compare_server_default=True,
        )

        with context.begin_transaction():
            context.run_migrations()

# CLI commands
# alembic revision --autogenerate -m "Add users table"
# alembic upgrade head
# alembic downgrade -1

テスト

# conftest.py
import pytest
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker

@pytest.fixture(scope='session')
def engine():
    return create_engine(os.environ['TEST_DATABASE_URL'])

@pytest.fixture(scope='session')
def tables(engine):
    Base.metadata.create_all(engine)
    yield
    Base.metadata.drop_all(engine)

@pytest.fixture
def db(engine, tables):
    connection = engine.connect()
    transaction = connection.begin()
    Session = sessionmaker(bind=connection)
    session = Session()

    yield session

    session.close()
    transaction.rollback()
    connection.close()

# tests/test_user.py
def test_create_user(db):
    user = User(name='Test', email='[email protected]')
    db.add(user)
    db.flush()
    assert user.id is not None

監視メトリクス

メトリクス	目標
クエリ時間（p99）	< 100ms
コネクションプール接続数	< max_size
低速クエリ	0（> 500ms）
接続エラー	0

チェックリスト

• コネクションプーリングの設定
• 本番環境での SSL
• pool_pre_ping の有効化
• 長時間実行アプリでの pool_recycle
• セッション用のコンテキストマネージャー
• N+1 を防ぐイーガーローディング
• リストクエリのページネーション
• Alembic マイグレーション
• 低速クエリのログ記録
• ロールバックによるテスト分離

参考ドキュメント

• リレーションシップ
• 非同期処理