データベース¶

SQLModel を使って、 FastAPI アプリでSQLデータベースに接続するために最低限必要なことを。
※コードはAPI ルーターからの続き

Note

SQLModel はインストールしておく。

uv add sqlmodel

モデル定義¶

まずは models.py を作成してテーブル構造を決める。

# models.py
from sqlmodel import SQLModel, Field


class User(SQLModel, table=True):
    id: int | None = Field(default=None, primary_key=True)
    name: str
    email: str

Info

table=True を付けておくと、データベースのテーブルになるクラスを定義できる。
id はデータベース側が自動採番するようにできることが多いので、ユーザーからのリクエストでは None でOK。 name , email はユーザーのリクエストとして必須で、文字列型しか受け付けない。

データベースとテーブル作成¶

SQLite

開発の時には、ファイルをデータベースとして扱える SQLite を使うと便利。

上で作成した models.py からデータベースとテーブルを作成するためのコード。

# db.py
from sqlmodel import SQLModel, create_engine

DATABASE_URL = "sqlite:///./app.db"

engine = create_engine(DATABASE_URL, connect_args={"check_same_thread": False})


def create_db_and_tables():
    SQLModel.metadata.create_all(engine)

create_db_and_tables() を呼び出せば、データベースとテーブルが作成される。
データベースとテーブルを作成して、レコードを追加するためのコード。

# db_init.py
from sqlmodel import Session

from db import engine, create_db_and_tables
from models import User

create_db_and_tables()

with Session(engine) as session:
    user = User(name="alice", email="alice@example.com")
    session.add(user)
    session.commit()
    session.refresh(user)

db_init.py を実行すれば、 app.db が作成されて、レコードも作成される。

uv run python db_init.py

sqlite3コマンド

sqlite3コマンドが使えるなら、下記のように中身を確認できる。VSCodeの拡張機能でも中身は確認できる。

% sqlite3 app.db 
SQLite version 3.51.0 2025-06-12 13:14:41
Enter ".help" for usage hints.
sqlite> .tables
user
sqlite> .header on
sqlite> SELECT * FROM user;
id|name|email
1|alice|alice@example.com

CRUD 処理¶

作成したデータベースを使って CRUD 処理ができるように、 db.py と users.py を修正。

# db.py
from typing import Annotated

from fastapi import Depends
from sqlmodel import (
    SQLModel,
    Session,
    create_engine,
)

DATABASE_URL = "sqlite:///./app.db"

engine = create_engine(DATABASE_URL)


def create_db_and_tables():
    SQLModel.metadata.create_all(engine)


def get_session():
    with Session(engine) as session:
        yield session


SessionDep = Annotated[Session, Depends(get_session)]

Tip

https://github.com/fastapi/fastapi/blob/master/fastapi/.agents/skills/fastapi/SKILL.md#for-dependencies にあるように、 Annotated と一緒に Depends を使った。
こうやって書いておけば、データベース接続が必要なエンドポイントの関数には、引数として session: SessionDep って書いとけばいいだけになる。

# users.py
from fastapi import APIRouter, HTTPException

from db import SessionDep
from models import User

# 前のページにあった脆弱な認証方法は削除

router = APIRouter(
    prefix="/users",
    tags=["users"],
)


@router.post("/")
def create_user(
    user: User,
    session: SessionDep
):
    session.add(user)
    session.commit()
    session.refresh(user)
    return user


@router.get("/{user_id}")
def read_user(
    user_id: int,
    session: SessionDep
):
    user = session.get(User, user_id)
    if not user:
        raise HTTPException(status_code=404, detail="User not found")
    return user


@router.patch("/{user_id}")
def update_user(
    user_id: int,
    user: User,
    session: SessionDep
):
    db_user = session.get(User, user_id)
    if not db_user:
        raise HTTPException(status_code=404, detail="User not found")
    user_data = user.model_dump(exclude_unset=True)
    for key, value in user_data.items():
        if key == "id":
            continue
        setattr(db_user, key, value)
    session.add(db_user)
    session.commit()
    session.refresh(db_user)
    return db_user


@router.delete("/{user_id}")
def delete_user(
    user_id: int,
    session: SessionDep
):
    user = session.get(User, user_id)
    if not user:
        raise HTTPException(status_code=404, detail="User not found")
    session.delete(user)
    session.commit()
    return {"message": "User deleted successfully"}