FastAPI を使ってWEBアプリを作ってみる　その6

Python,TIPSDocker,FastAPI,PostgreSQL

前回の投稿ではhedgehogsリソースを操作する基本的なAPIエンドポイントを構築しました。
今回の投稿ではユーザーが予約サイトにサインアップしてお気に入りのハリネズミさんを探し管理できるようにユーザーモデルを構築していきます。

過去の投稿はこちらから辿ることができます。

FastAPI を使ってWEBアプリを作ってみる その1FastAPIとDockerでHelloWorld
FastAPI を使ってWEBアプリを作ってみる その2AlembicとPostgreSQLでDB Migrate
FastAPI を使ってWEBアプリを作ってみる その3APIエンドポイントをPostgreSQLに接続
FastAPI を使ってWEBアプリを作ってみる その4PytestとDockerでテスト構築
FastAPI を使ってWEBアプリを作ってみる その5RESTAPIエンドポイントを実装
FastAPI を使ってWEBアプリを作ってみる その6今ここ

データベースのマイグレーション

データベース設計には多くパターンがあり、全てのユースケースにおいても「正しい」といえるアプローチはありません。

すべてのユーザー関連情報を1つのテーブルに格納することが効果的な場合もありますし、ユーザーを認証用のテーブルとプロファイルテーブルに切り出して分割するケースもあります。
後者は前者に比べてある程度テストが容易になり、認証テーブルの構造を気にすることなくユーザーモデルを拡張できるメリットがあると思います。

今回はこの分割したテーブル構成でモデリングしていきます。
まずはalembicのマイグレートファイルを更新していきましょう。

改修を加える前に以前の投稿で反映したマイグレートをロールバックします:

$ alembic downgrade base

次にマイグレートファイルをリファクタリングします。
※ ファイル名は環境ごとに異なります

"""create_first_tables

Revision ID: 12056735bd4e
Revises:
Create Date: 2021-01-22 18:27:50.234519

"""

from typing import Tuple

import sqlalchemy as sa
from alembic import op

# revision identifiers, used by Alembic
revision = '12056735bd4e'
down_revision = None
branch_labels = None
depends_on = None


def create_updated_at_trigger() -> None:
    op.execute(
        """
        CREATE OR REPLACE FUNCTION update_updated_at_column()
            RETURNS TRIGGER AS
        $$
        BEGIN
            NEW.updated_at = now();
            RETURN NEW;
        END;
        $$ language 'plpgsql';
        """
    )

def timestamps(indexed: bool = False) -> Tuple[sa.Column, sa.Column]:
    return (
        sa.Column(
            "created_at",
            sa.TIMESTAMP(timezone=True),
            server_default=sa.func.now(),
            nullable=False,
            index=indexed,
        ),
        sa.Column(
            "updated_at",
            sa.TIMESTAMP(timezone=True),
            server_default=sa.func.now(),
            nullable=False,
            index=indexed,
        ),
    )


def create_hedgehogs_table() -> None:
    op.create_table(
        "hedgehogs",
        sa.Column("id", sa.Integer, primary_key=True),
        sa.Column("name", sa.Text, nullable=False, index=True),
        sa.Column("description", sa.Text, nullable=True),
        sa.Column("color_type", sa.Text, nullable=False),
        sa.Column("age", sa.Numeric(10, 2), nullable=False),
        *timestamps(),
    )
    op.execute(
        """
        CREATE TRIGGER update_hedgehogs_modtime
            BEFORE UPDATE
            ON hedgehogs
            FOR EACH ROW
        EXECUTE PROCEDURE update_updated_at_column();
        """
    )


def create_users_table() -> None:
    op.create_table(
        "users",
        sa.Column("id", sa.Integer, primary_key=True),
        sa.Column("username", sa.Text, unique=True, nullable=False, index=True),
        sa.Column("email", sa.Text, unique=True, nullable=False, index=True),
        sa.Column("email_verified", sa.Boolean, nullable=False, server_default="False"),
        sa.Column("salt", sa.Text, nullable=False),
        sa.Column("password", sa.Text, nullable=False),
        sa.Column("is_active", sa.Boolean(), nullable=False, server_default="True"),
        sa.Column("is_superuser", sa.Boolean(), nullable=False, server_default="False"),
        *timestamps(),
    )
    op.execute(
        """
        CREATE TRIGGER update_user_modtime
            BEFORE UPDATE
            ON users
            FOR EACH ROW
        EXECUTE PROCEDURE update_updated_at_column();
        """
    )


def upgrade() -> None:
    create_updated_at_trigger()
    create_hedgehogs_table()
    create_users_table()


def downgrade() -> None:
    op.drop_table("users")
    op.drop_table("hedgehogs")
    op.execute("DROP FUNCTION update_updated_at_column")

ユーザテーブルの他に、データベース内のすべてのテーブルにタイムスタンプを簡単に追加できるようにするコードを追加しています。
timestamps() 関数は created_at と updated_at の2つのカラムを作成し、*timestamps() を使用することでhedgehogs と users テーブルに適用します。
どちらのカラムもデフォルトでは sqlalchemy.func.now() を使用して現在時刻を取得します。

また、create_updated_at_trigger 関数でテーブルごとに作成する PL/pgSQL トリガーを書いています。
このトリガーは与えられたテーブルのレコードが更新されるたびに実行され、updated_at カラムをカレントタイムで設定します。
PL/pgSQL について経験がない場合はこちらを参照してください。

usersテーブルにはすべての認証に関連する情報が格納され、Eメールとユーザー名のカラムにユニーク制約とインデックスも設定しています。

ここで一度マイグレートを実行しPydanticモデルの作成に進みましょう。

$ alembic upgrade head

モデルの作成と更新

まずはアプリケーション全体でタイムスタンプを管理するための新しい共通モデルを定義します。
core.pyファイルを以下のように更新してください:

from typing import Optional
from datetime import datetime, timedelta, timezone
from pydantic import BaseModel, validator


JST = timezone(timedelta(hours=+9), 'JST')

class CoreModel(BaseModel):
    pass


class DateTimeModelMixin(BaseModel):
    created_at: Optional[datetime]
    updated_at: Optional[datetime]

    @validator("created_at", "updated_at", pre=True)
    def default_datetime(cls, value: datetime) -> datetime:
        return value or datetime.datetime.now(JST)


class IDModelMixin(BaseModel):
    id: int

DateTimeModelMixin は pydantic のバリデータデコレーターを利用して、created_at と updated_at の両方のフィールドにデフォルトのタイムスタンプを設定します。
バリデータのオーバーライドについては 公式ドキュメントを参照して下さい。

Pydanticの追加モジュールを追加する必要があるので requirements.txt を追記しインストールします:

fastapi==0.63.0
uvicorn==0.13.3
pydantic==1.7.3
pydantic[email]  # 追加

databases[postgresql]==0.4.1
SQLAlchemy==1.3.23
alembic==1.5.4
psycopg2==2.8.6

pytest==6.2.2
pytest-asyncio==0.14.0
httpx==0.16.1
asgi-lifespan==1.0.1
docker==4.4.3

続いてusersテーブルに対応するモデルを定義しましょう:

$ touch /backend/app/models/user.py
import string
from typing import Optional

from app.models.core import CoreModel, DateTimeModelMixin, IDModelMixin
from pydantic import EmailStr, constr


def validate_username(username: str) -> str:
    allowed = string.ascii_letters + string.digits + "-" + "_"
    assert all(char in allowed for char in username), "ユーザー名に無効な文字が含まれています。"
    assert len(username) >= 3, "ユーザー名は3文字以上で入力してください。"
    return username


class UserBase(CoreModel):
    email: Optional[EmailStr]
    username: Optional[str]
    email_verified: bool = False
    is_active: bool = True
    is_superuser: bool = False


class UserCreate(CoreModel):
    email: EmailStr
    password: constr(min_length=7, max_length=100)
    username: constr(min_length=3, regex="[a-zA-Z0-9_-]+$")  # noqa


class UserUpdate(CoreModel):
    email: Optional[EmailStr]
    username: Optional[constr(min_length=3, regex="[a-zA-Z0-9_-]+$")]  # noqa


class UserPasswordUpdate(CoreModel):
    password: constr(min_length=7, max_length=100)
    salt: str


class UserInDB(IDModelMixin, DateTimeModelMixin, UserBase):
    password: constr(min_length=7, max_length=100)
    salt: str


class UserPublic(IDModelMixin, DateTimeModelMixin, UserBase):
    pass

ほとんどのAttributeはマイグレートファイルで定義したカラムをそのまま持ってきただけですが機密情報は除外しています。
ユーザー作成時にはメールアドレス、ユーザー名、パスワードを要求し、既存ユーザーはいつでも自身のメールアドレスやユーザー名を取得することができます。
また、パスワードをリセットする機能も提供しましょう

いずれも重要なことはパスワードとソルトを UserBase と UserPublic モデルから除外し、これらの情報がバックエンドに残らないようにしていることです。

ユーザーの email_verified のデフォルトは、メールが有効であることが確認できるまで False になっており、is_active と is_superuser のデフォルトはそれぞれ True と False になっています。

またpydanticでの追加検証もしています、constr型はpydanticの制約付きタイプの一つで constrained string（制約付き文字列）の略で、文字列に最小桁数と最大桁数を設定する機能を提供します。
今回はパスワードの7文字以上100文字以下にするよう定義しています。
regex="[a-zA-Z0-9_-]+$" このように regex を指定することで任意の正規表現にマッチする値のみ許可するように制限をかけています。

ユーザー登録のテスト

最初に一つ注意点があります。
絶対にデータベースへパスワードを平文で登録しないでください。

これは最も最悪なアプローチの一つです。
次回の投稿では passlib と bcrypt を使用しパスワードの暗号化を行う予定であるため本投稿では一時的に平文で登録をおこないます。
実務ではどんな状況でも避けるようにしてください。

それではいつもようにまずはテストファイルを作成します:

touch /backend/tests/test_users.py
import pytest
from fastapi import FastAPI
from httpx import AsyncClient
from starlette.status import HTTP_404_NOT_FOUND

pytestmark = pytest.mark.asyncio


class TestUserRoutes:
    async def test_routes_exist(self, app: FastAPI, client: AsyncClient) -> None:
        new_user = {
            "email": "test@email.com",
            "username": "test_username",
            "password": "testpassword"}

        res = await client.post(
            app.url_path_for("users:register-new-user"),
            json={"new_user": new_user})

        assert res.status_code != HTTP_404_NOT_FOUND

この状態でテストを実行するとお馴染みの starlette.routing.NoMatchFound が発生するのでルートを作成しパスさせます:

$ touch /backend/app/api/routes/users.py
from app.api.dependencies.database import get_repository
from app.db.repositories.users import UsersRepository
from app.models.user import UserCreate, UserPublic
from fastapi import APIRouter, Body, Depends
from starlette.status import HTTP_201_CREATED


router = APIRouter()
@router.post("/",
             response_model=UserPublic,
             name="users:register-new-user",
             status_code=HTTP_201_CREATED)
async def register_new_user(
    new_user: UserCreate = Body(..., embed=True),
    user_repo: UsersRepository = Depends(get_repository(UsersRepository)),
) -> UserPublic:
    created_user = await user_repo.register_new_user(new_user=new_user)
    return created_user

シンプルな実装です、新しいユーザーを UsersRepository に送信し、作成されたユーザーを返します。
続いてこのルートをルーターに登録します:

from fastapi import APIRouter
from app.api.routes.hedgehogs import router as hedgehogs_router
from app.api.routes.users import router as users_router


router = APIRouter()
router.include_router(hedgehogs_router, prefix="/hedgehogs", tags=["hedgehogs"])
router.include_router(users_router, prefix="/users", tags=["users"])

uvicornサーバを起動している場合、ターミナルにImportErrorが表示されるはずです。
UserRepository を用意しましょう:

$ touch /backend/app/db/repositories/users.py
from app.db.repositories.base import BaseRepository
from app.models.user import UserCreate, UserInDB


class UsersRepository(BaseRepository):
    async def register_new_user(self, *, new_user: UserCreate) -> UserInDB:
        return None

ここでもう一度テストを実行するとパスしますが、このコードは実際には役に立ちません。
UsersRepository に実際のロジックを肉付けしていきます。

まずはテストを書くことから始めましょう:

import pytest
from app.db.repositories.users import UsersRepository
from app.models.user import UserInDB
from databases import Database
from fastapi import FastAPI
from httpx import AsyncClient
from starlette.status import HTTP_201_CREATED, HTTP_404_NOT_FOUND

pytestmark = pytest.mark.asyncio


class TestUserRoutes:
    ...省略


class TestUserRegistration:
    async def test_users_can_register_successfully(
        self, app: FastAPI, client: AsyncClient, db: Database,
    ) -> None:
        user_repo = UsersRepository(db)
        new_user = {
            "email": "foo@bar.com",
            "username": "foobar",
            "password": "bazquxquux"}

        user_in_db = await user_repo.get_user_by_email(email=new_user["email"])
        assert user_in_db is None

        res = await client.post(
            app.url_path_for("users:register-new-user"),
            json={"new_user": new_user})
        assert res.status_code == HTTP_201_CREATED

        user_in_db = await user_repo.get_user_by_email(email=new_user["email"])
        assert user_in_db is not None
        assert user_in_db.email == new_user["email"]
        assert user_in_db.username == new_user["username"]

        created_user = UserInDB(
            **res.json(),
            password="whatever",
            salt="123").dict(
            exclude={
                "password",
                "salt"})
        assert created_user == user_in_db.dict(exclude={"password", "salt"})

    @pytest.mark.parametrize(
        "attr, value, status_code",
        (
            ("email", "foo@bar.com", 400),
            ("username", "foobar", 400),
            ("email", "invalid_email@one@two.com", 422),
            ("password", "short", 422),
            ("username", "foobar@#$%^<>", 422),
            ("username", "ab", 422),
        ),
    )
    async def test_user_registration_fails_when_credentials_are_taken(
        self,
        app: FastAPI,
        client: AsyncClient,
        attr: str,
        value: str,
        status_code: int,
    ) -> None:
        new_user = {
            "email": "nottaken@email.com",
            "username": "not_taken_username",
            "password": "foobarpassword"}
        new_user[attr] = value

        res = await client.post(
            app.url_path_for("users:register-new-user"),
            json={"new_user": new_user})
        assert res.status_code == status_code

2つテストを追加しました。

1つ目は、ユーザーが有効な資格情報を送信したケースです。
適切な資格情報を入力してPOSTリクエストを送信しそのユーザーがデータベースで作成されたかどうか、エンドポイントが期待通りのユーザーを返しているかどうかをチェックします。
ただパスワードと salt はユーザーを返す際に渡すべきではないため除外します。

2つ目は、電子メールやパスワードが既に使用されている場合や、いずれかの認証情報によって pydantic がバリデートエラーをスローしたケースです。
ユーザー名や電子メールが既に使用されている場合は 400コードをエンドポイントが返し、バリデートエラーが発生した場合は 422 コードを返すことを想定しています。

このままテストを実行すると UsersRepository オブジェクトには get_user_by_email が存在しないというAttributeError が表示されます。

これを修正しましょう:

from app.db.repositories.base import BaseRepository
from app.models.user import UserCreate, UserInDB
from fastapi import HTTPException, status
from pydantic import EmailStr


GET_USER_BY_EMAIL_QUERY = """
    SELECT
        id, username, email, email_verified, password,
        salt, is_active, is_superuser, created_at, updated_at
    FROM
        users
    WHERE
        email = :email;
"""
GET_USER_BY_USERNAME_QUERY = """
    SELECT
        id, username, email, email_verified, password,
        salt, is_active, is_superuser, created_at, updated_at
    FROM
        users
    WHERE
        username = :username;
"""
REGISTER_NEW_USER_QUERY = """
    INSERT INTO users
        (username, email, password, salt)
    VALUES
        (:username, :email, :password, :salt)
    RETURNING
        id, username, email, email_verified, password,
        salt, is_active, is_superuser, created_at, updated_at;
"""


class UsersRepository(BaseRepository):
    async def get_user_by_email(self, *, email: EmailStr) -> UserInDB:
        user_record = await self.db.fetch_one(
            query=GET_USER_BY_EMAIL_QUERY,
            values={"email": email})
        if not user_record:
            return None
        return UserInDB(**user_record)

    async def get_user_by_username(self, *, username: str) -> UserInDB:
        user_record = await self.db.fetch_one(
            query=GET_USER_BY_USERNAME_QUERY,
            values={"username": username})
        if not user_record:
            return None
        return UserInDB(**user_record)

    async def register_new_user(self, *, new_user: UserCreate) -> UserInDB:
        if await self.get_user_by_email(email=new_user.email):
            raise HTTPException(
                status_code=status.HTTP_400_BAD_REQUEST,
                detail="このメールアドレスはすでに登録されています")

        if await self.get_user_by_username(username=new_user.username):
            raise HTTPException(
                status_code=status.HTTP_400_BAD_REQUEST,
                detail="このユーザ名はすでに登録されています"
            )
        created_user = await self.db.fetch_one(
            query=REGISTER_NEW_USER_QUERY,
            values={**new_user.dict(), "salt": "123"})
        return UserInDB(**created_user)

まずは get_user_by_email と get_user_by_username メソッドについて説明します。
どちらも単一のメールアドレス or ユーザ名を受け取り、どちらかが一致するユーザが存在するかどうかデータベースに問い合わせます。
これらは後ほどログイン処理などユーザー取得する際にも使用しますが、register_new_user メソッドで既に登録が存在していないか確認したいため先行して実装しています。

このどちらかのメソッドがユーザーを返す場合は、適切なエラーメッセージで FastAPI の HTTPException を発生させます。
返さなかった場合は提供された資格情報とダミーのソルトを使用してユーザーを作成します。

※ 冒頭に記載した通り、次回の記事でパスワードを正しく暗号化して処理する際に本物のソルトを使用するようにリファクタリングします。

Remaind

冒頭に記載した通り、次回の記事でパスワードを正しく暗号化して処理する際に本物のソルトを使用するようにリファクタリングします。

もう一度テストを実行するとすべてがパスします、これでユーザー認証が準備ができました。

まとめ

少し早いですが区切りがいいので今回の投稿ではここまでにします。
ユーザ登録ができるようになったので次回の投稿ではパスワードの暗号化対応をし、JWTトークンを消してユーザがログインできるようにしてきます。

作成したコードはGitHubにアップしています。
part6ブランチが対応しています。

Python,TIPSDocker,FastAPI,PostgreSQL

Posted by Kenny