SQLAlchemyでリレーショナルデータベースに接続

前の章では、APIリクエストを受け取りPydanticでデータを検証する方法を学びました。しかし、サーバーを停止するとこれらのデータは消えてしまいます。実際のビジネスアプリケーションを構築するには、データをデータベースに永続的に保存する必要があります。

直接SQL文（INSERT INTO users...など）を書いてデータベースを操作することもできますが、現代のバックエンド開発では**ORM（Object-Relational Mapping）**技術の使用を強く推奨します。

Python界において、最も歴史があり堅牢で、業界標準に最も適合しているORMはSQLAlchemyです。

1. ORMとは？なぜ使うのか？

ORMの核心概念は、データベースの「テーブル」をPythonの「クラス」にマッピングすることです。

SELECT * FROM usersと書く代わりに、db.query(User).all()と書けます。
INSERT INTOと書く代わりに、User(name="Ken")オブジェクトを作成してdb.add()するだけです。

これには3つの大きな利点があります：

SQLインジェクション攻撃の防止：ORMは内部で危険な文字を自動的にフィルタリングします。
データベースの切り替えが容易：今日はSQLiteでテストし、明日は本番環境でPostgreSQLに切り替える場合、ロジックコードを一行も変更せず、接続文字列を変更するだけで済みます。
コード補完が可能：エディタでuser.と打つと全てのフィールドが自動的に表示されます。手書きSQLではこのような恩恵は受けられません。

2. 環境設定とデータベース接続

まず、SQLAlchemyとPostgreSQL接続用のドライバをインストールします：

pip install sqlalchemy psycopg2-binary

次に、database.pyを作成してデータベー��接続を設定します：

from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker, declarative_base

# 自分のデータベースURLに置き換えてください（形式: postgresql://ユーザー名:パスワード@ホスト:ポート/データベース名）
# ここではローカルテスト用に軽量なSQLiteを使用します
SQLALCHEMY_DATABASE_URL = "sqlite:///./sql_app.db"

# エンジンを作成（SQLiteはcheck_same_thread=Falseが必要）
engine = create_engine(
    SQLALCHEMY_DATABASE_URL, connect_args={"check_same_thread": False}
)

# Sessionファクトリを作成
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

# 全てのModelが継承する基底クラスを作成
Base = declarative_base()

3. SQLAlchemyモデルの定義

models.pyでデータテーブルの構造を定義します。これは前章のPydanticモデルとは異なることに注意してください。Pydanticは入力JSONの検証を担当し、SQLAlchemyはデータベースに保存される構造を定義します。

from sqlalchemy import Boolean, Column, Integer, String
from database import Base

class User(Base):
    __tablename__ = "users"

    id = Column(Integer, primary_key=True, index=True)
    email = Column(String, unique=True, index=True)
    hashed_password = Column(String)
    is_active = Column(Boolean, default=True)

4. 依存性注入（Dependency Injection）の設定

FastAPIには「依存性注入（Dependencies）」という非常に強力な機能があります。 APIが呼び出されるたびにデータベース接続（Session）を開き、API終了時に接続を閉じてデータベースに返す必要があります。各APIでこのロジックを手動で書く必要はありません。

main.pyに以下を追加します：

from fastapi import Depends, FastAPI
from sqlalchemy.orm import Session
import models
from database import SessionLocal, engine

# SQLAlchemyに存在しないテーブルを自動的に作成させる
models.Base.metadata.create_all(bind=engine)

app = FastAPI()

# Database Sessionを提供するDependencyを定義
def get_db():
    db = SessionLocal()
    try:
        yield db  # dbをAPIルートに渡す
    finally:
        db.close() # API実行後、確実に接続を閉じる

5. 実践：データをデータベースに書き込む

では、Pydantic、SQLAlchemy、Dependency Injectionをすべて組み合わせてみましょう！

from fastapi import FastAPI, Depends, HTTPException
from sqlalchemy.orm import Session

# 作成したものをインポート
from database import get_db
import models
from pydantic import BaseModel

app = FastAPI()

# 受信用Pydanticスキーマを定義
class UserCreate(BaseModel):
    email: str
    password: str

# 返信用Pydanticスキーマを定義（パスワードを隠す）
class UserResponse(BaseModel):
    id: int
    email: str
    is_active: bool

    class Config:
        from_attributes = True # SQLAlchemyオブジェクトからの直接変換を許可

@app.post("/users/", response_model=UserResponse)
def create_user(user: UserCreate, db: Session = Depends(get_db)):
    # 1. メールアドレスが既に登録されていないか確認
    db_user = db.query(models.User).filter(models.User.email == user.email).first()
    if db_user:
        raise HTTPException(status_code=400, detail="メールアドレスは既に登録されています")
    
    # 2. SQLAlchemy ORMオブジェクトを作成（実際にはパスワードをハッシュ化する必要あり！）
    fake_hashed_password = user.password + "notreallyhashed"
    new_user = models.User(email=user.email, hashed_password=fake_hashed_password)
    
    # 3. データベースに書き込み
    db.add(new_user)
    db.commit()
    db.refresh(new_user) # 生成されたIDを取得
    
    # 4. フロントエンドに返す
    return new_user

まとめ

これでバックエンド開発の3つの基盤を学びました：

ルーティング：HTTPリクエストを受け取る
検証：データ形式が正しいことを保証する
保存：データベースと安全に連携する

ただし、上記の例ではパスワードが平文で保存されており、誰でもこのAPIを呼び出せます。次の章では、セキュリティ保護の核心であるパスワードハッシュ化とJWT認証を学びます！