FastAPIとSQLAlchemyとPydanticを使ってCRUD用APIを実装しよう！

Last updated at 2025-04-28Posted at 2025-02-15

概要

FastAPI
SQLAlchemy
Pydantic

を使ってCRUD操作のAPIを実装します
今回はタスク管理アプリのCRUD操作を例に説明します

前提

FastAPI、SQLAlchemy、Pydanticともにインストール済み
今回はFastAPIとPostgres用のコンテナを使用します
SQLAlchemyのバージョン2以上を使用
Pydanticのバージョン2以上を使用
FastAPIのバージョン0.95.0を使用

ディレクトリ構成

・
├── application
│   ├── __init__.py
│   ├── alembic.ini
│   ├── database.py
│   ├── main.py
│   ├── models.py
│   ├── poetry.lock
│   ├── pyproject.toml
│   ├── schemas.py
├── containers
│   ├── fastapi
│   │   └── Dockerfile
│   └── postgres
│       ├── Dockerfile
│       └── init.sql
└── docker-compose.yml

database.py
models.py
schemas.py
main.py

の順に実装していきます

実装

データベースの設定

FastAPIのアプリケーションとDBを接続する設定をdatabase.pyに記載します

database.py

import os

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

SQLALCHEMY_DATABASE_URL = os.environ.get("SQLALCHEMY_DATABASE_URL")

# https://docs.sqlalchemy.org/en/20/core/engines.html#engine-configuration
engine = create_engine(SQLALCHEMY_DATABASE_URL)

# https://docs.sqlalchemy.org/en/20/orm/session_basics.html#session-basics
# DBセッションのオブジェクトを生成
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

# https://docs.sqlalchemy.org/en/20/orm/mapping_api.html#sqlalchemy.orm.DeclarativeBase
class Base(DeclarativeBase):
    pass

順番に解説します
まず、SQLALCHEMY_DATABASE_URLという変数を定義します
変数内にFastAPIとDBが接続できるためのURLを設定し、それを元にDB用エンジンのオブジェクトを生成します

SQLALCHEMY_DATABASE_URL = os.environ.get("SQLALCHEMY_DATABASE_URL")

engine = create_engine(SQLALCHEMY_DATABASE_URL)

URLのフォーマットは以下ドキュメントを参考にしてください

urlは以下のようになります

SQLALCHEMY_DATABASE_URL=postgresql://{ユーザ名}:{パスワード}@{ホスト名}:5432/{DB名}

その後、DBセッションのオブジェクトを定義します
APIを記載するmain.pyで後ほど使用します

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

Baseクラスを元にモデルを定義します
SQLAlchemyのバージョン2以降からdeclarative_baseではなく、DeclarativeBaseクラスを使うことが推奨されています
modelを定義するmodels.pyで使用します

class Base(DeclarativeBase):
    pass

modelの実装

Todosというmodelを実装します
database.pyで定義したBaseクラスを継承することで作成したmodelの定義をDBに反映させることができます
カラムはsqlalchemyのColumnなどを元に定義していきます
カラムの詳細は以下を参考にしてください

models.py

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


class Todos(Base):
    __tablename__ = "todos"

    id = Column(Integer, primary_key=True, index=True)
    title = Column(String)
    description = Column(String)
    priority = Column(Integer)
    complete = Column(Boolean, default=False)

schemaの実装

schemas.pyバリデーションを定義します
TodoModelはAPIでPostやPutする際のリクエストのバリデーションや型チェックに使用します
TodoResponseはAPIのレスポンスで出力したいフィールドを定義する際に使用します

schemas.py

from pydantic import BaseModel, Field


class TodoModel(BaseModel):
    title: str = Field(min_length=3)
    description: str = Field(min_length=3, max_length=100)
    priority: int = Field(gt=0, lt=6)
    complete: bool


class TodoResponse(BaseModel):
    id: int = Field()
    title: str = Field(min_length=3)
    description: str = Field(min_length=3, max_length=100)
    priority: int = Field(gt=0, lt=6)
    complete: bool

APIの実装

DBの接続設定、modelの定義、バリデーションの設定が完了したのでAPIの実装に入ります

main.py

from typing import Annotated, List

from database import SessionLocal, engine
from fastapi import Depends, FastAPI, HTTPException, status
from models import Base, Todos
from schemas import TodoModel, TodoResponse
from sqlalchemy import select, update
from sqlalchemy.orm import Session

app = FastAPI()

# https://docs.sqlalchemy.org/en/20/core/metadata.html#sqlalchemy.schema.MetaData.create_all
# テーブルを作成
Base.metadata.create_all(bind=engine)


# https://fastapi.tiangolo.com/tutorial/dependencies/dependencies-with-yield/?h=get_db
async def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

# https://fastapi.tiangolo.com/tutorial/sql-databases/#create-a-session-dependency
# 依存性注入をすることでget_dbメソッドが自動的に呼ばるので毎回セッションインスタンスの生成やセッションを切る処理を書かずに済む
# Annotatedを使うことでDepends(get_db)がSession型だとわかる
db_dependency = Annotated[Session, Depends(get_db)]

# https://fastapi.tiangolo.com/tutorial/dependencies/dependencies-with-yield/?h=get_db#sub-dependencies-with-yield
# https://github.com/fastapi/fastapi/pull/9298
# FastAPI0.95.0以降の機能
@app.get("/api/todos", response_model=List[TodoResponse])
def read_todos(db: db_dependency):
    # https://docs.sqlalchemy.org/en/20/orm/quickstart.html#simple-select
    # https://docs.sqlalchemy.org/en/20/orm/session_api.html#sqlalchemy.orm.Session.scalars
    # scalarsを使うことでTodosのインスタンスを返す
    todos = db.scalars(select(Todos).order_by(Todos.id)).all()
    return todos


@app.get("/api/todos/{todo_id}", response_model=TodoResponse)
def read_todo(db: db_dependency, todo_id: int):
    todo = db.get(Todos, todo_id)
    if not todo:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND, detail="Todo not found"
        )
    return todo


@app.post(
    "/api/todos", response_model=TodoResponse, status_code=status.HTTP_201_CREATED
)
def create_todo(db: db_dependency, todo_model: TodoModel):
    # pydantic2ではdict()ではなく、model_dumpが使用されている
    # https://docs.pydantic.dev/latest/concepts/serialization/#modelmodel_dump
    todo = Todos(**todo_model.model_dump())
    # https://docs.sqlalchemy.org/en/20/orm/session_api.html#sqlalchemy.orm.Session.add
    db.add(todo)
    db.commit()
    # # refreshをすることでauto-incrementしたIDやcreated_atが反映されるようになる
    # # https://docs.sqlalchemy.org/en/20/orm/session_api.html#sqlalchemy.orm.Session.refresh
    db.refresh(todo)
    return todo


@app.put("/todo/{todo_id}", response_model=TodoResponse)
async def update_todo(db: db_dependency, todo_model: TodoModel, todo_id: int):
    todo = db.get(Todos, todo_id)
    if not todo:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND, detail="Todo not found"
        )
    # https://docs.sqlalchemy.org/en/20/core/dml.html#sqlalchemy.sql.expression.update
    db.execute(
        update(Todos).where(Todos.id == todo_id).values(**todo_model.model_dump())
    )
    db.commit()
    return todo


@app.delete("/todo/{todo_id}", status_code=status.HTTP_204_NO_CONTENT)
async def delete_todo(db: db_dependency, todo_id: int):
    todo = db.get(Todos, todo_id)
    if not todo:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND, detail="Todo not found"
        )
    # https://docs.sqlalchemy.org/en/20/orm/session_api.html#sqlalchemy.orm.Session.delete
    db.delete(todo)
    db.commit()

順番に解説します
以下の記述を記載することでdatabase.pyで定義したengineの情報を元にDB内にテーブルを作成します

Base.metadata.create_all(bind=engine)

get_dbメソッドを使ってDBのセッションを作成し、使用後に接続を閉じるよう定義します
レスポンスを作成する前に実行されるのは以下の箇所です

    db = DBSession()
    try:
        yield db

以下の箇所はレスポンスが作成される前とレスポンスを返す前に実行されます

    finally:
        db.close()

詳細は以下の通りです

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

下記の記述で依存性を注入しています
依存性注入をするとAPI内でdb_dependencyを定義する際にget_dbメソッドが自動的に呼ばるため、毎回セッションインスタンスの生成(db = SessionLocal())やセッションを切る処理(db.close())を書かずに済みます
また、Annotatedを使うことでDepends(get_db)がSession型だとわかるので便利です

db_dependency = Annotated[Session, Depends(get_db)]

まず、一覧表示のAPIを実装します
先ほど定義したdb_dependencyを使って依存性の注入を実現させます
scalarsを使ってTodosのインスタンスをレスポンスとして返します

@app.get("/api/todos", response_model=List[TodoResponse])
def read_todos(db: db_dependency):
    todos = db.scalars(select(Todos).order_by(Todos.id)).all()
    return todos

詳細表示のAPIを実装します
idからTodoを取得し、あればレスポンスとして返し、なければ404を返します

@app.get("/api/todos/{todo_id}", response_model=TodoResponse)
def read_todo(db: db_dependency, todo_id: int):
    todo = db.get(Todos, todo_id)
    if not todo:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND, detail="Todo not found"
        )
    return todo

作成用のAPIを実装します

pydanticのバージョン2以上ではdict()ではなく、model_dump()を使用します
リクエスト内のフィールドを**でdict型に変換してTodosに代入してTodosのインスタンスを生成します
Todoのインスタンスをセッションにadd()で加えた後にcommit()でTodoの内容をDBにinsertします

    db.add(todo)
    db.commit()

refreshをすることでauto-incrementしたIDやcreated_atが反映されるようになります

@app.post(
    "/api/todos", response_model=TodoResponse, status_code=status.HTTP_201_CREATED
)
def create_todo(db: db_dependency, todo_model: TodoModel):
    todo = Todos(**todo_model.model_dump())
    db.add(todo)
    db.commit()

    db.refresh(todo)
    return todo

更新用のAPIを作成します
該当するidのTodoがなければ404を返します
存在する場合はmodel_dump()で全てのfieldに更新をかけてtodoのインスタンスを返します

@app.put("/todo/{todo_id}", response_model=TodoResponse)
async def update_todo(db: db_dependency, todo_model: TodoModel, todo_id: int):
    todo = db.get(Todos, todo_id)
    if not todo:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND, detail="Todo not found"
        )
    # https://docs.sqlalchemy.org/en/20/core/dml.html#sqlalchemy.sql.expression.update
    db.execute(
        update(Todos).where(Todos.id == todo_id).values(**todo_model.model_dump())
    )
    db.commit()
    return todo

削除用APIを作成します
該当するidのTodoがなければ404を返します
存在する場合はセッション内から削除し、commit()でTodoの内容をDBからdeleteしまs

@app.delete("/todo/{todo_id}", status_code=status.HTTP_204_NO_CONTENT)
async def delete_todo(db: db_dependency, todo_id: int):
    todo = db.get(Todos, todo_id)
    if not todo:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND, detail="Todo not found"
        )
    # https://docs.sqlalchemy.org/en/20/orm/session_api.html#sqlalchemy.orm.Session.delete
    db.delete(todo)
    db.commit()