FastAPIのORMをormarでセットアップする

背景

普段Djangoで開発をしている筆者ですが、最近そういえばFastAPIで開発していないなと思い、久しぶりに触ってみました。
Djangoで開発しているときはDjango ORMを使ってモデルを定義してマイグレーションして...とあまりデータベース周りのことを考えずに開発をするのですが、そういえばFastAPIはORMも自分で選んで導入できるなと思いだし、色々調べていました。
調べていく中で、Tortoise ORMとormarがいい感じだったので、今回はormarをFastAPIに導入してみたので導入の仕方をまとめようと思います。

簡単に本の情報を取得/保存/編集/削除するREST APIを基準に考えていきます。

環境

• Python 3.11.2
• poetry 1.4.1
• FastAPI 0.95.2
• ormar 0.12.1
• uvicorn 0.22.0
• SQLite

参照

コードを直接見たい人はこちらのGitHubを確認してください。

始める前に

この記事はFastAPIの基本的な部分はわかっているエンジニア向けの記事です。
FastAPIは触ったことない、という人はまずこちらからFastAPIに触れてみましょう。

プロジェクトの構成と中身

プロジェクトの構成は以下です。

.
├── README.md
├── app
│   ├── __init__.py
│   ├── books_api
│   │   ├── __init__.py
│   │   ├── models.py
│   │   ├── routers.py
│   │   ├── schemas.py
│   ├── configs
│   │   ├── __init__.py
│   │   └── settings.py
│   └── main.py
├── db
│   ├── dev.sqlite
├── poetry.lock
└── pyproject.toml

基本的にappというディレクトにコードが格納されています。
他のアプリケーションだとsrcもしくはlibに相当すると思ってください。
中を見ていきましょう

books_apiは本関係のAPIのコードが全て格納されています。
筆者がDjangoラブなので、Djangoのappに似た形にしています。

• models.py
  • 最終的にはDBのモデルが格納されますが、ここでは一旦pydanticのモデルを作成して、エンドポイント側ですぐに使えるようにします。

app/books_api/models.py

from pydantic import BaseModel


class Book(BaseModel):
    title: str
    description: str | None
    price: int
    unit: str
    is_active: bool = False

• schemas.py
  • ユーザーからのリクエスト用モデルを格納します。一旦仮置きとしてpydanticでモデルを作成しました。こちらは後でormarのモデルを使った形に直します。

app/books_api/schemas.py

import uuid

from pydantic import BaseModel


class BookIn(BaseModel):
    title: str
    description: str | None
    price: int
    unit: str


class BookUpdateIn(BaseModel):
    id: uuid.UUID = uuid.uuid4()
    title: str
    description: str | None
    price: int
    unit: str
    is_active = True

• routers.py
  • APIエンドポイントが格納されています。
  • 一旦仮置きでCRUDの基本的なAPIエンドポイントを作成しました。
    • 一覧取得
    • 詳細取得
    • 作成
    • 編集
    • 削除

app/books_api/routers.py

import random
from uuid import UUID

from fastapi import APIRouter, Body, FastAPI, HTTPException, Path, status

from .models import Book
from .schemas import BookIn, BookUpdateIn

router = APIRouter()

def setup_routers(app: FastAPI):
    app.include_router(router)


def get_books() -> list[Book]:
    prices = [y for y in range(1000, 3000, 100)]
    def build_book(x):
        title = f"example {x}"
        price = random.choice(prices)
        unit = "JPY"
        return Book(title=title, price=price, unit=unit)
    books = list(map(build_book, range(20)))
    return books


@router.get("/", response_model=list[Book])
async def fetch_books():
    try:
        # TODO: DBからデータを取得する
        books = get_books()
        return books
    except BaseException:
        status_code = status.HTTP_400_BAD_REQUEST
        message = "一覧の取得に失敗しました"
        raise HTTPException(status_code=status_code, detail=message) from None


@router.get("/{book_id}")
async def fetch_book(book_id: UUID = Path(...)) -> Book:
    try:
        # TODO: DBからデータを取得する
        books = get_books()
        book = random.choice(books)
        return book
    except BaseException:
        status_code = status.HTTP_400_BAD_REQUEST
        message = "詳細の取得に失敗しました"
        raise HTTPException(status_code=status_code, detail=message) from None


@router.post("/add", response_model=Book)
async def add_book(book_in: BookIn = Body(...)):
    try:
        book = Book(**book_in.dict())
        # TODO: 保存処理を走らせる
        return book
    except BaseException:
        status_code = status.HTTP_400_BAD_REQUEST
        message = "本の作成に失敗しました"
        raise HTTPException(status_code=status_code, detail=message) from None


@router.put("/{book_id}/update")
async def update_book(book_id: UUID = Path(...), book_update_in: BookUpdateIn = Body(...)) -> Book:
    try:
        updated_book = Book(**book_update_in.dict())
        # TODO: 更新処理を走らせる
        return updated_book
    except BaseException:
        status_code = status.HTTP_400_BAD_REQUEST
        message = "本の編集に失敗しました"
        raise HTTPException(status_code=status_code, detail=message) from None


@router.delete("/{book_id}/delete")
async def delete_book(book_id: UUID):
    try:
        # TODO: 論理削除を走らせる
        return {"message": "success"}
    except BaseException:
        status_code = status.HTTP_400_BAD_REQUEST
        message = "本の削除に失敗しました"
        raise HTTPException(status_code=status_code, detail=message) from None

configsの中にはsettings.pyがあり、アプリケーション全体で使用できる定数を設定しています。

app/configs/settings.py

from functools import lru_cache
from pathlib import Path

from pydantic import BaseSettings

BASE_DIR = Path(__file__).resolve(strict=True).parent.parent.parent


class Settings(BaseSettings):
    base_dir: Path = BASE_DIR
    db_url = f"sqlite:///{BASE_DIR!s}/db/dev.sqlite"


@lru_cache
def get_settings():
    return Settings()

db_urlは今回データベースとして使用するSQLiteのファイルへのパスを設定します。
PostgreSQLやMySQLを使用する場合はそれぞれのDBへのパスを設定してください。

本題

さて、ここからが本題です。
ormar+alembicを使って、FastAPIにDBを導入します。

まずそれぞれの役割から簡単に説明します。

• ormar
  • DB用のモデルを設定
  • DBを操作
  • いわゆるORマッパー
• alembic
  • sqlalchemyで作成されたモデル情報を元に、対象のDBへ情報をマイグレーション

alembicはsqlalchemyのマイグレーション用ですが、ormarはsqlalchemyをベースに作成されているため、alembicをそのまま使用することができます。

DBの接続とFastAPIとの連携

ではまずDBの接続用の設定をしていきます。
configsディレクトリにdatabase.pyを作成して次のコードを入れてください。

app/configs/database.py

from databases import Database
from fastapi import FastAPI
from sqlalchemy import MetaData

from .settings import get_settings

settings = get_settings()

def get_db_url():
    return settings.db_url


database = Database(get_db_url())
metadata = MetaData()

def setup_database(app: FastAPI):
    app.state.database = database

    @app.on_event("startup")
    async def startup() -> None:
        database_ = app.state.database
        if not database_.is_connected:
            await database_.connect()

    @app.on_event("shutdown")
    async def shutdown() -> None:
        database_ = app.state.database
        if database_.is_connected:
            await database_.disconnect()

DBのURLはSettingsにあるので、get_settingsをimportしてインスタンスを生成します。

from .settings import get_settings

settings = get_settings()

データベースはテスト用に切り替えたりできるように、get_db_urlを作ってそこからDBのURLを返却する形にします。

def get_db_url():
    return settings.db_url

次に、databasesからimportできるDatabaseのインスタンスとsqlalchemyからimportできるMetaDataのインスタンスを生成します。

from fastapi import FastAPI
from sqlalchemy import MetaData

from .settings import get_settings

settings = get_settings()

def get_db_url():
    return settings.db_url


database = Database(get_db_url())
metadata = MetaData()

これらのインスタンスはormarとalembicがモデルとDBを認識するのに使われます。

最後に、FastAPIが起動したときにDBへの接続と、FastAPIが停止したときにDBから接続を解除するのが自動で行われるようにFastAPIが用意するイベントにアクセスしてそこでDBの操作が行われるようにします。

def setup_database(app: FastAPI):
    app.state.database = database

    @app.on_event("startup")
    async def startup() -> None:
        database_ = app.state.database
        if not database_.is_connected:
            await database_.connect()

    @app.on_event("shutdown")
    async def shutdown() -> None:
        database_ = app.state.database
        if database_.is_connected:
            await database_.disconnect()

これを、FastAPI起動の起点となるmain.pyで呼び出すことでFastAPIと接続します。

app/main.py

from fastapi import FastAPI

from .books_api import setup_books_api
from .configs.database import setup_database

app = FastAPI()

setup_database(app)
setup_books_api(app)

モデルの作成

次のモデルを作成します。
books_api/models.pyを開き、以下のように変更します。

app/books_api/models.py

import uuid
from datetime import datetime

import ormar

from app.configs.database import database, metadata


class Book(ormar.Model):
    pk: int = ormar.Integer(primary_key=True)
    id: uuid.UUID = ormar.UUID(default=uuid.uuid4)
    title: str = ormar.String(max_length=150)
    description: str = ormar.Text(nullable=True)
    price: int = ormar.Integer()
    unit: str = ormar.String(max_length=3)
    created_at: datetime = ormar.DateTime(default=datetime.now)
    is_active: bool = ormar.Boolean(default=True)

    class Meta(ormar.ModelMeta):
        metadata = metadata
        database = database
        tablename = "books"

まず、Bookの継承先をormar.Modelに変更します。
これによりormarのモデルとしての基本的な設定が使えるようになります。
また、これはDBに保存されるデータのモデルなので、当然プライマリーキーが必要です。
なのでプライマリーキー用のフィールドpkが追加されています。
また、pkとは別に検索など使用するためのidの追加し、これは一意性を担保するためにUUIDに設定します。
そしてこれは御作法的なのですが、いつデータが作れらたかわかるようにcreated_atというフィールドも追加しました。
その他のフィールドに関してはormarが用意しているフィールドを適応しています。
descriptionはなくても保存できるように、nullable=Trueを設定しています。

どんなフィールドがあるのかはこちらから確認してみてください。
フィールドにどんなパラメーターがあるのかはこちらを確認してみてください。

ここで特筆すべきなのはMetaクラスです。

ModelMetaを継承した上で、3つのプロパティを設定する必要があります。

• metadata
• database
• tablename

このうち、metadataとdatabaseはapp/configs/database.pyで作成したインスタンスを渡します。
tablenameはそのままそのモデルのテーブルネームになります。好きな値を設定してください。

余談

このmetadataとdatabaseを毎回モデルに設定するのは結構面倒くさいです。
それに、全てのモデルでpkは必要ですし、idやcreated_at、is_activeなども基本的にどのモデルでも必要になってくるフィールドだと思います。
今はモデルが1つなのでいいですが、10個あったらどうでしょうか？
毎回同じ内容を書いていくのは面倒くさいです。

そこで、筆者はCoreModelというものを作成して、各モデルをそれを継承する形を取ることが多いです。

import uuid
from datetime import datetime

import ormar

from app.configs.database import database, metadata


class CoreModel(ormar.Model):
    pk: int = ormar.Integer(primary_key=True)
    id: uuid.UUID = ormar.UUID(default=uuid.uuid4)
    created_at: datetime = ormar.DateTime(default=datetime.now)
    is_active: bool = ormar.Boolean(default=True)

    class Meta:
        abstract = True
        metadata = metadata
        database = database

MetaクラスでabstractをTrueに設定すると、そのモデルは継承用だとormarが認識してくれます。

あとはモデル側でこのCoreModelを継承することで全てのモデルで必要なフィールドと、metadata・databaseを1回書くだけで済みます。

Bookモデルも、上の方法を使うと次のように書くことができます。

import ormar

from app.additional.models import CoreModel


class Book(CoreModel):
    title: str = ormar.String(max_length=150)
    description: str = ormar.Text(nullable=True)
    price: int = ormar.Integer()
    unit: str = ormar.String(max_length=3)

    class Meta(ormar.ModelMeta):
        tablename = "books"

結構スッキリしましたね！
この投稿ではこの形を取りませんでしたが、皆さんのプロジェクトでは上記の形を使うのも手かと思います。

モデルをマイグレーションする

ではalembicを使ってモデルをマイグレーションしましょう。

コンソールを開き、次のコマンドを実行してください

$ alembic init migrations

このinitコマンドは、基本的に1回だけ使うことになると思います。

migrationsはマイグレーションに必要なファイルや生成されるマイグレーションファイルが格納されるフォルダ名です。プロジェクトに合わせて変更してください。

このコマンドを実行すると、alembic.iniファイルと、設定した名前のフォルダ(ここでは
migrations)とenv.pyファイルがその中に生成されます。

.
├── app
│   ├── __init__.py
│   ├── books_api
│   │   ├── __init__.py
│   │   ├── models.py
│   │   ├── routers.py
│   │   └── schemas.py
│   ├── configs
│   │   ├── __init__.py
│   │   ├── database.py
│   │   └── settings.py
│   └── main.py
├── db
│   └── dev.sqlite
├── migrations <- NEW
│   ├── README
│   ├── env.py
│   ├── script.py.mako
├── alembic.ini <- NEW

次に、migrations/env.pyを開きます。次のような中身になっていると思います。

migrations/env.py

from logging.config import fileConfig

from sqlalchemy import engine_from_config
from sqlalchemy import pool

from alembic import context

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

# 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 = None

# 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,
    )

    with connectable.connect() as connection:
        context.configure(
            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()

target_metadata = Noneを見つけて、その上に次を追加してください。

from app.books_api.models import Book
from app.configs.database import metadata
from app.configs.settings import get_settings

settings = get_settings()

そして、target_metadataにmetadataを設定します。

target_metadata = metadata

そして最後に、target_metadataの下に次を追加してください。

target_metadata = metadata

# 追記
config.set_main_option("sqlalchemy.url", settings.db_url)

では説明します。
このenv.pyでは主に3つの作業が必要です。

1. 全てのモデルをimportする
2. target_metadataに生成したmetadataをセットする
3. alembic.iniの中にあるsqlalchemy.urlをDBのURLに上書きする

(1)についてですが、これは本当に作成したモデルを全てimportする必要があります。
(3)でセットするDBのURLは、databaseインスタンスを作成するときに設定したURLと同じものを設定してください。

ここまできたらあと少しです。

次に、マイグレーションファイルを作成します。
これは自動でやるためのコマンドをalembicが用意してくれています。

コンソールを開き、次のコマンドを実行してください。

alembic revision --autogenerate -m "<write migration comments>"

-mのあとは自分の好きなメッセージを入力できます。それがそのままファイル名になるので、英語で書くことをお勧めします。

実行が完了すると、migrationsの下にversionsというディレクトリとその中に自動でマイグレーションファイルが生成されます。開くと、次のようになっていると思います。

migrations/versions/file.py

"""init book model

Revision ID: 6d944fd59427
Revises: 
Create Date: 2023-06-05 21:45:16.059904

"""
from alembic import op
import sqlalchemy as sa


# revision identifiers, used by Alembic.
revision = '6d944fd59427'
down_revision = None
branch_labels = None
depends_on = None


def upgrade() -> None:
    # ### commands auto generated by Alembic - please adjust! ###
    op.create_table('books',
    sa.Column('pk', sa.Integer(), nullable=False),
    sa.Column('id', ormar.fields.sqlalchemy_uuid.CHAR(32), nullable=True),
    sa.Column('title', sa.String(length=150), nullable=False),
    sa.Column('description', sa.Text(), nullable=True),
    sa.Column('price', sa.Integer(), nullable=False),
    sa.Column('unit', sa.String(length=3), nullable=False),
    sa.Column('created_at', sa.DateTime(), nullable=True),
    sa.Column('updated_at', sa.DateTime(), nullable=True),
    sa.Column('is_active', sa.Boolean(), nullable=True),
    sa.PrimaryKeyConstraint('pk')
    )
    # ### end Alembic commands ###


def downgrade() -> None:
    # ### commands auto generated by Alembic - please adjust! ###
    op.drop_table('books')
    # ### end Alembic commands ###

実はこれをそのままマイグレーションしようとすると、失敗します。
というのも、このファイルの中身をよく見るとわかるのですが、ormar.fields.~という行があると思います。
ormarがあるにも関わらず、ormarがimportされていないんです。
なのでormarをimportする必要があります。

migrations/versions/file.py

"""init book model

Revision ID: 6d944fd59427
Revises: 
Create Date: 2023-06-05 21:45:16.059904

"""
from alembic import op
import sqlalchemy as sa

import ormar # 追加


# revision identifiers, used by Alembic.
revision = '6d944fd59427'
down_revision = None
branch_labels = None
depends_on = None


def upgrade() -> None:
    # ### commands auto generated by Alembic - please adjust! ###
    op.create_table('books',
    sa.Column('pk', sa.Integer(), nullable=False),
    sa.Column('id', ormar.fields.sqlalchemy_uuid.CHAR(32), nullable=True),
    sa.Column('title', sa.String(length=150), nullable=False),
    sa.Column('description', sa.Text(), nullable=True),
    sa.Column('price', sa.Integer(), nullable=False),
    sa.Column('unit', sa.String(length=3), nullable=False),
    sa.Column('created_at', sa.DateTime(), nullable=True),
    sa.Column('updated_at', sa.DateTime(), nullable=True),
    sa.Column('is_active', sa.Boolean(), nullable=True),
    sa.PrimaryKeyConstraint('pk')
    )
    # ### end Alembic commands ###


def downgrade() -> None:
    # ### commands auto generated by Alembic - please adjust! ###
    op.drop_table('books')
    # ### end Alembic commands ###

これでマイグレーションができます、コンソールを開いてマイグレーションをしましょう。

alembic upgrade head

これでDBにマイグレーションがされました。
データを保存できるようになりました。

ormarモデルからPydanticモデルを作成する

ormarではモデルからPydanticのモデルを作成するメソッドを用意してくれています。

books_api/schemas.pyを開いて、次のように更新します。

app/books_api/schemas.py

from .models import Book

BookIn = Book.get_pydantic(
    exclude={
        "pk",
        "id",
        "created_at",
        "updated_at",
        "is_active",
    }
)

BookUpdateIn = Book.get_pydantic(
    exclude={
        "pk",
        "id",
        "created_at",
        "updated_at",
        "is_active",
    }
)

これでormarモデルをベースにしたPydanticのモデルが使用できるようになりました。

エンドポイントでDBを操作する

最後に、エンドポイントを更新します。

app/books_api/routers.py

from uuid import UUID

from fastapi import APIRouter, Body, FastAPI, HTTPException, Path, status

from .models import Book
from .schemas import BookIn, BookUpdateIn

router = APIRouter()

def setup_routers(app: FastAPI):
    app.include_router(router)


@router.get("/", response_model=list[Book])
async def fetch_books():
    try:
        books = await Book.objects.filter(is_active=True).all()
        return books
    except BaseException:
        status_code = status.HTTP_400_BAD_REQUEST
        message = "一覧の取得に失敗しました"
        raise HTTPException(status_code=status_code, detail=message) from None


@router.get("/{book_id}")
async def fetch_book(book_id: UUID = Path(...)) -> Book:
    try:
        book = await Book.objects.get(id=book_id, is_active=True)
        return book
    except BaseException:
        status_code = status.HTTP_400_BAD_REQUEST
        message = "詳細の取得に失敗しました"
        raise HTTPException(status_code=status_code, detail=message) from None


@router.post("/add", response_model=Book)
async def add_book(book_in: BookIn = Body(...)):
    try:
        book = Book(**book_in.dict())
        await book.save()
        return book
    except BaseException:
        status_code = status.HTTP_400_BAD_REQUEST
        message = "本の作成に失敗しました"
        raise HTTPException(status_code=status_code, detail=message) from None


@router.put("/{book_id}/update")
async def update_book(book_id: UUID = Path(...), book_update_in: BookUpdateIn = Body(...)) -> Book:
    try:
        book = await Book.objects.get(id=book_id, is_active=True)
        updated_book = await book.update(**book_update_in.dict())
        return updated_book
    except BaseException:
        status_code = status.HTTP_400_BAD_REQUEST
        message = "本の編集に失敗しました"
        raise HTTPException(status_code=status_code, detail=message) from None


@router.delete("/{book_id}/delete")
async def delete_book(book_id: UUID):
    try:
        book = await Book.objects.get(id=book_id, is_active=True)
        await book.update(is_active=False)
        return {"message": "success"}
    except BaseException:
        status_code = status.HTTP_400_BAD_REQUEST
        message = "本の削除に失敗しました"
        raise HTTPException(status_code=status_code, detail=message) from None

これで各エンドポイントで実際にDBからデータの取得・作成・更新を行えるようになります。
asyncを使った操作なので、FastAPIとの相性抜群です。

本記事はormarの導入方法がメインなので、詳しいモデルの操作方法などはこちらを確認してください。

試してみる

ここまできたらFastAPIを起動して、実際にエンドポイントを使ってみてください。
データの取得・作成・更新ができると思います。　

uvicorn app.main:app --reload

ちなみに筆者はDBの中身を確認するのにTablePlusを使用しています。

最後に

初めてやった時は結構めんどくさい印象でしたが、やることがわかるとormarを使うととても簡単にFastAPIにDBを導入できるのは良いですね。
次はTortoise ORMを使った導入のやり方を書こうかな。