FastAPI データベース統合

FastAPI はさまざまなデータベースと統合でき、最も一般的な方法は SQLAlchemy を ORM として使用することです。このセクションでは、FastAPI + SQLAlchemy を使用して CRUD（作成、読み取り、更新、削除）機能を持つ API を構築する方法を紹介します。

依存関係のインストール

pip install sqlalchemy

プロジェクト構造

project/
├── main.py          # FastAPI 应用入口
├── database.py      # 数据库连接配置
├── models.py        # SQLAlchemy 模型
├── schemas.py       # Pydantic 数据模型
└── crud.py          # 数据库操作函数

1. データベース設定

にdatabase.pyでデータベース接続を設定します：

例

# database.py
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker, DeclarativeBase

# SQLite データベース（開発に適しています）
SQLALCHEMY_DATABASE_URL = "sqlite:///./sql_app.db"

# データベースエンジンを作成
# connect_args は SQLite のみに必要で、マルチスレッドアクセスを許可します
engine = create_engine(
    SQLALCHEMY_DATABASE_URL,
    connect_args={"check_same_thread": False}
)

# セッションファクトリを作成
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)


# 基底クラスを宣言
class Base(DeclarativeBase):
    pass

開発環境では SQLite で十分です。ファイルデータベースであり、追加のデータベースサービスをインストールする必要はありません。本番環境では PostgreSQL または MySQL の使用を推奨します。変更が必要なのはSQLALCHEMY_DATABASE_URLだけです。

2. SQLAlchemy モデル

にmodels.pyでデータベーステーブル構造を定義します：

例

# models.py
from sqlalchemy import 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)             # アクティブかどうか


class Item(Base):
    __tablename__ = "items"

    id = Column(Integer, primary_key=True, index=True)
    title = Column(String, index=True)                   # タイトル、インデックス作成
    description = Column(String)                          # 説明
    owner_id = Column(Integer, ForeignKey("users.id"))   # 外部キーで users テーブルに関連付け

3. Pydantic データモデル

にschemas.pyで API の入力/出力データ構造を定義します：

例

# schemas.py
from pydantic import BaseModel


# ===== ユーザーモデル =====
class UserBase(BaseModel):
    email: str


class UserCreate(UserBase):
    password: str          # 作成時にはパスワードが必要


class UserOut(UserBase):
    id: int
    is_active: bool

    model_config = ConfigDict(from_attributes=True)  # ORM オブジェクトをサポート


# ===== 商品モデル =====
class ItemBase(BaseModel):
    title: str
    description: str | None = None


class ItemCreate(ItemBase):
    pass


class ItemOut(ItemBase):
    id: int
    owner_id: int

    model_config = ConfigDict(from_attributes=True)

Pydantic モデルと SQLAlchemy モデルは分離されています。Pydantic は API 層のデータ検証を担当し、SQLAlchemy はデータベース層のテーブル構造を担当します。from_attributes=TruePydantic が SQLAlchemy オブジェクトからデータを読み取れるようにします。

4. データベース操作関数

にcrud.pyでデータベース操作をカプセル化します：

例

# crud.py
from sqlalchemy.orm import Session
import models, schemas


def get_user(db: Session, user_id: int):
    """ID でユーザーを取得"""
    return db.query(models.User).filter(models.User.id == user_id).first()


def get_user_by_email(db: Session, email: str):
    """メールアドレスでユーザーを取得"""
    return db.query(models.User).filter(models.User.email == email).first()


def get_users(db: Session, skip: int = 0, limit: int = 100):
    """ユーザー一覧を取得"""
    return db.query(models.User).offset(skip).limit(limit).all()


def create_user(db: Session, user: schemas.UserCreate):
    """ユーザーを作成"""
    fake_hashed_password = user.password + "notreallyhashed"  # 実際には passlib でハッシュ化する必要があります
    db_user = models.User(
        email=user.email,
        hashed_password=fake_hashed_password
    )
    db.add(db_user)
    db.commit()        # トランザクションをコミット
    db.refresh(db_user)  # オブジェクトを更新し、データベースで生成された id を取得
    return db_user


def get_items(db: Session, skip: int = 0, limit: int = 100):
    """商品一覧を取得"""
    return db.query(models.Item).offset(skip).limit(limit).all()


def create_item(db: Session, item: schemas.ItemCreate, user_id: int):
    """商品を作成"""
    db_item = models.Item(**item.model_dump(), owner_id=user_id)
    db.add(db_item)
    db.commit()
    db.refresh(db_item)
    return db_item

5. FastAPI アプリケーションエントリ

にmain.pyで全てのコンポーネントを統合します：

例

# main.py
from typing import Annotated
from fastapi import Depends, FastAPI, HTTPException
from sqlalchemy.orm import Session
from pydantic import ConfigDict

import models, schemas, crud
from database import engine, SessionLocal, Base

# データベーステーブルを作成
Base.metadata.create_all(bind=engine)

app = FastAPI()


# 依存性：データベースセッションを取得
def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()


# ===== ユーザールート =====
@app.post("/users/", response_model=schemas.UserOut)
def create_user(user: schemas.UserCreate, db: Session = Depends(get_db)):
    # メールアドレスが既に存在するか確認
    db_user = crud.get_user_by_email(db, email=user.email)
    if db_user:
        raise HTTPException(status_code=400, detail="メールアドレスは登録済みです")
    return crud.create_user(db=db, user=user)


@app.get("/users/", response_model=list[schemas.UserOut])
def read_users(skip: int = 0, limit: int = 100, db: Session = Depends(get_db)):
    users = crud.get_users(db, skip=skip, limit=limit)
    return users


@app.get("/users/{user_id}", response_model=schemas.UserOut)
def read_user(user_id: int, db: Session = Depends(get_db)):
    db_user = crud.get_user(db, user_id=user_id)
    if db_user is None:
        raise HTTPException(status_code=404, detail="ユーザーが存在しません")
    return db_user


# ===== 商品ルート =====
@app.post("/users/{user_id}/items/", response_model=schemas.ItemOut)
def create_item_for_user(
    user_id: int,
    item: schemas.ItemCreate,
    db: Session = Depends(get_db),
):
    return crud.create_item(db=db, item=item, user_id=user_id)


@app.get("/items/", response_model=list[schemas.ItemOut])
def read_items(skip: int = 0, limit: int = 100, db: Session = Depends(get_db)):
    items = crud.get_items(db, skip=skip, limit=limit)
    return items

データベースセッションの依存性注入

コアパターンは、依存性注入を使用してデータベースセッションを管理することです：

def get_db():
    db = SessionLocal()   # 创建会话
    try:
        yield db          # 提供会话给路由函数
    finally:
        db.close()        # 请求完成后关闭会话

各リクエストは独立したデータベースセッションを取得し、リクエスト完了後に自動的に閉じられ、リソースリークを防ぎます。

異なるデータベースの接続文字列

データベース接続文字列の例
SQLitesqlite:///./sql_app.db
PostgreSQLpostgresql://user:password@localhost/dbname
MySQLmysql+pymysql://user:password@localhost/dbname

データベースを切り替えるには接続文字列を変更するだけでよく、その他のコードは変更する必要はありません。これも ORM を使用する利点の一つです。

まとめ

  • SQLAlchemy を ORM として使用し、FastAPI は依存性注入を通じてデータベースセッションを管理します
  • Pydantic モデル（schemas.py）は API データ検証を担当し、SQLAlchemy モデル（models.py）はデータベーステーブル構造を担当します
  • CRUD 操作はcrud.pyにカプセル化され、コードを明確に保ちます
  • 開発では SQLite を使用し、本番では接続文字列を変更するだけで PostgreSQL/MySQL に切り替えられます
  • from_attributes=TruePydantic が ORM オブジェクトからデータを読み取れるようにします
その他の拡張