FastAPI と Tortoise-ORM の必殺コンボ

こんにちは、こんばんは、おはようございます。
K.S.ロジャース の島袋です。

今回は前回紹介したFastAPIを使用したテンプレートを作成してみました。さらに私イチオシのPython ORMであるTortoise-ORMを組み込んでシンプルなAPIを作ってみました。

サンプルコード

今回は事前に色々と弄りまわしたサンプルコードを用意しました。

こちらを解説する形で紹介したいと思います。

Model

まずはこちらModelを紹介。
流石、DjangoのModelをインスパイアしただけあって Tortoise-ORM使いやすい。何よりModel書けばそのままマイグレーションまでできるのが良いですね。
Pydanicにも対応していて使いまわしはホント最高です。

import uuid

from tortoise import fields
from tortoise.contrib.pydantic import pydantic_model_creator

from .base import BaseModel
from app.services import auth


class User(BaseModel):
    email = fields.CharField(max_length=100, unique=True)
    hashed_password = fields.CharField(max_length=200, null=True)
    refresh_token = fields.UUIDField(null=True)
    username = fields.CharField(max_length=100, required=True)

    @classmethod
    async def get_active_user(cls, user_id: int = None, email: str = None):
        if user_id is not None:
            return await cls.get_or_none(id=user_id, deleted_at=None)
        if email is not None:
            return await cls.get_or_none(email=email, deleted_at=None)
        return None

    @classmethod
    def create(cls, **kwargs):
        kwargs["hashed_password"] = auth.get_password_hash(kwargs["password"])
        kwargs["refresh_token"] = uuid.uuid4().hex
        return super().create(**kwargs)

    def get_access_token(self):
        return auth.create_access_token(data={
            "sub": self.email,
            "username": self.username
        })


User_Pydantic = pydantic_model_creator(User, name="User", include=(
    "email",
    "username",
))

UserIn_Pydantic = pydantic_model_creator(User, name="UserIn", include=(
    "username",
), exclude_readonly=True)

ただ、今回はService層を設けてビジネスロジックを疎結合でそっちに寄せたいけど、何だかんだとファットになりそうだなって思いました。そこら辺もDjangoインスパイア。

Controller

次はこちらController。
ユーザーアカウントのCRUDですが、かなりスッキリしていると思います。ValidationもPydanicが行ってくれるので、Controllerの記述が少なくてすみます。もちろん、カスタマイズが入れば記述は増えますが、FastAPI + Tortoise-ORMの性質上、Controllerの肥大化を押さえることができそうです。

from fastapi import APIRouter, Depends
from pydantic import BaseModel
from starlette.status import HTTP_204_NO_CONTENT

from app.services.auth import get_current_user
from app.models.user import User, User_Pydantic, UserIn_Pydantic


tags = ["users"]
router = APIRouter()


class UserIn(BaseModel):
    username: str
    email: str
    password: str


@router.post("/register", tags=tags, response_model=User_Pydantic)
async def register_user(form_data: UserIn):
    return await User.create(**form_data.dict(exclude_unset=True))


@router.get("/me", tags=tags, response_model=User_Pydantic)
async def get_user_data(user: User = Depends(get_current_user)):
    return user


@router.patch("/me", tags=tags, response_model=User_Pydantic)
async def update_user(form_data: UserIn_Pydantic, user: User = Depends(get_current_user)):
    await user.update_from_dict(form_data.dict(exclude_unset=True)).save()
    return user


@router.delete("/me", tags=tags, status_code=HTTP_204_NO_CONTENT)
async def delete_user(user: User = Depends(get_current_user)):
    await user.soft_delete()

こうして見るとDjangoのDRYはModelの機能が厚く、ノンコーティングで済むようになっているからなのかと思いました。Tortoise-ORMはいい仕事してますね。個人的にかなり応援しているプロダクトです。

Service

あと紹介するのがService層でしょうか？ こちらは認証周りの機能を集めた auth.py になります。
個人的に良いなと思ったのが、この位置でも HttpException をraiseすることでエラーレスポンスを返すことができることです。賛否両論あると思いますが、逐一Controllerでキャッチしなくても良いのは書きやすいですね。

from datetime import datetime, timedelta

import jwt
from fastapi import Depends
from fastapi.security import OAuth2PasswordBearer
from passlib.context import CryptContext

from app.configs.auth import AUTH_SECRET_KEY, ALGORITHM, ACCESS_TOKEN_EXPIRE_MINUTES
from app.models.user import User
from app.services.exceptions import HTTP_401_UNAUTHORIZED


pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="/token")


def verify_password(plain_password: str, hashed_password: str) -> bool:
    return pwd_context.verify(plain_password, hashed_password)


def get_password_hash(password: str) -> str:
    return pwd_context.hash(password)


def create_access_token(*, data: dict) -> str:
    expire = datetime.utcnow() + timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
    to_encode = data.copy()
    to_encode.update({"exp": expire})
    encoded_jwt = jwt.encode(to_encode, AUTH_SECRET_KEY, algorithm=ALGORITHM)
    return encoded_jwt.decode()


async def login_with_password(email: str, password: str) -> User:
    user = await User.get_active_user(email=email)
    if not user or not verify_password(password, user.hashed_password):
        raise HTTP_401_UNAUTHORIZED
    return user


async def get_current_user(token: str = Depends(oauth2_scheme)) -> User:
    try:
        payload = jwt.decode(token, AUTH_SECRET_KEY, algorithms=[ALGORITHM])
        user = await User.get_active_user(email=payload.get("sub"))
    except jwt.PyJWTError:
        raise HTTP_401_UNAUTHORIZED
    if not user:
        raise HTTP_401_UNAUTHORIZED
    return user

Service層はビジネスロジックの集約という意味であれば、もう少しServiceの階層を細分化しても良いかと思います。そこはプロダクト毎の規定になりますね。

main.py

最後はこちら起動ファイルを紹介。注目すべきは from tortoise.contrib.fastapi import register_tortoise これです。なんとTortoise-ORMは公式にFastAPIと連携がとれるのです。
実はこれがあるから今回この組み合わせで作ってみたのです。

from fastapi import FastAPI
from tortoise.contrib.fastapi import register_tortoise

from app.configs.db import DB_CONFIG
from app.routers import routers


app = FastAPI()

# routersを登録
for router in routers:
    app.include_router(router)

# DB
register_tortoise(
    app,
    config=DB_CONFIG,
    generate_schemas=True,
    add_exception_handlers=True,
)

まとめ

あっちこっちのドキュメントとサンプルコードを見ながら作ったのですがいかがでしたでしょうか？
ざっくりと作ってみましたが、FastAPIとTortoise-ORMの組み合わせかなり相性が良さそうです。
ただ注意点が一つあって、現段階（2020年5月時点）でどちらのプロダクトもv1.0以下で絶賛開発中。なので破壊的な変更が入る余地は多いにあるので、本番プロダクトに使うには留意が必要です。

あとづけ

ちなみに弊社、Tech系以外にも会社ブログも掲載してますので、気になった方は是非どうぞ。
https://www.wantedly.com/companies/ks-rogers