久々に開発ネタです.
- 野球のデータをシュッと見るためのDashboardを作る（理由は後ほど）.
- そんなDashboardのBackend APIをシュッと開発する.
を目標に立て現在進行系でやってるのですが,
午後の進捗その2
— Shinichi Nakagawa (@shinyorke) 2019年12月31日
Docker化が特に滞りなく完了.
API Docも見れるとかFast API強すぎぃ
昨日の夕方から開発してたAPIはアッサリ1st Ver.できたので, 大晦日の買い物終わったらフロントエンドを除夜の鐘が鳴るまでになんとかするぞ #大晦日ハッカソン pic.twitter.com/wWMiSvQDKu
Backendを昨日（12/30）の18:00から着手して（実質作業時間）約5時間ちょいで完成させてしまいました.
本年最後のブログは,
- FastAPIでバックエンドを5時間ちょいで作った話.
- それもカウボーイコーディングじゃなくてちゃんときれいに作ったよ, クリーンアーキテクチャ意識して.
っていう記録を備忘録的に残したいと思います.
なお, コードは公開しませんが必要なスニペットは載せますので参考にしてもらえると嬉しいです.
おしながき
- おしながき
- 前提条件
- 作ったモノ
- そもそもFastAPI #とは
- Docker（Docker Compose）
- Application（Router/Controller）
- Schema/Model/Database
- 実はクリーンアーキテクチャなのだ
- これを何に使うか
前提条件
一応前提条件書いときます.
- ローカル環境（Macbook Pro + Docker Desktop）の環境で動かしてます, クラウドサービス等には上げていない.*1
- データは事前にマイグレーションしてDocker上のMySQLで取得できる状態*2
- デモなので, ユーザー認証は作ってません*3
- 新規プロダクトのデモかつゼロからやってるのでテストおよびValidatorは未実装です.*4
- 参照しかしないアプリなため, post/put/deleteに当たるAPIは未実装です（そもそも仕様にないので作らないつもり.）.
ちなみにデータはSean Lahmanというメジャーリーグのデータベースを使っています, DB化は（ちょっと古いですが）過去記事が参考になると思います.
これらの前提条件のもと, 5時間で完成させました.
作ったモノ
野球のデータを取得するAPIで,
- 選手ID（player_id）から選手Profileを取得
- 同じくplayer_idから打撃成績（年度の降順）で取得
というAPIを作りました.
スクショで魅せるとこんなかんじ（サンプルはShohei Ohtaniさんです*5）.
起動後, http://0.0.0.0:8000/docs と打つとAPIのドキュメントが登場.
このドキュメントはそのままSwaggerチックなモノになっててこの場から直接実行することもできます, 強い.
実際にAPIを叩いてプロフィール出してみました, ちゃんとできてる.
打撃成績も今年の分までしっかり出てくれました.
以下の順でコードベースで追いかけていきます.
- Docker（Docker Compose）
- Application（Router/Controller）
- Schema/Model/Database
そもそもFastAPI #とは
一言で言うとPython製の非同期前提なWeb APIをいい感じに作るためのFrameworkです.
あくまで自分の解釈ですが,
- バックエンドが欲しい時にシュッと作れるWeb APIのFramework
- チャットなどで非同期処理前提のサーバがほしい時に重宝する「ASGI（Asynchronous Server Gateway Interface） 」に対応した今どきなFrameworkでもある
- 開発してる感覚的には, Flaskそのもの
といったモノです.
Hello worldを見ると一目瞭然かもしれません.
from fastapi import FastAPI app = FastAPI() @app.get("/") def read_root(): return {"Hello": "World"} @app.get("/items/{item_id}") def read_item(item_id: int, q: str = None): return {"item_id": item_id, "q": q}
デコレータで囲う感覚もそうですが, 他も似てます.
？？？「だったらFlaskでいいのでは？」
と思われるかもですが,
- 年末年始休みで時間あるので普段使ってないものにしたかった.*6
- 当初はresponderを想定していたが, 若干方言ある・依存ライブラリ多いのでちょっと...
- 自社（JX通信社）で使ってるFastAPIを同僚にオススメされたのでやってこ！ってなった.*7
ちなみに（うっすらですが）自社でもFastAPIを（自分とは別のチームで）使っており, サンプルめいたものはTechブログでもちょっと紹介されています.
Docker（Docker Compose）
まずDockerで動かしたモノの設定...の前にざっくりなプロジェクト構成です.
.
├── Dockerfile
├── Pipfile
├── Pipfile.lock
├── app.py
├── sabr_analytics
│ ├── __init__.py
│ └── router.py
├── docker-compose-local.yml
├── lahman
│ ├── __init__.py
│ ├── crud.py
│ ├── models.py
│ ├── router.py
│ └── schemas.py
└── service
├── __init__.py
└── database.py
Applicationとしてのエントリーポイントはapp.pyでして, Local Debug時は,
$ python app.py
もしくは, ASGIサーバーのuvicornを使って,
$ uvicorn app:api --reload
これで動きます.
Docker使ってのデバッグですが,
$ docker-compose -f docker-compose-local.yml up -d
これで済むようにしています.*8
そんなdocker-compose-local.ymlの内容はこちら（networksは事前に作ってます）.
version: "3.0" services: api: build: . image: shiyorke/yakiu/api container_name: yakiu_api env_file: .env ports: - "8000:8000" networks: default: external: name: baseball_network
そしてDockerfileはこれ.
pipenvで必要なモノを入れて動かしておしまいです.
FROM python:3.8
LABEL maintainer "shinyorke"
RUN set -ex \
&& apt-get update -y --fix-missing \
&& apt-get install -y -q --no-install-recommends \
curl \
file \
&& apt-get purge -y --auto-remove
# install
RUN pip install pipenv
ADD Pipfile Pipfile.lock /
RUN pipenv install --system
# add to application
ADD app.py /
ADD service service
ADD sabr_analytics sabr_analytics
ADD lahman lahman
EXPOSE 8000
CMD ["uvicorn", "app:api", "--host", "0.0.0.0", "--port", "8000"]
ちなみにPipfileはこちら.
[[source]] name = "pypi" url = "https://pypi.org/simple" verify_ssl = true [dev-packages] [packages] fastapi = "*" uvicorn = "*" sqlalchemy = "*" cryptography = "*" pymysql = "*" sabr = "*" [requires] python_version = "3.7"
メインで使ってるパッケージが少なくて気持ち的に楽です.
Application（Router/Controller）
Application本体です.
まずすべてのエントリーポイントとなってるapp.pyの実装.
import uvicorn from fastapi import FastAPI from lahman import router as api_lahman from sabr_analytics import router as api_analytics api = FastAPI() @api.get("/") async def dashboard(): return { 'playerIds': ['Ichiro', 'Altuve', 'Trout', 'Shohei'], 'api_batting': 'lahman', 'api_analytics': 'analytics' } api.include_router( api_lahman.router, prefix="/lahman", tags=["lahman"], ) api.include_router( api_analytics.router, prefix="/analytics", tags=["analytics"], ) if __name__ == "__main__": uvicorn.run(api, host="0.0.0.0", port=8000)
APIのエントリーポイント（URL）は少なめですが, 拡張性を考慮して最初からRoterでモジュールを刺す形で開発をはじめました.
これは開発当初からこの仕掛を作りたくて公式ドキュメントを探してましたがアッサリ見つかりました.
Bigger Applications - Multiple Files - FastAPI
これは明確な理由があって,
- 普通の野球成績データはlahmanパッケージに留める
- 今後開発する（昨日今日では作ってない）独自のモデルで分析した結果は別のパッケージとして明確に切り分けたい
という方針を最初から決めており, それによりサブドメインも分かれるのがわかりきってたので最初からrouter前提にしました.
ちなみに構成的には,
.
├── app.py # メインのアプリ. ここのrouterに刺したものだけ動く
├── sabr_analytics # 独自分析モデルのアプリ, これから開発
│ ├── __init__.py
│ └── router.py
├── lahman # 野球の成績データAPI, 今回メインで開発
│ ├── __init__.py
│ ├── crud.py
│ ├── models.py
│ ├── router.py
│ └── schemas.py
└── service # DBコネクション, 外部API接続などパッケージ横断で共通して使うもの
├── __init__.py
└── database.py
と明確に分けて開発してます.
この先メインで解説するlahmanパッケージのrouter定義（lahman/router.py）はこんな感じです.
from fastapi import APIRouter, Depends, HTTPException from sqlalchemy.orm import Session from typing import List from lahman import schemas from lahman.crud import Database from service.database import session router = APIRouter() @router.get("/") async def index(): return { 'apis': [ '/player', '/batting' ] } @router.get("/player/{player_id}", response_model=schemas.Player) async def get_player(player_id: str, db: Session = Depends(session)): """ Player Profile :param player_id: lahman playerID :param db: DB Session :return: Player Profile :rtype: JSON :exception: HTTPException(404) """ player = Database.get_player(db, player_id) if player: return player raise HTTPException(status_code=404, detail=f"player not found: {player_id}") @router.get("/batting/{player_id}", response_model=List[schemas.BattingTotal]) async def get_batting(player_id: str, db: Session = Depends(session)): """ Batting Stats :param player_id: lahman playerID :param db: DB Session :return: Batting Stats :rtype: JSON """ batting_stats = Database.get_batting(db, player_id) return batting_stats
めちゃくちゃシンプルですね.
Schema/Model/Database
データモデルの定義などがどうなってるか解説です.
こちらも公式ドキュメントを参考にアレンジしました.
SQL (Relational) Databases - FastAPI
lahmanパッケージのrouter定義（lahman/router.py）より.
関数のこちらに着目.
@router.get("/player/{player_id}", response_model=schemas.Player) async def get_player(player_id: str, db: Session = Depends(session)): """ Player Profile :param player_id: lahman playerID :param db: DB Session :return: Player Profile :rtype: JSON :exception: HTTPException(404) """ player = Database.get_player(db, player_id) if player: return player raise HTTPException(status_code=404, detail=f"player not found: {player_id}")
response_model=schemas.Player
ここは名前の通り戻り値を決めています, 具体的にはpydanticというライブラリでいい感じに定義しています.
Django REST Framework（DRF）で言う所のserializerに当たる箇所.
※ソースコードは「lahman/schema.py」
from datetime import datetime import pydantic from pydantic import BaseModel from sabr.stats import Stats class BattingBase(BaseModel): playerID: str yearID: int G: int = None G_batting: int = None AB: int = None R: int = None H: int = None DOUBLE: int = None TRIPLE: int = None HR: int = None SINGLE: int = None RBI: int = None SB: int = None CS: int = None BB: int = None SO: int = None IBB: int = None HBP: int = None SH: int = None SF: int = None GIDP: int = None G_old: int = None BA: float = None OBP: float = None SLG: float = None OPS: float = None BABIP: float = None @pydantic.validator('SINGLE', pre=True, always=True) def single(cls, v, *, values, **kwargs): """ Single Hits :param v: Default Value :param values: Row Values :param kwargs: Option :return: SINGLE """ if values['AB'] > 0: return values['H'] - (values['HR'] + values['TRIPLE'] + values['DOUBLE']) return 0.0 @pydantic.validator('BA', pre=True, always=True) def ba(cls, v, *, values, **kwargs): """ Batting Average :param v: Default Value :param values: Row Values :param kwargs: Option :return: BA """ if values['AB'] > 0: return Stats.avg(h=values['H'], ab=values['AB']) return 0.0 @pydantic.validator('OBP', pre=True, always=True) def obp(cls, v, *, values, **kwargs): """ On Base Per :param v: Default Value :param values: Row Values :param kwargs: Option :return: OBP """ return Stats.obp(h=values['H'], bb=values['BB'], hbp=values['HBP'], sf=values['SF'], ab=values['AB']) @pydantic.validator('SLG', pre=True, always=True) def slg(cls, v, *, values, **kwargs): """ Slugging :param v: Default Value :param values: Row Values :param kwargs: Option :return: OBP """ if values['AB'] > 0: single = values['H'] - (values['HR'] + values['TRIPLE'] + values['DOUBLE']) tb = Stats.tb(single=single, _2b=values['DOUBLE'], _3b=values['TRIPLE'], hr=values['HR']) return Stats.slg(tb=tb, ab=values['AB']) return 0.0 @pydantic.validator('OPS', always=True) def ops(cls, v, *, values, **kwargs): """ On Base Plus Slugging :param v: Default Value :param values: Row Values :param kwargs: Option :return: OPS """ return values['OBP'] + values['SLG'] @pydantic.validator('BABIP', always=True) def babip(cls, v, *, values, **kwargs): """ BABIP :param v: Default Value :param values: Row Values :param kwargs: Option :return: BABIP """ return Stats.babip(h=values['H'], hr=values['HR'], ab=values['AB'], sf=values['SF'], so=values['SO']) class Batting(BattingBase): teamID: str lgID: str stint: int class Config: orm_mode = True class BattingTotal(BattingBase): class Config: orm_mode = True class PlayerBase(BaseModel): playerID: str birthYear: int birthMonth: int birthDay: int nameFirst: str nameLast: str bats: str throws: str class Player(PlayerBase): playerID: str birthCountry: str birthState: str birthCity: str nameGiven: str weight: int height: float debut: datetime finalGame: datetime retroID: str bbrefID: str class Config: orm_mode = True
validatorデコレータでモデルに無い値（例えば打率（BA）など）を計算してるのが若干気持ち悪いですがこれが正しい使い方っぽいです苦笑.
もちろんDBからデータを探してSchemaにマッピングするのですが, これはSQLAlchemyを使っています（Python使いならおなじみ*9ですね）.
Databaseクラスには検索のための関数を生やしており,
from sqlalchemy.orm import Session from sqlalchemy import desc from lahman import models class Database: @staticmethod def get_player(db: Session, player_id: str): """ Get Player :param db: SQLAlchemy Session :param player_id: lahman db playerID :return: Player Profile """ return db.query(models.People).filter(models.People.playerID == player_id).first() @staticmethod def get_batting(db: Session, player_id: str): """ Get Batting Stats :param db: SQLAlchemy Session :param player_id: lahman db playerID :return: Batting Stats(Years Desc) """ return db.query(models.BattingTotal).order_by(desc(models.BattingTotal.yearID)) \ .filter(models.BattingTotal.playerID == player_id).all()
モデルはいつもどおりSQLAlchemyでいい感じに定義です.
from sqlalchemy import Column, DateTime, Float, Integer, String, text from service.database import Base class People(Base): __tablename__ = u'People' playerID = Column(String(10), primary_key=True) birthYear = Column(Integer) birthMonth = Column(Integer) birthDay = Column(Integer) birthCountry = Column(String(50)) birthState = Column(String(2)) birthCity = Column(String(50)) deathYear = Column(Integer) deathMonth = Column(Integer) deathDay = Column(Integer) deathCountry = Column(String(50)) deathState = Column(String(2)) deathCity = Column(String(50)) nameFirst = Column(String(50)) nameLast = Column(String(50)) nameGiven = Column(String(255)) weight = Column(Integer) height = Column(Float(asdecimal=True)) bats = Column(String(1)) throws = Column(String(1)) debut = Column(DateTime) finalGame = Column(DateTime) retroID = Column(String(9)) bbrefID = Column(String(9)) class BattingTotal(Base): __tablename__ = u'BattingTotal' playerID = Column(String(9), primary_key=True, nullable=False, server_default=text("''")) yearID = Column(Integer, primary_key=True, nullable=False, server_default=text("'0'")) G = Column(Integer) G_batting = Column(Integer) AB = Column(Integer) R = Column(Integer) H = Column(Integer) DOUBLE = Column(u'2B', Integer) TRIPLE = Column(u'3B', Integer) HR = Column(Integer) RBI = Column(Integer) SB = Column(Integer) CS = Column(Integer) BB = Column(Integer) SO = Column(Integer) IBB = Column(Integer) HBP = Column(Integer) SH = Column(Integer) SF = Column(Integer) GIDP = Column(Integer) G_old = Column(Integer)
データベースの関数はserviceパッケージにまとめています.
なぜならConnectionはパッケージ関係なく使うからです（こなみ）.
import os from sqlalchemy import create_engine from sqlalchemy.ext.declarative import declarative_base from sqlalchemy.orm import sessionmaker params = { 'user': os.getenv('DB_USER'), 'password': os.getenv('DB_PASSWORD'), 'host': os.getenv('DB_HOST'), 'database': os.getenv('DATABASE') } DATABASE_URL = "mysql+pymysql://{user}:{password}@{host}:3306/{database}".format(**params) engine = create_engine(DATABASE_URL) SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine) Base = declarative_base() def session(): """ Get Database Session :return: """ db = None try: db = SessionLocal() yield db finally: db.close()
公式のSQL使う時解説だと, 上記の「def session()」に当たる箇所をapp.py上で定義してましたが, 依存する箇所が気持ち悪かったのでそこは流石にDatabaseに寄せました.
実はクリーンアーキテクチャなのだ
という細かい構成・こだわりを元にプロトタイプをシュッとつくりました.
実はこの設計, クリーンアーキテクチャを志向してまして.
Clean Architecture 達人に学ぶソフトウェアの構造と設計
- 作者:Robert C.Martin
- 出版社/メーカー: KADOKAWA
- 発売日: 2018/07/27
- メディア: 単行本
- 依存性を一貫して, 「（DB・インターフェース層） -> （ビジネスモデル層）」の一方通行にする.
- DBや（今回登場しなかったが）外部APIなどはservice層にまとめ, 他のビジネスモデル層と明確に切り離し.
- ビジネスモデル層（今回はlahmanとかsabr_analyticsパッケージ）は各々が独立し, 必要なモノをserviceからもらう.
- routerやapp.pyはDB何使おうが, ビジネスモデルをどう組もうが気にさせない.
を徹底しました.
が, もっといい設計あるよって人はぜひコメントとかもらえると嬉しいです.
これを何に使うか
デブサミ2020の発表時のデモ...のプロトタイプとして使います.*10
自分の発表は「残席わずか」らしいので, みんな来てね.
良いお年を🎍&これから大晦日ハッカソン2019 #大晦日ハッカソンに復帰してフロントエンド作ります.
*1:とはいえDocker化済ませているのでやろうと思えば半日かな.
*2:実はここに一番時間掛かってるかもしれない.
*3:これは後学の為にちゃんと作りたい. FastAPI使いこなしたい.
*4:これは方針として正しいと思っています, 公開する前に作れるようにキレイに書いたので（真顔）.
*5:来年の二刀流復活はマジで期待したい.
*6:Djangoは飽きるほど触ってる, Flaskもさんざん触ってるのと, 現在の開発状況とか考えるとリスクが多そうなので新規プロダクトで使う気になりませんでした.
*7:有志メンバーで昨日もくもく会をしてたときにこの話題で盛り上がりました.その後帰宅してから開発スタート.
*8:といいつつ, これはフロントエンド開発時の構成で実際はPyCharmでガッツリやってます.
*9:好き嫌いは分かれると思いますが笑
*10:そのままバックエンドのAPIとして使えるとは思ってますが, 本命を別に作りたい&あくまでデモの仕様を決めるためのプロトタイプなのでアッサリと捨てる可能性すらありますFastAPIのコードは.
