FastAPIの負荷実験環境を作ってみる

.
├── api
│   ├── db.py
│   ├── main.py
│   └── model.py
├── docker
│   ├── mysql
│   │   ├── Dockerfile
│   │   ├── conf.d
│   │   │   └── my.cnf
│   │   └── initdb.d
│   │       └── schema.sql
│   └── uvicorn
│       ├── Dockerfile
│       └── requirements.txt
├── docker-compose.yml
└── locustfile.py

version: '3'
services:
  # MySQL
  db:
    container_name: "db"
    # path配下のDockerfile読み込み
    build: ./docker/mysql
    # コンテナが落ちたら再起動する
    restart: always
    tty: true
    environment:
      MYSQL_DATABASE: sample_db
      MYSQL_USER: user
      MYSQL_PASSWORD: password # ユーザのパスワード
      MYSQL_ROOT_PASSWORD: password # ルートパスワード
    ports:
      - "3306:3306"
    volumes:
      - ./docker/mysql/initdb.d:/docker-entrypoint-initdb.d # 定義通りテーブルを作成
      - ./docker/mysql/conf.d:/etc/mysql/conf.d # MySQLの基本設定(文字化け対策)
    networks:
      - local-net

  # FastAPI
  api:
    # db起動後に立ち上げる
    links:
      - db
    container_name: "api"
    build: ./docker/uvicorn
    restart: always
    tty: true
    ports:
      - 8000:8000
    volumes:
      - ./api:/usr/src/api
    networks:
      - local-net

# コンテナ間で通信を行うためのネットワークブリッジ
networks:
  local-net:
    driver: bridge

$ python -m pip install locustio

# -*- coding: utf-8 -*-
from fastapi import FastAPI
from db import session  # DBと接続するためのセッション
from model import UserTable  # Userテーブルのモデル


# fastapiでエンドポイントを定義するために必要
app = FastAPI()

# ユーザ情報を返す GET
@app.get("/users/{user_id}")
async def read_user(user_id: int):
    # DBからuser_idを元にユーザを検索
    user = session.query(UserTable).\
        filter(UserTable.id == user_id).first()
    # ユーザが見つからなかった場合
    if (user is None):
        return {"code": 404, "message": "User Not Found"}
    return user

# ユーザ情報を登録 POST
@app.post("/users")
# クエリでnameとageを受け取る
async def create_user(name: str, age: int):
    user = UserTable()
    user.name = name
    user.age = age
    session.add(user)
    session.commit()

from locust import HttpLocust, TaskSet, task, between
import random
import string


class UserTaskSet(TaskSet):
    def on_start(self):
        self.client.headers = {'Content-Type': 'application/json;'}

    @task(1)
    def fetch_user(self):
        self.client.get("/users/{}".format(random.randint(0, 100)))

    @task(2)
    def create_user(self):
        # ランダムに適当に名前を生成
        name = ''.join(random.choices(string.ascii_letters, k=10))
        age = random.randint(0, 80)
        self.client.post("/users?name={}&age={}".format(name, age))


class UserLocust(HttpLocust):
    task_set = UserTaskSet

    wait_time = between(0.100, 1.500)

$ docker-compose up -d --build

$ locust -f locustfile.py --host=http://localhost:8000

[2019-12-02 02:13:58,632] hoge/INFO/locust.main: Starting web monitor at *:8089
[2019-12-02 02:13:58,633] hoge/INFO/locust.main: Starting Locust 0.13.2