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

## 背景 

普段Djangoで開発をしている筆者ですが、最近そういえばFastAPIで開発していないなと思い、久しぶりに触ってみました。 
Djangoで開発しているときはDjango ORMを使ってモデルを定義してマイグレーションして...とあまりデータベース周りのことを考えずに開発をするのですが、そういえばFastAPIはORMも自分で選んで導入できるなと思いだし、色々調べていました。 
調べていく中で、[Tortoise ORM](https://tortoise.github.io)と[ormar](https://collerek.github.io/ormar/)がいい感じだったので、今回は[ormar](https://collerek.github.io/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](https://github.com/joshua-hashimoto/fastapi-ormar/tree/feature/basics)を確認してください。

## 始める前に 

この記事はFastAPIの基本的な部分はわかっているエンジニア向けの記事です。 
FastAPIは触ったことない、という人はまず[こちら](https://fastapi.tiangolo.com/tutorial/)からFastAPIに触れてみましょう。


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

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

```bash 
.
├── 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`のモデルを作成して、エンドポイント側ですぐに使えるようにします。 

```python: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`のモデルを使った形に直します。

```python: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エンドポイントを作成しました。
        - 一覧取得
        - 詳細取得
        - 作成
        - 編集
        - 削除

```python: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`があり、アプリケーション全体で使用できる定数を設定しています。 

```python: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`を作成して次のコードを入れてください。

```python: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してインスタンスを生成します。 

```python 
from .settings import get_settings

settings = get_settings()
```

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

```python 
def get_db_url():
    return settings.db_url
```

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

```python 
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の操作が行われるようにします。 

```python 
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と接続します。 

```python: 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`を開き、以下のように変更します。 

```python: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`を設定しています。

どんなフィールドがあるのかは[こちら](https://collerek.github.io/ormar/fields/field-types/)から確認してみてください。
フィールドにどんなパラメーターがあるのかは[こちら](https://collerek.github.io/ormar/fields/common-parameters/)を確認してみてください。 

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

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

- metadata
- database
- tablename

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

#### 余談 

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

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

```python 
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`モデルも、上の方法を使うと次のように書くことができます。


```python 

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`を使ってモデルをマイグレーションしましょう。 

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

```console 
$ alembic init migrations
```

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


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


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

```console 
.
├── 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`を開きます。次のような中身になっていると思います。 

```python: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`を見つけて、その上に次を追加してください。 

```python 
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`を設定します。 

```python 
target_metadata = metadata
```

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

```python 
target_metadata = metadata

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

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

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

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

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

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


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

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

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

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

```python: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する必要があります。

```python: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 ###
```

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

```console 
alembic upgrade head
```

これでDBにマイグレーションがされました。 
データを保存できるようになりました。
 
### ormarモデルからPydanticモデルを作成する 

ormarではモデルからPydanticのモデルを作成するメソッドを用意してくれています。 
 
`books_api/schemas.py`を開いて、次のように更新します。 

```python: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の導入方法がメインなので、詳しいモデルの操作方法などは[こちら](https://collerek.github.io/ormar/queries/)を確認してください。

### 試してみる

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

```console 
uvicorn app.main:app --reload
```

ちなみに筆者はDBの中身を確認するのに[TablePlus](https://tableplus.com)を使用しています。 

## 最後に 

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