🐍

FastAPI x MySQL on Docker ~ マイグレーション・API作成 ~

2020/11/05に公開

Docker

FastAPI は Python ベースのフレームワークで、その名の通り API を実装するのにうってつけです。軽量で非常に高速なフレームワークですが、FastAPI そのものには DB のシードやマイグレーションといった機能はなく、SQLAlchemy や Alembic など、他のライブラリを組み合わせて使うことが前提となります。

今回は、FastAPI で MySQL（厳密には MariaDB）を用いる際に、どのような実装が必要になってくるのかを見ていきたいと思います。セットアップは Docker コンテナ上で行います。

※ 最終成果物はこちらのリポジトリにアップしています。

ファイル・ディレクトリ構成概観

まずはじめに、ファイル・ディレクトリ構成を示します。

├── .env.development
├── app
│   ├── cruds
│   │   ├── __init__.py
│   │   └── user.py
│   ├── database.py
│   ├── env.py
│   ├── main.py
│   ├── models
│   │   ├── __init__.py
│   │   ├── book.py
│   │   ├── mixins.py
│   │   └── user.py
│   ├── routers
│   │   ├── __init__.py
│   │   └── user.py
│   └── schemas
│       ├── __init__.py
│       ├── book.py
│       └── user.py
├── db
│   ├── alembic.ini
│   ├── migrations
│   │   ├── README
│   │   ├── env.py
│   │   ├── script.py.mako
│   │   └── versions
│   │       └── [hash]_create_users_books.py
│   └── seed.py
├── docker
│   ├── api
│   │   ├── Dockerfile
│   │   └── requirements.txt
│   └── db
│       ├── Dockerfile
│       └── conf.d
│           └── my.cnf
├── docker-compose.yml
├── log
│   └── db
│       └── mysqld.log
└── scripts
    └── run.sh

DB コンテナのセットアップ

今回必要となるコンテナは FastAPI用コンテナ (api)、DB用コンテナ (db) の２つです。まずは db からセットアップしていきましょう。実際の構築には Docker Compose を用います。本セクションで扱うファイルは以下の通りです。

├── .env.development
├── docker
│   └── db
│       ├── Dockerfile
│       └── conf.d
│           └── my.cnf
└── docker-compose.yml

まずは docker-compose.yml の中身を見てみましょう。

version: "3.7"

services:
  # api: APIコンテナの記述は後ほど
  db:
    image: "fastapi_starter_db:0.1.0"
    container_name: "fastapi_starter_db"
    build:
      context: ./docker/db
      dockerfile: Dockerfile
    restart: always
    tty: true
    expose:
      - "3306"
    volumes:
      - ./docker/db/conf.d:/etc/mysql/conf.d:cached
      - ./log/db:/var/log/mysql:cached
    networks:
      - fastapi_network
    environment:
      APP_ENV: "development"
      TZ: "Asia/Tokyo"
    env_file:
      - .env.development

networks:
  default:
    external:
      name: bridge
  fastapi_network:
    name: fastapi_network
    driver: bridge
    external: true

api と db コンテナをつなぐために（そして今後更に別のコンテナから接続することも考えて）、ネットワークの設定を行っています。下記のコマンドを実行して、事前にネットワークの作成を行ってください。一度だけ実行すれば大丈夫です。

$ docker network create fastapi_network

また、DBの設定関連を .env.development で行い、docker-compose.yml でファイルを読み込むようにしています。.env.development の中身は下記の通りです。

MYSQL_USER=user
MYSQL_PASSWORD=password
MYSQL_ROOT_PASSWORD=password
MYSQL_HOST=db
MYSQL_DATABASE=db

PYTHONPATH=/usr/src/app/app

user や password は何でも良いですが、プロダクション環境の設定は git に含めないように注意しましょう。また、PYTHONPATH はAPI実装で必要になります（import 用のパスを通す）。

Dockerfile の中身は次の通りです。MySQL互換の MariaDB を用います。

FROM mariadb:10.3

RUN apt update \
    && apt install --no-install-recommends -y tzdata \
    && apt clean

RUN touch /var/log/mysql/mysqld.log

API コンテナのセットアップ

続いて、api コンテナのセットアップをしていきましょう。利用するファイルは以下の通りです。

├── .env.development
├── app
│   └── main.py
├── docker
│   ├── api
│   │   ├── Dockerfile
│   │   └── requirements.txt
│   └── db
│       ├── Dockerfile
│       └── conf.d
│           └── my.cnf
├── docker-compose.yml
└── scripts
    └── run.sh

docker-compose.yml の中身を見てみましょう。先ほどの内容に api 部分を追記します。

version: "3.7"

services:
  api:
    image: "fastapi_starter:1.0.0"
    container_name: "fastapi_starter"
    depends_on:
      - db
    build:
      context: ./docker/api
      dockerfile: Dockerfile
    ports:
      - "8888:8000"
    volumes:
      - ./app:/usr/src/app/app:cached
      - ./db:/usr/src/app/db:cached
      - ./scripts:/usr/src/app/scripts:cached
    working_dir: /usr/src/app/scripts
    command: bash -c "./run.sh"
    networks:
      - fastapi_network
    environment:
      APP_ENV: "development"
      TZ: "Asia/Tokyo"
    env_file:
      - .env.development
  db:
    # 省略

networks:
  # 省略

こちらの Dockerfile は python コンテナをベースにしています。また、api コンテナから MySQL に接続するため、MySQLクライアントをインストールしています。Dockerfile および requirements.txt の中身は以下の通りです。

# Dockerfile

FROM python:3.8

RUN apt update \
    && apt install -y default-mysql-client \
    && apt install --no-install-recommends -y tzdata \
    && apt clean

WORKDIR /usr/src/app
ADD requirements.txt .
RUN pip install -U pip \
    && pip install --trusted-host pypi.python.org -r requirements.txt

# requirements.txt

alembic==1.4.3
fastapi==0.61.1
# gunicorn==20.0.4
mysqlclient==2.0.1
pytest==6.1.2
SQLAlchemy==1.3.20
sqlalchemy_utils==0.36.8
uvicorn==0.12.2

requirements.txt 内の gunicorn は、プロダクション環境でサーバー立ち上げの際に用いるパッケージになります（uvicorn でも問題なく動くには動くのですが、基本は開発環境向けです）。プロダクション環境用の設定関連は、また別の機会に紹介したいと思います。

api コンテナでは、起動時に下記の通り ./run.sh を実行しています。

command: bash -c "./run.sh"

この run.sh の中身は下記の通りです。

#!/bin/bash

echo "Waiting for mysql to start..."
until mysql -h"$MYSQL_HOST" -u"$MYSQL_USER" -p"$MYSQL_PASSWORD" &> /dev/null
do
    sleep 1
done

cd /usr/src/app/app && uvicorn main:app --reload --port=8000 --host=0.0.0.0

ここでは、API サーバーのDB接続エラーを回避するための処理を記述しています。docker-compose.yml で depends_on: db を記述していますが、コンテナが立ち上がってから MySQL が起動するまでにタイムラグがあるため、until - do - done 内で、MySQLの起動を確実に待ってからサーバー立ち上げをするようにしています。

もしデータベースを用いず、db コンテナが不要ならば、command: の部分は下記のサーバー起動だけで問題ありません。

command: "uvicorn main:app --reload --port=8000 --host=0.0.0.0"

まだモデルの実装は行っていないので DB を用いることはないのですが、現時点で main.py を次のように実装し、docker-compose up を実行すると、２つのコンテナが無事に起動するのが確認できるかと思います。

from fastapi import FastAPI

app = FastAPI()


@app.get('/')
async def root():
    return {'message': 'Hello World'}

ホストのポートは 8888 に設定してあるので、サーバー起動後に http://localhost:8888/docs にアクセスすると、以下のようなAPI一覧の画面が表示されるかと思います。

各APIセクション右上の Try it out をクリックし、Execute をクリックすると、APIを叩いた結果を確認することができます。

モデルの実装

※ このセクションは公式チュートリアルも見ながら進めると理解が深まるかと思います。

さて、ここからは API の実装を考えていきます。まずはモデルの実装です。今回は User クラスと Book クラスを準備し、User と Book は一対多の関係にあるものとします（図書館の本を誰が借りているか、という状況を考えるとイメージしやすいかもしれません）。 User は uuid と username、Book は uuid と title を持ち、また両者ともにタイムスタンプ（created_at, updated_at）を持つとします。

実装していくのは以下のファイルとなります。

└── app
    ├── database.py
    ├── env.py
    └── models
        ├── __init__.py
        ├── book.py
        ├── mixins.py
        └── user.py

モデル定義自体は models/ にありますが、実装には env.py および database.py が必要となるため、まずは env.py を見てみましょう。こちらは API から DB に接続するための環境変数を読み込んでいるファイルとなります。

import os

APP_ENV = os.environ.get('APP_ENV')

DB_USER = os.environ.get('MYSQL_USER')
DB_PASSWORD = os.environ.get('MYSQL_PASSWORD')
DB_HOST = os.environ.get('MYSQL_HOST')
DB_NAME = os.environ.get('MYSQL_DATABASE')

続いて database.py です。こちらは実際に DB に接続するための実装を行っています。最後に定義している get_db が、各API実装から呼び出されることになります。

from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker, scoped_session
from env import DB_USER, DB_PASSWORD, DB_HOST, DB_NAME

DATABASE = 'mysql://%s:%s@%s/%s?charset=utf8' % (
    DB_USER,
    DB_PASSWORD,
    DB_HOST,
    DB_NAME,
)

engine = create_engine(
    DATABASE,
    encoding='utf-8',
    echo=True
)

# 実際の DB セッション
SessionLocal = scoped_session(
    sessionmaker(
        autocommit=False,
        autoflush=False,
        bind=engine
    )
)

Base = declarative_base()
Base.query = SessionLocal.query_property()


# Dependency Injection用
def get_db():
    try:
        db = SessionLocal()
        yield db
    finally:
        db.close()

この database.py 内で定義している Base がモデルの定義で必要になってきます。では、実際のモデルの実装を見ていきましょう。user.py および book.py の中身はそれぞれ下記のようになります。

# user.py

from database import Base
from sqlalchemy import Column, String
from sqlalchemy.orm import relationship
from sqlalchemy_utils import UUIDType
from uuid import uuid4
from .mixins import TimestampMixin  # 後述


class User(Base, TimestampMixin):
    __tablename__ = 'users'

    uuid = Column(UUIDType(binary=False),
                  primary_key=True,
                  default=uuid4)
    username = Column(String(128), nullable=False)

    # リレーション設定
    books = relationship(
        'Book',
        back_populates='user'
    )

# book.py

from database import Base
from sqlalchemy import Column, ForeignKey, String
from sqlalchemy.orm import relationship
from sqlalchemy_utils import UUIDType
from uuid import uuid4
from .mixins import TimestampMixin  # 後述


class Book(Base, TimestampMixin):
    __tablename__ = 'books'

    uuid = Column(UUIDType(binary=False),
                  primary_key=True,
                  default=uuid4)
    title = Column(String(256), nullable=False)

    # リレーション設定
    user_id = Column(
        UUIDType(binary=False),
        ForeignKey('users.uuid'),
        nullable=True
    )
    user = relationship(
        'User',
        back_populates='books'
    )

両者でインポートしている TimestampMixin は、mixins.py に以下のように実装しています。

# mixins.py

from sqlalchemy import Column, text
from sqlalchemy.dialects.mysql import TIMESTAMP as Timestamp


class TimestampMixin(object):
    created_at = Column(Timestamp, nullable=False,
                        server_default=text('current_timestamp'))
    updated_at = Column(Timestamp, nullable=False,
                        server_default=text('current_timestamp on update current_timestamp'))

モデルの実装は以上ですが、マイグレーションを考える場合、__init__.py にモデルクラスをまとめてインポートしておくと、後々の実装が楽になります（ファイル内の noqa: F401 は Lint 用の記述です）。

# __init__.py

from database import Base  # noqa: F401

from .book import Book  # noqa: F401
from .user import User  # noqa: F401

モデルクラスの実装はできましたが、これだけではDBに反映されるわけではないので、別途マイグレーションの設定が必要になります。続いて、この設定方法について見ていきます。

マイグレーション設定

冒頭で触れたように、FastAPI自体にマイグレーション機能はなく、Alembic を用いることになります。セットアップするためには、コンテナ内で作業をしていく必要があります。下記でコンテナ内に入ります。

$ docker-compose run api bash

ホストのプロジェクトルートはコンテナ内では /usr/src/app/ に対応しており、DB関連は /usr/src/app/db/ まとめていきます。

# cd /usr/src/app/db

ディレクトリに移動したら、下記を実行することで設定ファイルが生成されます。

# alembic init migrations

└── db
    ├── alembic.ini
    └── migrations
        ├── README
        ├── env.py
        ├── script.py.mako
        └── versions

db ディレクトリ以下に、alembic.ini および migrations ディレクトリが生成されているのを確認してください。alembic.ini および migrations/env.py を編集することで、マイグレーションができるようになります。

まずは alembic.ini ですが、下記のように sqlalchemy.url 部分をコメントアウトします。

# alembic.ini

# A generic, single database configuration.

[alembic]

...

# 下行をコメントアウト（該当部分は env.py に移動する）
# sqlalchemy.url = driver://user:pass@localhost/dbname

[post_write_hooks]

...

続いて env.py です。こちらはいくつか追加の設定が必要です。下記コード内にコメントで書いていますが、特に uuid を使いたい場合、デフォルトでは alembic はマイグレーションに対応していませんので、自分で関数 (render_item) を実装する必要があります。

# migrations/env.py

# インポート追加
import os
import sqlalchemy_utils
import models
# インポート追加ここまで

...

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

# DB接続用設定追加
# connect to database
DB_USER = os.environ.get('MYSQL_USER')
DB_PASSWORD = os.environ.get('MYSQL_PASSWORD')
DB_ROOT_PASSWORD = os.environ.get('MYSQL_ROOT_PASSWORD')
DB_HOST = os.environ.get('MYSQL_HOST')
DB_NAME = os.environ.get('MYSQL_DATABASE')

DATABASE = 'mysql://%s:%s@%s/%s?charset=utf8' % (
    DB_USER,
    DB_PASSWORD,
    DB_HOST,
    DB_NAME,
)
config.set_main_option('sqlalchemy.url', DATABASE)
# 設定追加ここまで

# Interpret the config file for Python logging.
# This line sets up loggers basically.
fileConfig(config.config_file_name)

# add your model's MetaData object here
# for 'autogenerate' support
target_metadata = models.Base.metadata  # 追加

...

# UUIDのマイグレーション対応用関数追加
def render_item(type_, obj, autogen_context):
    """Apply custom rendering for selected items."""

    if type_ == 'type' and isinstance(
            obj, sqlalchemy_utils.types.uuid.UUIDType):
        autogen_context.imports.add("import sqlalchemy_utils")
        autogen_context.imports.add("import uuid")
        return "sqlalchemy_utils.types.uuid.UUIDType(binary=False), default=uuid.uuid4"

    return False
# 関数追加ここまで

def run_migrations_offline():
    ...  # 変更なし

def run_migrations_online():
    ...

    with connectable.connect() as connection:
        context.configure(
            connection=connection, target_metadata=target_metadata,
            render_item=render_item  # 追加
        )

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

...

これでマイグレーション用の設定は完了です。/usr/src/app/db 上で、下記を実行しましょう。

# alembic revision --autogenerate -m 'create users, books'

これにより、モデルファイルの構成が読み込まれ（env.py に記述した models.Base.metadata の部分）、現状のDBのテーブルとの差分から自動でマイグレーションファイルが db/migrations/versions/ 以下に生成されます。

└── db
    └── migrations
        └── versions
            └── 402b2dfeb53e_create_users_books.py

これ以降も、モデルファイルを変更し、マイグレーションを行いたいタイミングで alembic revision --autogenerate -m 'コメント' を実行することで、versions/ 以下にファイルが追加されていきます。

そして、実際にマイグレーションをする場合は下記を実行します。

# alembic upgrade head

複数のマイグレーションファイルが生成されており、１つずつマイグレーションしたい場合は、

# alembic upgrade +1

を実行します。また、反対にマイグレーションを１つ取り消したい場合は、

# alembic downgrade -1

すべて取り消したい場合は

# alembic downgrade base

となります。

このように、マイグレーション自体は手動で行うこととなりますが、自動で行うようにしたい場合は、scripts/run.sh を以下のように書き換えると、コンテナ起動時にマイグレーションが実行されるようになります。

#!/bin/bash

echo "Waiting for mysql to start..."
until mysql -h"$MYSQL_HOST" -u"$MYSQL_USER" -p"$MYSQL_PASSWORD" &> /dev/null
do
    sleep 1
done

# 自動マイグレーションを追加
cd /usr/src/app/db && alembic upgrade head


cd /usr/src/app/app && uvicorn main:app --reload --port=8000 --host=0.0.0.0

マイグレーション関連の実装は以上です！

シード設定

シードを行いたい場合は、自分で Python スクリプトを書くことになります。以下のように、db/seed.py を作成して、そこに実装をしていきましょう。

└── db
    └── seed.py

今回のシードデータは User １人、Book ３冊で、すべての Book はそのひとりの User が所持しているものとします。

# seed.py

from database import SessionLocal
from models import Book, User

db = SessionLocal()


def seed():
    book_titles = [
        '深層学習教科書 ディープラーニング G検定(ジェネラリスト) 公式テキスト',
        '詳解ディープラーニング',
        'PythonとKerasによるディープラーニング'
    ]
    books = [Book(title=title) for title in book_titles]

    user = User(username='yusugomori')
    user.books = books  # リレーションも丸ごと表現できる

    db.add(user)
    db.commit()


if __name__ == '__main__':
    BOS = '\033[92m'  # 緑色表示用
    EOS = '\033[0m'

    print(f'{BOS}Seeding data...{EOS}')
    seed()

シードもコンテナ内で実行します。下記の手順でシードを行うことができます。

$ docker-compose run api bash
# cd /usr/src/app/db
# python seed.py

シードも自動で行いたい場合、マイグレーションの時と同様、run.sh に記述するなどすると良いでしょう。

APIの実装

ここまででデータ関連の実装はすべて終えたので、最後に API の実装をしてデータの取得ができるか確認してみましょう。User の Read API（Index, Detail）のみ実装してみます。

実装方法は様々ですが、下記の構成にすると、コードの可読性が高まるかと思います。

└── app
    ├── cruds
    │   ├── __init__.py
    │   └── user.py
    ├── database.py
    ├── env.py
    ├── main.py
    ├── models
    │   ├── __init__.py
    │   ├── book.py
    │   ├── mixins.py
    │   └── user.py
    ├── routers
    │   ├── __init__.py
    │   └── user.py
    └── schemas
        ├── __init__.py
        ├── book.py
        └── user.py

新しく追加されたのは cruds, routers, schemas です。それぞれ、以下の役割を担います。

cruds:    実際にDBとやりとりを行う
routers:  URLの定義および cruds の呼び出し
schemas:  Response の型定義

まずは cruds/user.py を見てみましょう。DB からデータを読み込む実装を行っています（例外処理などは簡略化しています）。

# cruds/user.py

from fastapi import HTTPException
from models import User
from sqlalchemy.orm import Session
from starlette.status import HTTP_404_NOT_FOUND
from uuid import UUID


def read_users(db: Session):
    items = db.query(User).all()
    return items


def read_user(db: Session, user_id: UUID):
    try:
        item = db.query(User).get(user_id)
    except BaseException:
        raise HTTPException(status_code=HTTP_404_NOT_FOUND,
                            detail='Record not found.')

    return item

続いて、この cruds を呼び出す routers/user.py の実装です。 User API の URL は /users や /users/:user_id などが想定されますが、接頭辞の /users は後ほどまとめて設定できるので、ここでは接頭辞以降のみを記述すれば問題ありません。

# routers/user.py

import cruds.user as crud
from database import get_db
from fastapi import APIRouter, Depends
# schemas の中身は後述
from schemas.user import \
    User as UserSchema, UserDetail as UserDetailSchema
from sqlalchemy.orm import Session
from typing import List
from uuid import UUID

router = APIRouter()


# URL は /users 以降のみを書けばOK
@router.get('/', response_model=List[UserSchema])
async def read_users(db: Session = Depends(get_db)):
    return crud.read_users(db=db)


@router.get('/{user_id}', response_model=UserDetailSchema)
async def read_user(user_id: UUID, db: Session = Depends(get_db)):
    return crud.read_user(user_id=user_id, db=db)

各 API で response_model= を指定していますが、ここでデータのどの値を返すかを細かく指定することができます。逆に、これを指定しないと、リレーション込みでデータを返すことができないなど、不便な点が発生します。中身に関しては schemas/user.py, schemas/book.py で定義しています。実装は以下の通りです。

# schemas/user.py

# from datetime import datetime
from typing import List
from pydantic import BaseModel
from uuid import UUID
from .book import Book


class User(BaseModel):
    uuid: UUID
    username: str
    # 返す必要のない値は記述しなければ
    # レスポンスから切り捨てられる
    # created_at: datetime
    # updated_at: datetime

    # リレーションを張っている場合は
    # 書かないとバリデーションエラーになる
    class Config:
        orm_mode = True


class UserDetail(User):
    books: List[Book] = []

# schemas/book.py

from pydantic import BaseModel
from uuid import UUID


class Book(BaseModel):
    uuid: UUID
    title: str

    class Config:
        orm_mode = True

最後に、先ほど実装した routers/user.py を main.py で読み込み、ルーティングの設定をすれば実装は完了です！

# main.py

from fastapi import FastAPI, APIRouter
from routers.user import router as user_router

router = APIRouter()
router.include_router(
    user_router,
    prefix='/users',
    tags=['users']
)

app = FastAPI()
app.include_router(router)

以上を実装し終えたら、再度 docker-compose up でコンテナを起動した後（またはすでに起動済ならばそのまま）、http://localhost:8888/docs にアクセスすると、User APIs が追加されているのが確認でき、それぞれ実行してみると、期待通りのレスポンスが得られることが分かります。

長くなりましたが、今回の内容は以上になります。FastAPI は非常に簡単に API が実装できるフレームワークになっていますので、ぜひ活用してみてください！