FastAPI, MySQLで簡易Todoアプリ

本記事はPython, FastAPI, MySQLで簡単なTodoアプリ作成を通して、FastAPIを学んでいくという記事になります。

環境構築
- backend
- db
アプリ作成
- database.py
- model
- schemas
- cruds
- api
マイグレーション
CORSオリジン設定

環境構築

まずは環境構築をしてきます。下記のようにディレクトリとファイルを用意します。

.
├── backend
│   ├── Dockerfile
│   └── main.py
├── db
│   ├── Dockerfile
│   └── conf.d
│       └── mysql.conf
└── docker-compose.yml

backend

backendにはpython:3.8のイメージを使います。

FROM python:3.8
ENV PYTHONDONTWRITEBYTECODE 1
ENV PYTHONUNBUFFERED 1

WORKDIR /server
COPY Pipfile Pipfile.lock /server/
RUN pip install pipenv && pipenv install --system
COPY ./ /server

FastAPIなどをインストールしていきます。 pipのバージョンは20.2.1です。

$ cd backend
$ pipenv install fastapi uvicorn sqlalchemy alembic mysql-connector-python python-dotenv

続いてmain.pyを作成してきます。一旦はサーバの起動の確認だけを行います。

from fastapi import FastAPI

app = FastAPI()


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

docker-compose.ymlにbackend起動コマンドを書いていきます。

version: '3.7'

services:
  api:
    build:
      context: ./backend/
      dockerfile: ./Dockerfile
    command: uvicorn main:app --reload --host 0.0.0.0 --port 8000
    volumes:
      - ./backend:/server
    ports:
      - 8000:8000

最後に後でDB接続のための情報を環境変数に書いていきます。

$ pwd
~/FastAPI-example-pro/backend
$ touch .env
$ vi .env
DB_USER_NAME=todo_user
DB_PASSWORD=password
DB_HOST=db
DB_NAME=fastapi-example

準備ができたので起動して確認していきます。

$ pwd
~/FastAPI-example-pro
$ docker-compose up

下記のような文が出てきたらURLにアクセスしてみましょう。
{"message":"Hello World"}が出ていたら成功です。

api_1  | INFO:     Uvicorn running on http://0.0.0.0:8000 (Press CTRL+C to quit)
api_1  | INFO:     Started reloader process [1] using statreload
api_1  | INFO:     Started server process [9]
api_1  | INFO:     Waiting for application startup.
api_1  | INFO:     Application startup complete.

db

次にMySQLの環境を作っていきます。

初めにDockerファイルを作ります。

FROM mysql:5.7

次にmysql.confを作ります。

[mysqld]
character-set-server=utf8
default-storage-engine=INNODB
explicit-defaults-for-timestamp=true

[mysqldump]
default-character-set=utf8

[mysql]
default-character-set=utf8

[client]
default-character-set=utf8

次にMySQLの環境変数を書いていきます。

$ pwd
~/FastAPI-example-pro/db
$ touch .env
$ vi .env
MYSQL_DATABASE=fastapi-example
MYSQL_USER=todo_user
MYSQL_ROOT_PASSWORD=password
MYSQL_PASSWORD=password

最後にdocker-composeに追記していきます。

version: '3.7'

services:
  api:
    build:
      context: ./backend/
      dockerfile: ./Dockerfile
    command: uvicorn main:app --reload --host 0.0.0.0 --port 8000
    volumes:
      - ./backend:/server
    ports:
      - 8000:8000
    depends_on:
      - db

  db:
    build:
      context: ./db/
      dockerfile: ./Dockerfile
    env_file:
      - ./db/.env
    ports:
      - 3306:3306
    volumes:
      - ./db/conf.d:/etc/mysql/conf.d

まだbackend側からdbにアクセスする処理は書いていないのですが、一旦docker-compose up で起動してエラーがないことを確認していきます。

$ docker-compose up --build

以下のようなメッセージが最後の方に出たら成功です。

db_1   | 2021-05-22T08:53:02.789762Z 0 [Note] mysqld: ready for connections.
db_1   | Version: '5.7.34'  socket: '/var/run/mysqld/mysqld.sock'  port: 3306  MySQL Community Server (GPL)

以上で環境構築は終了です。

アプリ作成

次は実際にbackend側をtodoアプリのCRUDをapiで提供するように修正していきます。

作成するディレクトリとファイルは以下になります。

.
├── backend
│   ├── db
│   ├── main.py
│   └── todo
│       ├── api
│       │   ├── __init__.py
│       │   └── todo.py
│       ├── cruds
│       │   ├── __init__.py
│       │   └── todo.py
│       ├── database.py
│       ├── models
│       │   ├── __init__.py
│       │   └── todo.py
│       └── schemas
│           ├── __init__.py
│           └── todo.py

database.py

初めにdatabase.pyを作成していきます。

ここでは接続するDB情報を定義し、実際にDB接続するためのコネクションを提供します。

import os
from os.path import dirname, join

from dotenv import load_dotenv
from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker, scoped_session


load_dotenv(verbose=True)
dotenv_path = join(dirname(__file__), '.env')
load_dotenv(dotenv_path)


user_name = os.environ.get('DB_USER_NAME')
password = os.environ.get('DB_PASSWORD')
host = os.environ.get('DB_HOST')
database_name = os.environ.get('DB_NAME')

DATABASE = f'mysql+mysqlconnector://{user_name}:{password}@{host}/{database_name}?charset=utf8'

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

session = scoped_session(
    sessionmaker(autocommit=False, autoflush=False, bind=ENGINE)
)

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


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

model

次にmodelを作っていきます。これはdatabaseに作成するテーブルやカラムを定義します。 sqlalchemyを使うことでPythonのクラスで表現することができます。


from sqlalchemy import Column, Integer, String, Text

from todo.database import Base


class Todo(Base):
    __tablename__ = 'todo'

    id = Column(Integer, primary_key=True, index=True)
    title = Column(String(128), index=True)
    text = Column(Text)

schemas

次はschemaを書いていきます。

ここではクライアントからの要求とmodelとの間の差分を埋める働きをします。つまりapiのリクエストをschemaで定義したオブジェクトで取得するのに使います。

このschemaを作るのにpydanticというライブラリを使います。
pydanticはFastAPIのインストール時に一緒にインストールされます。

from pydantic import BaseModel


class TodoBase(BaseModel):
    title: str
    text: str


class TodoCreate(TodoBase):
    pass


class TodoUpdate(TodoBase):
    pass


class Todo(TodoBase):
    id: int

    class Config:
        orm_mode = True

cruds

次にcrudsを書いていきます。用意するのは以下の機能です。

Todoの取得(id指定)
Todoの取得(全部)
Todoの作成
Todoの更新
Todoの削除

では実装していきます。

from sqlalchemy.orm import Session

from todo.models import Todo
from todo.schemas import TodoCreate, TodoUpdate


def get_todo(db: Session, todo_id: int):
    return db.query(Todo).filter(Todo.id == todo_id).first()


def get_todos(db: Session, limit: int = 100):
    return db.query(Todo).limit(limit).all()


def create_todo(db: Session, todo: TodoCreate):
    db_todo = Todo(title=todo.title, text=todo.text)
    db.add(db_todo)
    db.commit()
    db.refresh(db_todo)
    return db_todo


def update_todo(db: Session, todo_id: int, todo: TodoUpdate):
    db_todo = db.query(Todo).filter(Todo.id == todo_id).first()
    db_todo.title = todo.title
    db_todo.text = todo.text
    db.commit()
    return db_todo


def delete_todo(db: Session, todo_id: int):
    db_todo = db.query(Todo).filter(Todo.id == todo_id).first()
    db.delete(db_todo)
    db.commit()

api

次にapiを作っていきます。ここはURLとのマッピング及びcrudsに処理を投げる役割をするように実装していきます。

from typing import List

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

from todo import cruds
from todo.database import get_db
from todo.schemas.todo import Todo, TodoCreate, TodoUpdate


router = APIRouter()


@router.get('/todos/{todo_id}', response_model=Todo)
def read_todo(todo_id: int, db: Session = Depends(get_db)):
    db_todo = cruds.get_todo(db, todo_id=todo_id)
    if not db_todo:
        raise HTTPException(status_code=404, detail='Todo not found')
    return db_todo


@router.get('/todos', response_model=List[Todo])
def read_todos(limit: int = 100, db: Session = Depends(get_db)):
    todos = cruds.get_todos(db, limit=limit)
    return todos


@router.post('/todos', response_model=Todo)
def create_todo(todo: TodoCreate, db: Session = Depends(get_db)):
    return cruds.create_todo(db=db, todo=todo)


@router.put('/todos/{todo_id}', response_model=Todo)
def update_todo(todo_id: int, todo: TodoUpdate, db: Session = Depends(get_db)):
    return cruds.update_todo(db=db, todo_id=todo_id, todo=todo)


@router.delete('/todos/{todo_id}')
def delete_todo(todo_id: int, db: Session = Depends(get_db)):
    cruds.delete_todo(db=db, todo_id=todo_id)

また、main.pyにメインルータの設定のみに修正します。

from fastapi import FastAPI, APIRouter

from todo.api.todo import router as todo_router


router = APIRouter()
router.include_router(
    todo_router,
    prefix='',
    tags=['todo']
)

app = FastAPI()
app.include_router(router)

マイグレーション

アプリの作成は完了しました。次にDBのマイグレーションをしていきます。

Djangoの場合はpython manage.py makemigrationsのような便利なコマンドが内臓していますがFastAPIにはありません。

そこでalembicというライブラリを使っていきます。 pydanticと合わせてFastAPIを支える重要なライブラリになります。

今回はbackend配下にdbディレクトリを作成し、ここにマイグレーション関係を作成していくことにします。

$ cd db

初期コマンドは以下になります。

$ pipenv run alembic init migrations
$ tree
.
├── alembic.ini
└── migrations
    ├── README
    ├── env.py
    ├── script.py.mako
    └── versions

alembic.iniとmigrations/env.pyを編集していきます。

# alembic.ini

( 省略)
prepend_sys_path = ../ # 修正

( 省略)
# コメントアウト
# sqlalchemy.url = driver://user:pass@localhost/dbname

( 省略)

# migrations/env.py

from logging.config import fileConfig
import os
from os.path import dirname, join

from alembic import context
from sqlalchemy import engine_from_config
from dotenv import load_dotenv
from sqlalchemy import engine_from_config, pool

from todo import models


load_dotenv(verbose=True)
dotenv_path = join(dirname(__file__), '../.env')
load_dotenv(dotenv_path)

user_name = os.environ.get('DB_USER_NAME')
password = os.environ.get('DB_PASSWORD')
host = os.environ.get('DB_HOST')
database_name = os.environ.get('DB_NAME')

DATABASE = f'mysql+mysqlconnector://{user_name}:{password}@{host}/{database_name}?charset=utf8'


# this is the Alembic Config object, which provides
# access to the values within the .ini file in use.
config = context.config
config.set_main_option('sqlalchemy.url', DATABASE)

( 省略）

編集が終わったらマイグレーションを行っていきます。

今回は開発環境にdockerを使っているので、 docker上で操作をする必要があります。

$ docker-compose run api bash
$ cd db
# マイグレーションファイル作成
$ pipenv run alembic revision --autogenerate -m 'some comments'
# マイグレーション実行
$ pipenv run alembic upgrade head

以上でマイグレーションファイルの作成とマイグレートが完了です。

CORSオリジン設定

上記でTodoAPIの作成は終わりですが、このままアクセスしても、エラーで弾かれる可能性があります。

その原因の1つにCORSオリジン設定をしていないからです。

なので、本記事の最後にこの設定をしていきます。

backend/main.pyを以下のように編集します。

from fastapi import FastAPI, APIRouter
from fastapi.middleware.cors import CORSMiddleware # 追加

from todo.api.todo import router as todo_router


router = APIRouter()
router.include_router(
    todo_router,
    prefix='',
    tags=['todo']
)

app = FastAPI()
app.include_router(router)

# 以下を追加
origins = (
    'http://0.0.0.0:8080',
    'http://localhost:8080'
)
app.add_middleware(
    CORSMiddleware,
    allow_origins=origins,
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

確認していきます。

http://0.0.0.0:8000/docs からswaggerを通して確認することもできます。

以下はPythonスクリプトで確認の実行例になります。

import requests

url = 'http://localhost:8000/todos'

# create
params = {'title': 'test title1', 'text': 'test text1'}
res = requests.post(url, json=params) # キーワード引数のjsonに渡すこと

# update
params = {'title': 'test title1', 'text': 'updated1'}
res = requests.put(f'{url}/1', json=params)

# get(all)
res = requests.get(url)

# get(id=1)
res = requests.get(f'{url}/1')

# delete
res = requests.delete(f'{url}/1')

以上です。