【第5回】FastAPI×SQLiteでユーザー管理APIを実装！モンスター育成ゲームプロジェクトCRUD入門

「本ページはプロモーションが含まれています」

こんにちは。これまでの連載では、

第1回: モンスター育成ゲームAPIの概要や狙い
第2回: Mermaidでシーケンス図を描き、ゲーム内フローを可視化
第3回: MermaidでDB設計をざっくり考え（まだマイグレーションは導入せず）
第4回: FastAPI環境構築とフォルダ構成、Hello Worldでサーバー起動

前回はコチラ kunio-ud-zatta.hatenablog.com

…という流れで準備してきました。今回はユーザー管理APIの作成です。ユーザーを新規登録したり、一覧・詳細を取得したり、削除したりと、いわゆるCRUDを実装します。まずはシンプルな動作確認を目指してます。

1. 今回のゴール

Userテーブルを定義（SQLite使用）
Userモデル（SQLAlchemy）を作成
ユーザー用のAPIRouterで、基本的なCRUD（Create, Read, Delete）を実装
(時間があれば)レスポンスモデルをPydanticで定義し、返却値を整備
とりあえずデータの登録＆取得＆削除が動く状態にする

ブルーアーカイブゲーム開発部だいぼうけん！ 3巻 (デジタル版ガンガンコミックスONLINE)

Amazon

2. DB接続まわりの準備

2.1 database.py にDBセッションを用意

まず、インストール

pip install sqlalchemy

app/database.py

from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker

DATABASE_URL = "sqlite:///./test.db"  # SQLiteファイルへのパス
engine = create_engine(DATABASE_URL, connect_args={"check_same_thread": False})

SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

DATABASE_URL = "sqlite:///./test.db" で、プロジェクトルートに test.db が作られるイメージ
get_db() という依存関係（Dependency）を介して、FastAPIでDBセッションを取得可能に

2.2 models.py でテーブル定義

※結局、ここでマイグレーションをしようと思います。前回は、クエリーの練習！ってことで。

app/models.py

from sqlalchemy import Column, Integer, String, DateTime, func
from sqlalchemy.ext.declarative import declarative_base

Base = declarative_base()

class User(Base):
    __tablename__ = "users"

    id = Column(Integer, primary_key=True, index=True)
    username = Column(String, unique=True, nullable=False)
    created_at = Column(DateTime, server_default=func.now())

username をユニーク制約にし、同じ名前での重複登録を防ぐ(予定)

2.3 create_all でテーブルを作成

起動時にテーブルがなければ自動作成するようにするには、main.py か database.py のどこかで Base.metadata.create_all(engine) を呼び出します。例えば main.py の先頭あたりで：

これでFastAPIを起動すると、test.db に users テーブルが生成されます。

インディーゲーム・サバイバルガイド

作者:一條貴彰
技術評論社

Amazon

3. ユーザー管理のルーターを作成

3.1 users.py

from fastapi import APIRouter, Depends, HTTPException
from sqlalchemy.orm import Session
from pydantic import BaseModel

from app.database import get_db
from app.models import User

router = APIRouter()

# Pydanticモデル
class UserCreate(BaseModel):
    username: str

class UserRead(BaseModel):
    id: int
    username: str

    class Config:
        orm_mode = True

# CREATE (ユーザー新規登録)
@router.post("/", response_model=UserRead)
def create_user(user_data: UserCreate, db: Session = Depends(get_db)):
    # 既存ユーザー重複チェック
    existing = db.query(User).filter(User.username == user_data.username).first()
    if existing:
        raise HTTPException(status_code=400, detail="Username already exists")

    new_user = User(
        username=user_data.username
    )
    db.add(new_user)
    db.commit()
    db.refresh(new_user)

    return new_user

# READ (一覧取得)
@router.get("/", response_model=list[UserRead])
def get_all_users(db: Session = Depends(get_db)):
    users = db.query(User).all()
    return users

# READ by ID
@router.get("/{user_id}", response_model=UserRead)
def get_user_by_id(user_id: int, db: Session = Depends(get_db)):
    user = db.query(User).filter(User.id == user_id).first()
    if not user:
        raise HTTPException(status_code=404, detail="User not found")
    return user

# DELETE
@router.delete("/{user_id}")
def delete_user(user_id: int, db: Session = Depends(get_db)):
    user = db.query(User).filter(User.id == user_id).first()
    if not user:
        raise HTTPException(status_code=404, detail="User not found")

    db.delete(user)
    db.commit()
    return {"detail": "User deleted"}