database-management

データベース管理スキル

有効化するタイミング

次の場合にこのスキルを有効化します：

• データベース機能をセットアップするとき
• データベーススキーマを作成するとき
• データ永続化を実装するとき
• データベースクエリを記述するとき
• SQLite またはその他のデータベースを使用するとき

重要：データベース機能が必要なすべてのプロジェクトで、このスキルは必須です。

コアアーキテクチャ

原則

1. SQLite のみ：デフォルトのデータベースとして SQLite を使用します
2. 単一インターフェース：すべてのデータベース操作を database.py 経由で行います
3. SQL の完全な分離：すべての SQL 文を database.py に記述します
4. 関数ベース：シンプルで再利用可能な関数インターフェースを使用します

ファイル構成

project/
├── database.py         # ALL SQL lives here
├── app.py              # Uses database functions (no SQL!)
└── tests/
    └── test_database.py

標準インターフェースのパターン

# database.py - All database code lives here
import sqlite3
from typing import List, Dict, Optional, Any

DB_PATH = "app.db"

def get_connection(db_path: str = DB_PATH) -> sqlite3.Connection:
    conn = sqlite3.connect(db_path)
    conn.row_factory = sqlite3.Row
    return conn

def init_db(db_path: str = DB_PATH) -> None:
    """Initialize database with schema."""
    conn = get_connection(db_path)
    cursor = conn.cursor()
    cursor.execute("""
        CREATE TABLE IF NOT EXISTS users (
            id INTEGER PRIMARY KEY AUTOINCREMENT,
            username TEXT UNIQUE NOT NULL,
            email TEXT UNIQUE NOT NULL,
            created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
        )
    """)
    conn.commit()
    conn.close()

def db_query(query: str, params: tuple = ()) -> List[Dict[str, Any]]:
    """Execute SELECT query and return results."""
    conn = get_connection()
    cursor = conn.cursor()
    cursor.execute(query, params)
    rows = cursor.fetchall()
    conn.close()
    return [dict(row) for row in rows]

def db_execute(query: str, params: tuple = ()) -> int:
    """Execute INSERT/UPDATE/DELETE and return affected rows."""
    conn = get_connection()
    cursor = conn.cursor()
    cursor.execute(query, params)
    conn.commit()
    affected = cursor.rowcount
    conn.close()
    return affected

def db_insert(query: str, params: tuple = ()) -> int:
    """Execute INSERT and return last row ID."""
    conn = get_connection()
    cursor = conn.cursor()
    cursor.execute(query, params)
    conn.commit()
    last_id = cursor.lastrowid
    conn.close()
    return last_id

ドメイン固有の関数

# Add to database.py - Clean API for application code

def get_user_by_id(user_id: int) -> Optional[Dict[str, Any]]:
    """Get user by ID."""
    results = db_query("SELECT * FROM users WHERE id = ?", (user_id,))
    return results[0] if results else None

def get_user_by_email(email: str) -> Optional[Dict[str, Any]]:
    """Get user by email."""
    results = db_query("SELECT * FROM users WHERE email = ?", (email,))
    return results[0] if results else None

def create_user(username: str, email: str) -> int:
    """Create new user and return ID."""
    return db_insert(
        "INSERT INTO users (username, email) VALUES (?, ?)",
        (username, email)
    )

def delete_user(user_id: int) -> bool:
    """Delete user by ID."""
    return db_execute("DELETE FROM users WHERE id = ?", (user_id,)) > 0

アプリケーションでの使用方法

# app.py - NO SQL HERE!
from database import init_db, get_user_by_id, create_user

def main():
    init_db()

    # Create user (no SQL!)
    user_id = create_user("alice", "[email protected]")
    print(f"Created user: {user_id}")

    # Get user (no SQL!)
    user = get_user_by_id(user_id)
    print(f"User: {user['username']}")

避けるべきアンチパターン

❌ 間違い：アプリケーションコード内の SQL

def process_user(user_id):
    conn = sqlite3.connect('app.db')
    cursor = conn.cursor()
    cursor.execute("SELECT * FROM users WHERE id = ?", (user_id,))

✅ 正しい方法：データベース関数を使用する

def process_user(user_id):
    user = get_user_by_id(user_id)

セキュリティ：常にパラメーターを使用する

# ❌ WRONG: SQL injection vulnerability!
query = f"SELECT * FROM users WHERE email = '{email}'"

# ✅ CORRECT: Parameterized query
query = "SELECT * FROM users WHERE email = ?"
results = db_query(query, (email,))

よくあるパターン

ページネーション

def get_users_paginated(page: int = 1, per_page: int = 10):
    offset = (page - 1) * per_page
    return db_query(
        "SELECT * FROM users ORDER BY created_at DESC LIMIT ? OFFSET ?",
        (per_page, offset)
    )

トランザクション

def db_transaction(operations: List[tuple]) -> bool:
    conn = get_connection()
    cursor = conn.cursor()
    try:
        for query, params in operations:
            cursor.execute(query, params)
        conn.commit()
        return True
    except:
        conn.rollback()
        return False
    finally:
        conn.close()

守るべきルール

1. ✅ すべての SQL を database.py に記述する - それ以外の場所には記述しません
2. ✅ パラメーター化クエリ - SQL インジェクションを防ぎます
3. ✅ 意味のある戻り値の型 - Optional、List、bool、int
4. ✅ トランザクションをサポートする - 複数操作の一貫性を保つため
5. ✅ *.db を .gitignore に追加する - データベースをコミットしないでください

関連リソース

完全なドキュメントについては、AgentUsage/db_usage.md を参照してください。次の内容が含まれています：

• database.py の完全なテンプレート
• マイグレーションのパターン
• データベース関数のテスト
• パフォーマンスの最適化