はじめに

Django のようなフルスタックフレームワークを使えば、ORM やマイグレーションツールは最初からついていることが多いです。
あえてそんな便利な子たちを封印して、必要最小限の機能だけで簡単なバックエンドを作ってみようというチャレンジです ✊

FastAPI, SQLAlchemy, Alembic って？

FastAPI

軽量で HTTP ベースの Web API に特化した Python フレームワーク

SQLAlchemy

Python の ORM (Object Relational Mapping)

Alembic

Python の RDB マイグレーションツール

作っていく 🔨

成果物 🏁

いわゆる Web 3 層構造の超シンプルな TODO アプリです。
フロントもくっついていますが、この記事では触れません。
web-server を Node.js の開発用サーバーでホストしてるのはご愛嬌 🙈

ディレクトリ構成

.
├── api-server
│   ├── Dockerfile
│   ├── alembic.ini
│   ├── app
│   │   ├── crud.py
│   │   ├── database.py
│   │   ├── endpoints.py
│   │   ├── main.py
│   │   ├── models.py
│   │   └── schemas.py
│   ├── entrypoint.sh
│   ├── migrations
│   │   ├── README
│   │   ├── env.py
│   │   ├── script.py.mako
│   │   └── versions
│   │       └── 7c881a2854a7_create_table.py
│   └── requirements.txt
└── compose.yml

1. コンテナを定義する 📦

必要なライブラリたちをインストールして、サーバーを立ち上げます。
接続先 DB の URL を環境変数 DATABASE_URL として定義していますが、compose.yml で上書きするのでテキトーです。

./api-server/Dockerfile

FROM python:3.12-slim

WORKDIR /api-server
COPY requirements.txt ./
RUN pip install --no-cache-dir -r requirements.txt

ENV DATABASE_URL=driver://user:pass@localhost/dbname

COPY . .
RUN apt-get update && apt-get install -y curl
RUN curl -o /usr/local/bin/wait-for-it.sh https://raw.githubusercontent.com/vishnubob/wait-for-it/master/wait-for-it.sh
RUN chmod +x /usr/local/bin/wait-for-it.sh
RUN chmod +x ./entrypoint.sh

ENTRYPOINT ["./entrypoint.sh"]

インストールするライブラリたち。
black, flake8, isort とかはお好みでどうぞ 💁‍♂️

./api-server/requirement.txt

fastapi
uvicorn

databases[postgresql]
SQLAlchemy
psycopg2-binary
pydantic
alembic

compose.yml で depends_on してますが、database コンテナの起動をちゃんと待ってあげないといけないみたいです 🤔
database コンテナが起動したら、uvicorn サーバーを起動します。

./api-server/entrypoint.sh

#!/bin/sh

# wait for database to be ready
# https://github.com/vishnubob/wait-for-it
/usr/local/bin/wait-for-it.sh database:5432 --timeout=60 --strict -- echo "Database is up"

# start app
uvicorn app.main:app --reload --host 0.0.0.0 --port 8000

# wait for uvicorn process to complete
wait

環境変数など、コンテナの諸々の設定はこんな感じです。
DB のデータをボリュームマウントしておきます。

./compose.yml

services:
  api-server:
    build: ./api-server
    image: api-server:latest
    container_name: api-server
    hostname: api-server
    environment:
      - DATABASE_URL=postgresql://postgres:postgres@database:5432/postgres
    volumes:
      - ./api-server:/opt/api-server
    ports:
      - 8000:8000
    depends_on:
      - database

  database:
    image: postgres:15
    container_name: database
    hostname: database
    platform: linux/amd64
    environment:
      - POSTGRES_PASSWORD=postgres
      - POSTGRES_USER=postgres
      - POSTGRES_DB=postgres
    volumes:
      - db-data:/var/lib/postgresql/data
    ports:
      - 5432:5432

volumes:
  db-data:

2. SQLAlchemy を設定する ⚙️

やっていることは大きく 2 つ。

すべてのモデルの基底クラス Base の作成
- モデルを定義するときは、Base を継承します。
DB セッションオブジェクト session の作成、管理
- DB コネクションの作成や関連のエラーを一元管理したかったため。
- DB クエリを発行する関数には、session を DI（依存性の注入）します 💉

./api-server/app/database.py

import os
from typing import Generator

from fastapi import HTTPException
from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker

# 環境変数から接続先DBのURLを取得する
database_url = os.getenv("DATABASE_URL")

# DBセッションオブジェクトのファクトリクラスを作成する
engine = create_engine(database_url, echo=False)
SessionMaker = sessionmaker(autocommit=False, autoflush=False, bind=engine)

# すべてのモデルの基底クラスを作成する
Base = declarative_base()


def get_session() -> Generator:
    """DBセッションオブジェクトを作成、管理する
    Returns:
        Generator: DBセッションオブジェクト
    """

    session = None

    try:
        session = SessionMaker()
        yield session

    except Exception as e:
        if session:
            session.rollback()
        raise HTTPException(status_code=500, detail=str(e))

    finally:
        session.close()

3. テーブルを作成する

モデルを定義する

前述の通り、database.py で定義した基底クラス Base を継承します。

./api-server/app/models.py

from sqlalchemy import Boolean, Column, Integer, String

from .database import Base


class ToDo(Base):
    __tablename__ = "todos"
    id = Column(Integer, primary_key=True, autoincrement=True)
    text = Column(String)
    done = Column(Boolean, default=False)

alembic の初期設定をする ⚙️

api-server コンテナの中に入ります。

docker compose exec api-server bash

ソースがマウントされているディレクトリで alembic init します。

cd /opt/api-server/
alembic init migrations

alembic init すると、以下のように alembic の設定ファイルたちが作成されます。

.
└── api-server
    ├── alembic.ini
    └── migrations
        ├── README
        ├── env.py
        ├── script.py.mako
        └── versions

alembic に接続先 DB の URL を教えてあげます 🧑‍🏫
api-server コンテナの環境変数から読み込みます。

./api-server/migrations/env.py

+ import os
from logging.config import fileConfig

import app.models  # NOQA
from alembic import context
from app.database import Base
from sqlalchemy import engine_from_config, pool

# this is the Alembic Config object, which provides
# access to the values within the .ini file in use.
config = context.config

+ # set database URL to connect to
+ database_url = os.getenv("DATABASE_URL")
+ if database_url:
+     config.set_main_option("sqlalchemy.url", database_url)
+
# Interpret the config file for Python logging.
# This line sets up loggers basically.
if config.config_file_name is not None:
    fileConfig(config.config_file_name)

# add your model's MetaData object here
# for 'autogenerate' support
# from myapp import mymodel
# target_metadata = mymodel.Base.metadata
+ target_metadata = Base.metadata

# other values from the config, defined by the needs of env.py,
# can be acquired:
# my_important_option = config.get_main_option("my_important_option")
# ... etc.


def run_migrations_offline() -> None:
    """Run migrations in 'offline' mode.

    This configures the context with just a URL
    and not an Engine, though an Engine is acceptable
    here as well.  By skipping the Engine creation
    we don't even need a DBAPI to be available.

    Calls to context.execute() here emit the given string to the
    script output.

    """
    url = config.get_main_option("sqlalchemy.url")
    context.configure(
        url=url,
        target_metadata=target_metadata,
        literal_binds=True,
        dialect_opts={"paramstyle": "named"},
    )

    with context.begin_transaction():
        context.run_migrations()


def run_migrations_online() -> None:
    """Run migrations in 'online' mode.

    In this scenario we need to create an Engine
    and associate a connection with the context.

    """
    connectable = engine_from_config(
        config.get_section(config.config_ini_section, {}),
        prefix="sqlalchemy.",
        poolclass=pool.NullPool,
    )

    url = config.get_main_option("sqlalchemy.url")
    with connectable.connect() as connection:
        context.configure(url=url, connection=connection, target_metadata=target_metadata)

        with context.begin_transaction():
            context.run_migrations()


if context.is_offline_mode():
    run_migrations_offline()
else:
    run_migrations_online()

マイグレーションする

api-server コンテナの中で、以下のコマンドを実行する。

# マイグレーションファイルを作成する
alembic revision --autogenerate -m 'create table'
# マイグレーションを実行する
alembic upgrade head

マイグレーションファイルは、./api-server/migrations/versions/ 配下に作成されます。

.
└── api-server
    ├── alembic.ini
    └── migrations
        ├── README
        ├── env.py
        ├── script.py.mako
        └── versions
            └── 7c881a2854a7_create_table.py

以降、コンテナ起動時に未実行のマイグレーションが実行されるよう、./api-server/entrypoint.sh を以下のように変更します。

./api-server/entrypoint.sh

#!/bin/sh

# wait for database to be ready
# https://github.com/vishnubob/wait-for-it
/usr/local/bin/wait-for-it.sh database:5432 --timeout=60 --strict -- echo "Database is up"

# start app
- uvicorn app.main:app --reload --host 0.0.0.0 --port 8000
+ uvicorn app.main:app --reload --host 0.0.0.0 --port 8000 &
+
+ # execute migration
+ alembic -c /opt/api-server/alembic.ini upgrade head

# wait for uvicorn process to complete
wait

4. CRUD する ✊

レスポンスモデルを定義します。
普段は標準の dataclass を使うことが多いですが、今回は pydantic というライブラリを使ってみました。
より型安全な開発ができるらしい。（本題ではないので深追いしません 🙈）

./api-server/app/schemas.py

from pydantic import BaseModel
from pydantic.alias_generators import to_camel


class ToDo(BaseModel):
    id: int | None = None
    text: str
    done: bool

    class Config:
        alias_generator = to_camel
        allow_population_by_field_name = True

エンドポイントはデコレータで指定します。
FastAPI、シンプルだ... 😳

前述の通り、./api-server/app/database.py で定義した DB セッションオブジェクトを DI しています 💉

./api-server/app/endpoints.py

from fastapi import APIRouter, Depends, HTTPException
from sqlalchemy.orm.session import Session

from . import crud, models, schemas
from .database import get_session

router = APIRouter()


@router.get("/api/v1/todos", response_model=list[schemas.ToDo])
def read_todos(session: Session = Depends(get_session)) -> list[models.ToDo]:
    todos = crud.read_todos(session)

    return todos


@router.get("/api/v1/todos/{id}", response_model=schemas.ToDo)
def read_todo(id: int, session: Session = Depends(get_session)) -> models.ToDo:
    todo = crud.read_todo(id, session)

    return todo


@router.post("/api/v1/todos", response_model=schemas.ToDo)
def create_todo(todo: schemas.ToDo, session: Session = Depends(get_session)) -> models.ToDo:
    text = todo.text
    done = todo.done

    todo = crud.create_todo(text, done, session)

    return todo


@router.put("/api/v1/todos", response_model=schemas.ToDo)
def update_todo(todo: schemas.ToDo, session: Session = Depends(get_session)) -> models.ToDo:
    id = todo.id
    text = todo.text
    done = todo.done

    if id is None:
        raise HTTPException(status_code=400, detail="error")

    todo = crud.update_todo(id, text, done, session)

    return todo


@router.delete("/api/v1/todos/{id}", response_model=schemas.ToDo)
def delete_todo(id: int, session: Session = Depends(get_session)) -> models.ToDo:
    todo = crud.delete_todo(id, session)

    return todo

ORM 経由で DB にクエリを投げます。

./api-server/app/crud.py

from fastapi import HTTPException
from sqlalchemy import asc
from sqlalchemy.orm.session import Session

from . import models


def read_todos(session: Session) -> list[models.ToDo]:
    try:
        todos = session.query(models.ToDo).order_by(asc(models.ToDo.id)).all()

    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))

    return todos


def read_todo(id: int, session: Session) -> models.ToDo:
    try:
        todo = session.query(models.ToDo).filter(models.ToDo.id == id).first()

        if todo is None:
            raise HTTPException(status_code=404, detail="ToDo not found")

    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))

    return todo


def create_todo(text: str, done: bool, session: Session) -> models.ToDo:
    todo = models.ToDo(text=text, done=done)

    try:
        session.add(todo)
        session.commit()
        session.refresh(todo)

    except Exception as e:
        session.rollback()
        raise HTTPException(status_code=500, detail=str(e))

    return todo


def update_todo(id: int, text: str, done: bool, session: Session) -> models.ToDo:
    try:
        todo = session.query(models.ToDo).filter(models.ToDo.id == id).first()

        if todo is None:
            raise HTTPException(status_code=404, detail="ToDo not found")

        todo.text = text
        todo.done = done

        session.commit()
        session.refresh(todo)

    except Exception as e:
        session.rollback()
        raise HTTPException(status_code=500, detail=str(e))

    return todo


def delete_todo(id: int, session: Session) -> models.ToDo:
    try:
        todo = session.query(models.ToDo).filter(models.ToDo.id == id).first()

        if todo is None:
            raise HTTPException(status_code=404, detail="ToDo not found")

        session.delete(todo)
        session.commit()

    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))

    return todo

動かしてみる 🤖

FastAPI は Swagger UI でブラウザから直接 API を呼び出せます。
https://fastapi.tiangolo.com/ja/features/

以下は、GET /api/v1/todos を呼び出した結果です。
問題なく動作していそうです 🎉

おわりに

結論、かなり大変でした 😖
api-server コンテナが立ち上がらない、環境変数を読み込めず DB と繋げない etc...

しかし、フレームワークが「いい感じ」にやってくれている部分を自分で触ることで、アーキテクチャを少しだけ意識でき、良い勉強になったと思います。

みなさんも是非チャレンジしてみてください💪

参考

【Python】SQLAlchemy を試してみる
https://qiita.com/ktamido/items/ebdbe5a85dbc3e6004ae
【SQLAlchemy】エラーから学ぶ Session 管理の重要性
https://zenn.dev/arsaga/articles/f431ee007efae9
Depends による依存性注入
https://qiita.com/Otohu_mimizu/items/edab35e2669719746cc4
DB マイグレーションツールの Alembic の使い方
https://zenn.dev/shimakaze_soft/articles/4c0784d9a87751
FastAPI を使って CRUD API を作成する
https://qiita.com/t-iguchi/items/d01b24fed05db43fd0b8