building-with-sqlalchemy-orm

SQLAlchemy ORM を使った構築

SQLAlchemy ORM 2.0 以降、汎用的な PostgreSQL のパターン、Neon 固有のサーバーレスに関する考慮事項を活用して、本番環境に対応したデータベースアプリケーションを構築します。

実装前

実装を成功させるために、コンテキストを収集します。

ソース	収集する情報
コードベース	既存のモデル、データベースのセットアップ、接続パターン
会話	学生の具体的なユースケース（何を構築するのか）、制約
スキルのリファレンス	references/ にあるドメインパターン（API ドキュメント、ベストプラクティス、アーキテクチャ）
ユーザーガイドライン	プロジェクトの規約、習熟度

学生には本人の要件のみを尋ねてください（ドメインの専門知識はこのスキルに組み込まれています）。

---

ペルソナ

あなたは、SQLAlchemy ORM を使用したアプリケーション構築の本番環境での経験を持つ、Python データベースアーキテクトです。汎用的な PostgreSQL のパターン（どこでも適用可能）と Neon 固有のサーバーレスに関する考慮事項（自動スケーリング、ゼロへのスケール、ブランチ作成）の両方を理解しています。適切なトランザクション処理、リレーションシップ、コネクションプーリングを備えた、複数テーブルのアプリケーションを構築してきました。

---

使用する場面

• 要件からデータベースモデルを構築する場合（テーブルを Python クラスとして定義する）
• トランザクションを使って安全に CRUD 操作を実装する場合
• テーブル間のリレーションシップを管理する場合（外部キー、結合）
• フィルター、並べ替え、複雑な結合を使ってデータをクエリする場合
• 適切な設定で PostgreSQL または Neon に接続する場合
• 永続化を学ぶ初心者にデータベースの基礎を教える場合

---

基本概念

1. クラスとしてのモデル（ORM 抽象化）

SQLAlchemy は Python クラスをデータベースのテーブルにマッピングします。

from sqlalchemy import Column, Integer, String, Float, ForeignKey
from sqlalchemy.orm import declarative_base

Base = declarative_base()

class Expense(Base):
    __tablename__ = 'expenses'

    id = Column(Integer, primary_key=True)
    description = Column(String(200))
    amount = Column(Float)
    category_id = Column(Integer, ForeignKey('categories.id'))

重要な理由：Python を記述すれば、SQLAlchemy が SQL を生成します。SQL を手作業で書く必要はありません。

2. トランザクションとしてのセッション（Unit of Work）

セッションはデータベース操作をひとつの不可分なトランザクションにまとめます。

with Session(engine) as session:
    new_expense = Expense(description="Groceries", amount=45.50, category_id=1)
    session.add(new_expense)
    session.commit()  # All or nothing

重要な理由：何かが失敗すると、何もコミットされません。データベースの一貫性が保証されます。

3. リレーションシップ（ナビゲーションとしての外部キー）

手動で結合する代わりに、Python でリレーションシップを定義します。

class Category(Base):
    __tablename__ = 'categories'
    id = Column(Integer, primary_key=True)
    name = Column(String(50))
    expenses = relationship("Expense", back_populates="category")

class Expense(Base):
    __tablename__ = 'expenses'
    id = Column(Integer, primary_key=True)
    category = relationship("Category", back_populates="expenses")

使用例：

category = session.query(Category).first()
print(category.expenses)  # All expenses in this category

4. クエリ（フィルタリング、並べ替え、結合）

生の SQL を書かずに、安全にクエリを構築します。

# Filter: expenses > $50
expensive = session.query(Expense).filter(Expense.amount > 50).all()

# Order: sorted by amount descending
sorted_expenses = session.query(Expense).order_by(Expense.amount.desc()).all()

# Join: expenses with their categories
results = session.query(Expense, Category).join(Category).all()

5. Neon 接続に関する詳細

Neon は自動スケーリングとブランチ作成に対応したサーバーレス PostgreSQL です。主な違いは次のとおりです。

• 接続文字列：postgresql+psycopg2://user:pass@host/dbname?sslmode=require
• SSL を常に使用：?sslmode=require（Neon では必須です）
• 環境変数：認証情報は .env に保存します（ハードコードしないでください）
• 自動一時停止：Neon はアイドル状態になるとコンピュートを一時停止します。コネクションプールが役立ちます

---

判断ロジック

シナリオ	パターン	理由
初めてのデータベースモデル	単一テーブル、ひとつの Column 型	リレーションシップを導入する前に、最もシンプルなメンタルモデルを学べます
データを関連付けたい	relationship() + ForeignKey を使用	ORM が複雑な結合を処理します
同時リクエストが多い	pool_size を使ったコネクションプーリング	Neon のコンピュートのスケーリングを最大限に活用できます
データの一貫性が重要	try/except を使ったトランザクション	エラー時にロールバックし、不可分性を保証します
ゼロへのスケールを実現したい	Neon サーバーレス + echo_pool を使ったプール	アイドル時は自動一時停止し、最初のリクエストで再開します
クエリをデバッグする	エンジンで echo=True を有効にする	生成された SQL を確認できます

---

ワークフロー：予算管理ツールの構築

ステップ 1：モデルを定義する

from sqlalchemy import create_engine, Column, Integer, String, Float, DateTime, ForeignKey, func
from sqlalchemy.orm import declarative_base, relationship, Session
from datetime import datetime

Base = declarative_base()

class Category(Base):
    __tablename__ = 'categories'
    id = Column(Integer, primary_key=True)
    name = Column(String(50), unique=True)
    expenses = relationship("Expense", back_populates="category")

class Expense(Base):
    __tablename__ = 'expenses'
    id = Column(Integer, primary_key=True)
    description = Column(String(200))
    amount = Column(Float)
    date = Column(DateTime, default=datetime.utcnow)
    category_id = Column(Integer, ForeignKey('categories.id'))
    category = relationship("Category", back_populates="expenses")

ステップ 2：エンジンとテーブルを作成する

import os
from dotenv import load_dotenv

load_dotenv()

# Connection string from environment
DATABASE_URL = os.getenv("DATABASE_URL")
# Format: postgresql+psycopg2://user:password@host/dbname?sslmode=require

engine = create_engine(DATABASE_URL)
Base.metadata.create_all(engine)

ステップ 3：CRUD を実装する

def create_expense(session, description, amount, category_id):
    """Create a new expense."""
    try:
        expense = Expense(
            description=description,
            amount=amount,
            category_id=category_id
        )
        session.add(expense)
        session.commit()
        return expense
    except Exception as e:
        session.rollback()
        print(f"Error creating expense: {e}")
        return None

def read_expenses(session, category_id=None):
    """Read expenses, optionally filtered by category."""
    query = session.query(Expense)
    if category_id:
        query = query.filter(Expense.category_id == category_id)
    return query.all()

def update_expense(session, expense_id, amount=None, description=None):
    """Update an expense."""
    expense = session.query(Expense).filter(Expense.id == expense_id).first()
    if expense:
        if amount is not None:
            expense.amount = amount
        if description is not None:
            expense.description = description
        session.commit()
        return expense
    return None

def delete_expense(session, expense_id):
    """Delete an expense."""
    expense = session.query(Expense).filter(Expense.id == expense_id).first()
    if expense:
        session.delete(expense)
        session.commit()
        return True
    return False

ステップ 4：リレーションシップを使ってクエリする

# Get all expenses for a category
category = session.query(Category).filter_by(name="Food").first()
print(category.expenses)  # Uses relationship

# Total spent by category
totals = session.query(
    Category.name,
    func.sum(Expense.amount).label('total')
).join(Expense).group_by(Category.name).all()

for name, total in totals:
    print(f"{name}: ${total:.2f}")

ステップ 5：トランザクションを安全に処理する

def transfer_expense(session, expense_id, new_category_id):
    """Move expense to different category (must succeed fully or not at all)."""
    try:
        expense = session.query(Expense).filter(Expense.id == expense_id).first()
        if not expense:
            raise ValueError(f"Expense {expense_id} not found")

        expense.category_id = new_category_id
        session.commit()
        return True
    except Exception as e:
        session.rollback()
        print(f"Transaction failed, rolled back: {e}")
        return False

ステップ 6：Neon に接続する

環境ファイル（.env）：

DATABASE_URL=postgresql+psycopg2://user:[email protected]/dbname?sslmode=require

プール設定を使った接続：

from sqlalchemy.pool import QueuePool

engine = create_engine(
    DATABASE_URL,
    poolclass=QueuePool,
    pool_size=5,
    max_overflow=10,
    pool_pre_ping=True,  # Verify connections before use
    echo=False  # Set to True for debugging
)

---

MCP 統合

SQLAlchemy のデータベース操作を AI エージェントに接続するには、次のようにします。

# Define an MCP tool that the agent can use
def query_expenses_by_category(category_name: str) -> list:
    """Agent can ask: 'How much did I spend on groceries?'"""
    with Session(engine) as session:
        return session.query(Expense).join(Category).filter(
            Category.name == category_name
        ).all()

def summarize_spending(start_date, end_date) -> dict:
    """Agent can generate reports."""
    with Session(engine) as session:
        return session.query(
            Category.name,
            func.sum(Expense.amount).label('total'),
            func.count(Expense.id).label('count')
        ).join(Expense).filter(
            Expense.date.between(start_date, end_date)
        ).group_by(Category.name).all()

これらを MCP ツールとして登録すると、エージェント（Budget Manager）が利用できるようになります。

---

安全性とガードレール

絶対にしないでください

• ❌ Python ファイルに認証情報をハードコードしないでください
• ❌ トランザクションのエラー処理を省略しないでください
• ❌ 検証せずにユーザー入力を信頼しないでください
• ❌ シークレットを git にコミットしないでください
• ❌ 本番環境でコネクションプーリングを省略しないでください

必ず実施してください

• ✅ 接続文字列には環境変数（.env ファイル）を使用してください
• ✅ トランザクションを try/except ブロックで囲み、ロールバックしてください
• ✅ データベース操作の前に、すべてのユーザー入力を検証してサニタイズしてください
• ✅ session.commit() を明示的に使用してください（自動コミットは使用しないでください）
• ✅ エラー発生時は session.rollback() を使用してください
• ✅ 接続の状態を確認するため、pool_pre_ping=True を有効にしてください
• ✅ Neon では ?sslmode=require を使用してください（いずれにしても強制されます）

よくある間違い

間違い	影響	対処方法
session.commit() の呼び忘れ	変更が保存されない	必ず commit() を呼び出すか、コンテキストマネージャーを使用してください
エラー時にロールバックしない	データベースに不完全なデータが残る	try/except で囲み、rollback() を実行してください
認証情報のハードコード	セキュリティ侵害	環境変数を使用してください
コネクションプーリングを設定しない	Neon のコンピューティングリソースのスケーリングが非効率になる	pool_size パラメーターを設定してください
クエリでユーザー入力をそのまま使用する	SQL インジェクション	パラメーター化クエリを使用してください（ORM が対応します）

---

予算管理のサンプル（完全版）

動作する予算管理アプリケーションの完全版は、references/budget-tracker-complete.py を参照してください。以下が含まれています。

• モデル定義
• データベースのセットアップ
• CRUD 関数
• トランザクション処理
• Neon 接続
• 使用例

---

トラブルシューティング

問題	原因	解決策
ModuleNotFoundError: No module named 'sqlalchemy'	インストールされていない	pip install sqlalchemy または uv add sqlalchemy
ModuleNotFoundError: No module named 'psycopg2'	PostgreSQL ドライバーがない	pip install psycopg2-binary または uv add psycopg2-binary
OperationalError: could not connect to server	接続文字列が間違っているか、Neon がオフライン	DATABASE_URL の形式を確認し、Neon プロジェクトが稼働中か確認してください
IntegrityError: duplicate key value	一意なフィールドに重複した値を挿入している	値がすでに存在するか確認し、代わりに更新してください
ForeignKeyError: could not create foreign key	カテゴリーが存在しない	先にカテゴリーを作成するか、有効な category_id を使用してください
クエリが遅い	インデックスがない、リレーションシップが不足している	インデックス設定のパターンについては references/architecture.md を確認してください

---

リソース

• 公式ドキュメント：https://docs.sqlalchemy.org/en/20/orm/quickstart.html
• Neon のドキュメント：https://neon.com/docs/
• PostgreSQL のデータ型：https://www.postgresql.org/docs/current/datatype.html