【連載】Python FastAPI入門：第4回 FastAPIでCRUD操作を実装する【DB連携の入り口編】

前回は、Pydanticを使ってリクエストボディの構造化とバリデーションを行う方法を解説しました。

前回の記事はこちら：

第3回：Pydanticによるリクエストボディとデータバリデーション

番外編：FastAPIの心臓部「Pydantic」を徹底解説！

これまでは実行中のメモリ上だけでデータを扱ってきましたが、サーバーを再起動するとデータは消えてしまいます。第4回では、いよいよデータベース（SQLite）と連携し、データを保存・読み出しする「CRUD操作」の入り口を実装してみましょう。

データベース連携の要「ORM」とは？
- 組み込みエンジニア視点でのORM
準備：必要なライブラリのインストール
PydanticとSQLAlchemyの違いを理解する
実装：データベース接続とモデル定義
実装：APIエンドポイントの作成
Swagger UIで確認する
まとめ：第4回はここまで

データベース連携の要「ORM」とは？

Web開発でデータベースを扱う際、SQL文を直接書く代わりに、Pythonのオブジェクトとして操作できるようにする仕組みをORM（Object-Relational Mapping）と呼びます。

FastAPIでよく使われるのが「SQLAlchemy」というライブラリです。

組み込みエンジニア視点でのORM

組み込み開発で例えるなら、レジスタを直接叩くのではなく、構造体やAPI（HALなど）を介して周辺ペリフェラルを操作する感覚に近いです。ハードウェア（DB）の詳細を隠蔽し、コードの可読性と移植性を高めてくれます。

準備：必要なライブラリのインストール

今回は軽量なSQLiteを使用します。以下のライブラリをインストールしてください。

pip install sqlalchemy

PydanticとSQLAlchemyの違いを理解する

ここが初心者が一番混乱するポイントです。

Pydanticモデル: 「外（クライアント）」とやり取りするデータの形（バリデーション用）
SQLAlchemyモデル: 「内（データベース）」に保存するデータの形（テーブル定義用）

似ていますが、役割が違います。この2つを連携させるのがFastAPIの王道スタイルです。

実装：データベース接続とモデル定義

まずはデータベースの設定と、テーブルの定義（database.py）を作成します。

from sqlalchemy import Column, Integer, String, create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker

# SQLiteを使用（ファイル名は test.db）
SQLALCHEMY_DATABASE_URL = "sqlite:///./test.db"

engine = create_engine(SQLALCHEMY_DATABASE_URL, connect_args={"check_same_thread": False})
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
Base = declarative_base()

# データベースのテーブル定義
class User(Base):
    __tablename__ = "users"
    id = Column(Integer, primary_key=True, index=True)
    name = Column(String)
    email = Column(String, unique=True, index=True)

実装：APIエンドポイントの作成

次に、メインのプログラム（main.py）で、データベースへの保存（Create）と取得（Read）を実装します。

from fastapi import FastAPI, Depends, HTTPException
from sqlalchemy.orm import Session
import database # 上記の設定ファイルをインポート

# テーブル作成
database.Base.metadata.create_all(bind=database.engine)

app = FastAPI()

# DBセッションの取得用関数
def get_db():
    db = database.SessionLocal()
    try:
        yield db
    finally:
        db.close()

# ユーザー作成（Create）
@app.post("/users/")
def create_user(name: str, email: str, db: Session = Depends(get_db)):
    db_user = database.User(name=name, email=email)
    db.add(db_user)
    db.commit()
    db.refresh(db_user)
    return db_user

# ユーザー取得（Read）
@app.get("/users/{user_id}")
def read_user(user_id: int, db: Session = Depends(get_db)):
    user = db.query(database.User).filter(database.User.id == user_id).first()
    if user is None:
        raise HTTPException(status_code=404, detail="User not found")
    return user