FastAPI + MySQL + Dockerを利用したAPI開発方法 メモ

• PythonのAPI構築用フレームワークFastAPIとMySQL、Dockerを利用したAPI開発方法についてメモする。

FastAPIとは

• Pythonを使ったAPI構築のためのフレームワーク(FlaskやDjangoに類する)

• Python3.6以上で動作する。

• 特徴

  • 実装が手軽。
  • 動作が早い。
  • Swagger式のAPI仕様書を作成してくれる。

  など

構成

fastapi_test    __ be   - api   - db.py
                |   |           - main.py
                |   |           - model.py
                |   L_  Dockerfile
                |   L_  requirements.txt
                |
                L_ db   __  data
                |   |
                |   L____   initdb.d    __ create_test_user_table.sql
                |   |           L_________ init_test_user.sql
                |   L____   my.cnf
                |
                L_ docker-compose.yml

docker-compose.yml

version: "3"

services:
  db:
    image: mysql:5.7
    container_name: db
    environment:
      MYSQL_ROOT_PASSWORD: rootpass
      MYSQL_DATABASE: sample_db
      MYSQL_USER: mysqluser
      MYSQL_PASSWORD: mysqlpass
    volumes:
      - ./db/data:/var/lib/mysql
      - ./db/my.cnf:/etc/mysql/conf.d/my.cnf
      - ./db/initdb.d:/docker-entrypoint-initdb.d
    ports:
      - 3308:3308
    command: --port 3308
    tty: true

  api:
    depends_on:
      - db
    container_name: "api"
    build: ./be
    ports:
      - "8000:8000"
    volumes:
      - ./be/api:/usr/src/server

データベース(db)

create_test_user_table.sql

• test_userテーブル作成用

CREATE TABLE test_user (
    id INT NOT NULL AUTO_INCREMENT,
    name VARCHAR(30) NOT NULL,
    email VARCHAR(128) NOT NULL,
    PRIMARY KEY (id)
);

init_test_user.sql

• 初期データ登録用

INSERT INTO test_user (name, email) VALUES ("tanaka taro", "[e-mail]");
INSERT INTO test_user (name, email) VALUES ("yamada hanako", "[e-mail]");

my.cnf

• DB文字コード設定

[mysqld]   
character-set-server=utf8mb4
collation-server=utf8mb4_unicode_ci
skip-character-set-client-handshake

[mysqldump]
default-character-set=utf8mb4

[client]
default-character-set=utf8mb4

API(be)

Dockerfile

• 依存ライブラリのインストールとmain編集時のリロードオプションを指定する。

FROM python:3.7

WORKDIR /usr/src/server
ADD requirements.txt .
RUN pip install -r requirements.txt

CMD ["uvicorn", "main:app", "--reload", "--host", "0.0.0.0", "--port", "8000"]

requrements.txt

• FastAPI用ライブラリ(上2つ)とMySQL接続用ライブラリ(下2つ)を指定する。

uvicorn
fastapi
mysqlclient
sqlalchemy

db.py

• DB接続設定

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

host = "db:3308"
db_name = "sample_db"
user = "mysqluser"
password = "mysqlpass"

DATABASE = 'mysql://%s:%s@%s/%s?charset=utf8' % (
    user,
    password,
    host,
    db_name,
)

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()

model.py

from sqlalchemy import Column, Integer, String
from pydantic import BaseModel
from db import Base
from db import ENGINE

# テーブル定義
class TestUserTable(Base):
    __tablename__ = 'test_user'
    id = Column(Integer, primary_key=True, autoincrement=True)
    name = Column(String(30), nullable=False)
    email = Column(String(128), nullable=False)

# モデル定義 
class TestUser(BaseModel):
    id: int
    name: str
    email: str


def main():
    # テーブル構築
    Base.metadata.create_all(bind=ENGINE)


if __name__ == "__main__":
    main()

main.py

• テストユーザーの取得、一覧取得、登録、更新API定義

from fastapi import FastAPI
from db import session
from model import TestUserTable, TestUser

app = FastAPI()

#　ユーザー情報一覧取得
@app.get("/test_users")
def get_user_list():
    users = session.query(TestUserTable).all()
    return users


# ユーザー情報取得(id指定)
@app.get("/test_users/{user_id}")
def get_user(user_id: int):
    user = session.query(TestUserTable).\
        filter(TestUserTable.id == user_id).first()
    return user


# ユーザ情報登録
@app.post("/test_users")
def post_user(user: TestUser):
    db_test_user = TestUser(name=user.name,
                            email=user.email)
    session.add(db_test_user)
    session.commit()


# ユーザ情報更新
@app.put("/test_users/{user_id}")
def put_users(user: TestUser, user_id: int):
    target_user = session.query(TestUserTable).\
        filter(TestUserTable.id == user_id).first()
    target_user.name = user.name
    target_user.email = user.email
    session.commit()

動作確認

1. コンテナ起動

docker-compose up -d --build

1. API呼び出し

   1. http://localhost:8000/docs (Swagger API仕様ページ)にアクセス
   2. 対象APIを選択し、「Try it out」を選択

   例：ユーザー情報一覧取得

     curl -X 'GET' \
       'http://localhost:8000/test_users' \
       -H 'accept: application/json'
   

     [
       {
         "id": 1,
         "name": "tanaka taro",
         "email": "[e-mail]"
       },
       {
         "id": 2,
         "name": "yamada hanako",
         "email": "[e-mail]"
       }
     ]
   

参考情報

• FastAPI
• 爆速PythonフレームワークFastAPIをMySQLに繋いでRESTfulなAPIを作成してみた。