FastAPI を使ってWEBアプリを作ってみる その10
色々と忙しくなり4月になってしまいました。
前回の投稿ではユーザーのプロファイルを構築し必要に応じてAPIがユーザーのプロファイルを返すようにしました。
この投稿では店舗ユーザーが自身にハリネズミリソースを紐付けて管理できるようコードをリファクタリングをしていきます。
過去の投稿はこちらから辿ることができます。
| FastAPI を使ってWEBアプリを作ってみる その1 | FastAPIとDockerでHelloWorld |
| FastAPI を使ってWEBアプリを作ってみる その2 | AlembicとPostgreSQLでDB Migrate |
| FastAPI を使ってWEBアプリを作ってみる その3 | APIエンドポイントをPostgreSQLに接続 |
| FastAPI を使ってWEBアプリを作ってみる その4 | PytestとDockerでテスト構築 |
| FastAPI を使ってWEBアプリを作ってみる その5 | RESTAPIエンドポイントを実装 |
| FastAPI を使ってWEBアプリを作ってみる その6 | ユーザーモデルの作成 |
| FastAPI を使ってWEBアプリを作ってみる その7 | パスワードの暗号化とJWT認証 |
| FastAPI を使ってWEBアプリを作ってみる その8 | 認証機能のDependencies化 |
| FastAPI を使ってWEBアプリを作ってみる その9 | ユーザプロファイルの作成 |
| FastAPI を使ってWEBアプリを作ってみる その10 | 今ここ |
Contents
ハリネズミリソースの修正
現在のハリネズミさん関連のデータベース構造はかなり単純なものになっています。
どこの店舗にハリネズミさんが紐づいているのかを追跡する方法がないのでここから修正しています。
マイグレーション
データベースをロールバックします。
$ alembic downgrade base"""create_first_tables
Revision ID: 12056735bd4e
Revises:
Create Date: 2021-01-22 18:27:50.234519
"""
...省略
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),
# 追加
sa.Column("owner", sa.Integer, sa.ForeignKey("users.id", ondelete="CASCADE")),
*timestamps(),
)
op.execute(
"""
CREATE TRIGGER update_hedgehogs_modtime
BEFORE UPDATE
ON hedgehogs
FOR EACH ROW
EXECUTE PROCEDURE update_updated_at_column();
"""
)
...省略
def upgrade() -> None:
# 実行順を変更
create_updated_at_trigger()
create_users_table()
create_profiles_table()
create_hedgehogs_table()
def downgrade() -> None:
# 実行順を変更
op.drop_table("hedgehogs")
op.drop_table("profiles")
op.drop_table("users")
op.execute("DROP FUNCTION update_updated_at_column")
hedgehogsテーブルの作成をusersとprofilesテーブルの作成後に移しました。
理由はhedgehogsテーブルにownerカラムを定義する際にusersテーブルを参照したいからです。
この変更は他のリソースに少し影響がありますが後ほど触れます。
upgrade headコマンドを実行してマイグレートしましょう。
$ alembic upgrade headハリネズミリソースの所有権をユーザーに与えたのでコードをリファクタリングします。
どのユーザーもハリネズミリソースにアクセスできますが作成したユーザーだけが更新や削除をできるようにします。
またハリネズミを一覧表示したいユーザーは、自分で作成したものだけを受け取るようにします。
Hedgehogリソースのリファクタリング
いつもならばテストから始めますがデータベース構造を変更したので今回は models/hedgehog.py ファイルから修正していきましょう。
ハリネズミモデルの修正
from enum import Enum
from typing import Optional, Union
from app.models.core import CoreModel, DateTimeModelMixin, IDModelMixin # 更新
from app.models.user import UserPublic # 追加
...省略
class HedgehogInDB(IDModelMixin, DateTimeModelMixin, HedgehogBase):
name: str
age: float
color_type: ColorType
owner: int # 追加
# 更新
class HedgehogPublic(IDModelMixin, DateTimeModelMixin, HedgehogBase):
owner: Union[int, UserPublic]
所有しているユーザID を表すintが格納される owner アトリビュートを HedgehogInDB モデルに追加し、HedgehogInDB モデルで DateTimeModelMixin を利用しタイムスタンプを活用できるようにします。HedgehogPublicモデルは、単にHedgehogInDBを継承していますが、ownerアトリビュートがユーザーIDの整数かUserPublicモデル自体になるよう指定しています。
この変更によりテストを実行するとほとんどがエラーになります。
テストが通るように今度はテストコードを更新していきましょう。
conftest.py ファイルの test_hedgehog フィクスチャから修正します。
...省略
@pytest.fixture
async def test_hedgehog(db: Database, test_user: UserInDB) -> HedgehogInDB:
hedgehog_repo = HedgehogsRepository(db)
new_hedgehog = HedgehogCreate(
name="fake hedgehog name",
description="fake description",
age=2.2,
color_type="SOLT & PEPPER",
)
return await hedgehog_repo.create_hedgehog(
new_hedgehog=new_hedgehog, requesting_user=test_user
)
...省略test_userフィクスチャを引数に追加し新しいハリネズミが作成されるたびにHedgehogsRepositoryに送信しています。
このユーザーを取得してIDをownerアトリビュートとして渡します。
POST hedgehogの修正
HedgehogsRepositoryを以下のように変更します:
from typing import List
from app.db.repositories.base import BaseRepository
from app.models.hedgehog import HedgehogCreate, HedgehogInDB, HedgehogUpdate
from fastapi import HTTPException
from starlette.status import HTTP_400_BAD_REQUEST
import app.db.repositories.queries.hedgehogs as query
from app.models.user import UserInDB # 追加
class HedgehogsRepository(BaseRepository):
async def create_hedgehog(
self, *, new_hedgehog: HedgehogCreate, requesting_user: UserInDB
) -> HedgehogInDB:
hedgehog = await self.db.fetch_one(
query=query.CREATE_HEDGEHOG_QUERY,
values={**new_hedgehog.dict(), "owner": requesting_user.id},
)
return HedgehogInDB(**hedgehog)
...省略続いてSQLにもownerアトリビュートを追加します:
CREATE_HEDGEHOG_QUERY = """
INSERT INTO hedgehogs (name, description, age, color_type, owner)
VALUES (:name, :description, :age, :color_type, :owner)
RETURNING id, name, description, age, color_type, owner, created_at, updated_at;
"""
...省略今回はCREATE_HEDGEHOG_QUERYを更新しただけですがSQLのほとんどを変更する必要があります。
現在テストを実行してもログインユーザーを HedgehogRepository に渡していないためエラーになります。
この対応よりも先に tests/test_hedgehogs.pyファイルを更新しますが既存のテストコードを殆ど捨てて新しく書き直すのと同義になります。
from typing import Dict, List, Union
import pytest
from app.db.repositories.hedgehogs import HedgehogsRepository
from app.models.hedgehog import HedgehogCreate, HedgehogInDB, HedgehogPublic
from app.models.user import UserInDB
from databases import Database
from fastapi import FastAPI, status
from httpx import AsyncClient
pytestmark = pytest.mark.asyncio
@pytest.fixture
def new_hedgehog():
return HedgehogCreate(
name="test hedgehog",
description="test description",
age=0.0,
color_type="SOLT & PEPPER",
)
@pytest.fixture
async def test_hedgehogs_list(db: Database, test_user2: UserInDB) -> List[HedgehogInDB]:
hedgehog_repo = HedgehogsRepository(db)
return [
await hedgehog_repo.create_hedgehog(
new_hedgehog=HedgehogCreate(
name="test hedgehog {i}",
description="test description",
age=0.0,
color_type="SOLT & PEPPER",
),
requesting_user=test_user2,
)
for i in range(5)
]
class TestHedgehogsRoutes:
async def test_routes_exist(self, app: FastAPI, client: AsyncClient) -> None:
res = await client.post(app.url_path_for("hedgehogs:create-hedgehog"), json={})
assert res.status_code != status.HTTP_404_NOT_FOUND
class TestCreateHedgehog:
async def test_valid_input_creates_hedgehog_belonging_to_user(
self,
app: FastAPI,
authorized_client: AsyncClient,
test_user: UserInDB,
new_hedgehog: HedgehogCreate,
) -> None:
res = await authorized_client.post(
app.url_path_for("hedgehogs:create-hedgehog"),
json={"new_hedgehog": new_hedgehog.dict()},
)
assert res.status_code == status.HTTP_201_CREATED
created_hedgehog = HedgehogPublic(**res.json())
assert created_hedgehog.name == new_hedgehog.name
assert created_hedgehog.age == new_hedgehog.age
assert created_hedgehog.description == new_hedgehog.description
assert created_hedgehog.color_type == new_hedgehog.color_type
assert created_hedgehog.owner == test_user.id
async def test_unauthorized_user_unable_to_create_hedgehog(
self, app: FastAPI, client: AsyncClient, new_hedgehog: HedgehogCreate
) -> None:
res = await client.post(
app.url_path_for("hedgehogs:create-hedgehog"),
json={"new_hedgehog": new_hedgehog.dict()},
)
assert res.status_code == status.HTTP_401_UNAUTHORIZED
@pytest.mark.parametrize(
"invalid_payload, status_code",
(
(None, 422),
({}, 422),
({"name": "test_name"}, 422),
({"age": 2}, 422),
({"name": "test_name", "description": "test"}, 422),
),
)
async def test_invalid_input_raises_error(
self,
app: FastAPI,
authorized_client: AsyncClient,
invalid_payload: Dict[str, Union[str, float]],
test_hedgehog: HedgehogCreate,
status_code: int,
) -> None:
res = await authorized_client.post(
app.url_path_for("hedgehogs:create-hedgehog"),
json={"new_hedgehog": invalid_payload},
)
assert res.status_code == status_code
認証されていないユーザーがハリネズミを作成できないことを確認する新しいテストを追加し、新しく作成されたハリネズミのオーナーが現在ログインしているユーザーであることを確認しています。
続いてapi/routes/hedgehogs.pyを直しますがここでも殆ど初めから書き直していきます:
from typing import List
from app.api.dependencies.auth import get_current_active_user
from app.api.dependencies.database import get_repository
from app.db.repositories.hedgehogs import HedgehogsRepository
from app.models.hedgehog import HedgehogCreate, HedgehogPublic, HedgehogUpdate
from app.models.user import UserCreate, UserInDB, UserPublic, UserUpdate
from fastapi import APIRouter, Body, Depends, status
router = APIRouter()
@router.post(
"/",
response_model=HedgehogPublic,
name="hedgehogs:create-hedgehog",
status_code=status.HTTP_201_CREATED,
)
async def create_new_hedgehog(
new_hedgehog: HedgehogCreate = Body(..., embed=True),
current_user: UserInDB = Depends(get_current_active_user),
hedgehogs_repo: HedgehogsRepository = Depends(get_repository(HedgehogsRepository)),
) -> HedgehogPublic:
created_hedgehog = await hedgehogs_repo.create_hedgehog(
new_hedgehog=new_hedgehog, requesting_user=current_user
)
return created_hedgehog
以前の投稿で定義したdependencies.authを使用してログインしたユーザーをHedgehogsRepositoryに渡しています。
これにより新しいハリネズミには、現在認証されているユーザーがownerとしてアタッチされます。
テストを再度実行しTestCreatehedgehogのテストがすべてパスすることを確認してください。
GET hedgehogsの修正
POSTメソッドができたので次はGETメソッドをリファクタします。
例の如くテストコードから行います:
...省略
class TestGetHedgehog:
async def test_get_hedgehog_by_id(
self, app: FastAPI, authorized_client: AsyncClient, test_hedgehog: HedgehogInDB
) -> None:
res = await authorized_client.get(
app.url_path_for(
"hedgehogs:get-hedgehog-by-id", hedgehog_id=test_hedgehog.id
)
)
assert res.status_code == status.HTTP_200_OK
hedgehog = HedgehogInDB(**res.json())
assert hedgehog == test_hedgehog
async def test_unauthorized_users_cant_access_hedgehogs(
self, app: FastAPI, client: AsyncClient, test_hedgehog: HedgehogInDB
) -> None:
res = await client.get(
app.url_path_for(
"hedgehogs:get-hedgehog-by-id", hedgehog_id=test_hedgehog.id
)
)
assert res.status_code == status.HTTP_401_UNAUTHORIZED
@pytest.mark.parametrize(
"id, status_code",
((50000, 404), (-1, 422), (None, 422)),
)
async def test_wrong_id_returns_error(
self, app: FastAPI, authorized_client: AsyncClient, id: int, status_code: int
) -> None:
res = await authorized_client.get(
app.url_path_for("hedgehogs:get-hedgehog-by-id", hedgehog_id=id)
)
assert res.status_code == status_code
async def test_get_all_hedgehogs_returns_only_user_owned_hedgehogs(
self,
app: FastAPI,
authorized_client: AsyncClient,
test_user: UserInDB,
db: Database,
test_hedgehog: HedgehogInDB,
test_hedgehogs_list: List[HedgehogInDB],
) -> None:
res = await authorized_client.get(
app.url_path_for("hedgehogs:list-all-user-hedgehogs")
)
assert res.status_code == status.HTTP_200_OK
assert isinstance(res.json(), list)
assert len(res.json()) > 0
hedgehogs = [HedgehogInDB(**item) for item in res.json()]
assert test_hedgehog in hedgehogs
for hedgehog in hedgehogs:
assert hedgehog.owner == test_user.id
assert all(c not in hedgehogs for c in test_hedgehogs_list)基本的にはPOSTメソッドの改修と同じようなことをしています。
店舗ユーザーは認証されている場合のみハリネズミリソースを取得することができます。
また店舗ユーザーがすべてのハリネズミリストを要求した場合は自分に属するものだけを送り返します。
HedgehogsRepositoryを開き以下のように更新します:
...省略
class HedgehogsRepository(BaseRepository):
async def create_hedgehog(
self, *, new_hedgehog: HedgehogCreate, requesting_user: UserInDB
) -> HedgehogInDB:
hedgehog = await self.db.fetch_one(
query=query.CREATE_HEDGEHOG_QUERY,
values={**new_hedgehog.dict(), "owner": requesting_user.id},
)
return HedgehogInDB(**hedgehog)
# ----------- 以下更新 -----------
async def get_hedgehog_by_id(
self, *, id: int, requesting_user: UserInDB
) -> HedgehogInDB:
hedgehog = await self.db.fetch_one(
query=query.GET_HEDGEHOG_BY_ID_QUERY, values={"id": id}
)
if not hedgehog:
return None
return HedgehogInDB(**hedgehog)
async def list_all_user_hedgehogs(
self, requesting_user: UserInDB
) -> List[HedgehogInDB]:
hedgehog_records = await self.db.fetch_all(
query=query.LIST_ALL_USER_HEDGEHOGS_QUERY,
values={"owner": requesting_user.id},
)
return [HedgehogInDB(**item) for item in hedgehog_records]各メソッドでrequesting_userを要求しています。
get_hedgehog_by_idメソッドではこのパラメータを使用していませんが一貫性を保つためと後ほど使うことになるのでそのままにしておきます。
SQLクエリも更新します:
...省略
GET_HEDGEHOG_BY_ID_QUERY = """
SELECT id, name, description, age, color_type, owner, created_at, updated_at
FROM hedgehogs
WHERE id = :id;
"""
LIST_ALL_USER_HEDGEHOGS_QUERY = """
SELECT id, name, description, age, color_type, owner, created_at, updated_at
FROM hedgehogs
WHERE owner = :owner;
"""この変更をサポートするためにルートも変更しましょう:
...省略
@router.get(
"/{hedgehog_id}/",
response_model=HedgehogPublic,
name="hedgehogs:get-hedgehog-by-id",
)
async def get_hedgehog_by_id(
hedgehog_id: int = Path(..., ge=1),
current_user: UserInDB = Depends(get_current_active_user),
hedgehogs_repo: HedgehogsRepository = Depends(get_repository(HedgehogsRepository)),
) -> HedgehogPublic:
hedgehog = await hedgehogs_repo.get_hedgehog_by_id(
id=hedgehog_id, requesting_user=current_user
)
if not hedgehog:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail="No hedgehog found with that id.",
)
return hedgehog
@router.get(
"/", response_model=List[HedgehogPublic], name="hedgehogs:list-all-user-hedgehogs"
)
async def list_all_user_hedgehogs(
current_user: UserInDB = Depends(get_current_active_user),
hedgehogs_repo: HedgehogsRepository = Depends(get_repository(HedgehogsRepository)),
) -> List[HedgehogPublic]:
return await hedgehogs_repo.list_all_user_hedgehogs(requesting_user=current_user)
ルートを保護するためにget_current_active_userをdependencyで使用しユーザーをHedgehogsRepositoryに渡しています。
これでGETメソッドの改修が完了しました、テストを実行し全てパスすることを確認してください。
PUT hedgehogの修正
ハリネズミを更新するためのテストはほんの少しの修正で済みます:
...省略
class TestUpdateHedgehog:
@pytest.mark.parametrize(
"attrs_to_change, values",
(
(["name"], ["new fake hedgehog name"]),
(["description"], ["new fake hedgehog description"]),
(["age"], [3.14]),
(["color_type"], ["CHOCOLATE"]),
(
["name", "description"],
["extra new fake hedgehog name", "extra new fake hedgehog description"],
),
(["age", "color_type"], [2.00, "CHOCOLATE"]),
),
)
async def test_update_hedgehog_with_valid_input(
self,
app: FastAPI,
authorized_client: AsyncClient,
test_hedgehog: HedgehogInDB,
attrs_to_change: List[str],
values: List[str],
) -> None:
hedgehog_update = {
"hedgehog_update": {
attrs_to_change[i]: values[i] for i in range(len(attrs_to_change))
}
}
res = await authorized_client.put(
app.url_path_for(
"hedgehogs:update-hedgehog-by-id", hedgehog_id=test_hedgehog.id
),
json=hedgehog_update,
)
assert res.status_code == status.HTTP_200_OK
updated_hedgehog = HedgehogInDB(**res.json())
assert updated_hedgehog.id == test_hedgehog.id
for i in range(len(attrs_to_change)):
assert getattr(updated_hedgehog, attrs_to_change[i]) != getattr(
test_hedgehog, attrs_to_change[i]
)
assert getattr(updated_hedgehog, attrs_to_change[i]) == values[i]
for attr, value in updated_hedgehog.dict().items():
if attr not in attrs_to_change and attr != "updated_at":
assert getattr(test_hedgehog, attr) == value
async def test_user_recieves_error_if_updating_other_users_hedgehog(
self,
app: FastAPI,
authorized_client: AsyncClient,
test_hedgehogs_list: List[HedgehogInDB],
) -> None:
res = await authorized_client.put(
app.url_path_for(
"hedgehogs:update-hedgehog-by-id", hedgehog_id=test_hedgehogs_list[0].id
),
json={"hedgehog_update": {"age": 3.2}},
)
assert res.status_code == status.HTTP_403_FORBIDDEN
async def test_user_cant_change_ownership_of_hedgehog(
self,
app: FastAPI,
authorized_client: AsyncClient,
test_hedgehog: HedgehogInDB,
test_user: UserInDB,
test_user2: UserInDB,
) -> None:
res = await authorized_client.put(
app.url_path_for(
"hedgehogs:update-hedgehog-by-id", hedgehog_id=test_hedgehog.id
),
json={"hedgehog_update": {"owner": test_user2.id}},
)
assert res.status_code == status.HTTP_200_OK
hedgehog = HedgehogPublic(**res.json())
assert hedgehog.owner == test_user.id
@pytest.mark.parametrize(
"id, payload, status_code",
(
(-1, {"name": "test"}, 422),
(0, {"name": "test2"}, 422),
(500, {"name": "test3"}, 404),
(1, None, 422),
(1, {"color_type": "invalid color type"}, 422),
(1, {"color_type": None}, 400),
),
)
async def test_update_hedgehog_with_invalid_input_throws_error(
self,
app: FastAPI,
authorized_client: AsyncClient,
id: int,
payload: Dict[str, Optional[str]],
status_code: int,
) -> None:
hedgehog_update = {"hedgehog_update": payload}
res = await authorized_client.put(
app.url_path_for("hedgehogs:update-hedgehog-by-id", hedgehog_id=id),
json=hedgehog_update,
)
assert res.status_code == status_code変更したのは認証されたユーザが他のユーザが所有するハリネズミリソースを更新する権限を持っていないことを確認することだけです。
またユーザが自分のハリネズミリソースの所有者を変更できないようにしています。
その他はすべて同じです。
リポジトリ側を更新しましょう:
...省略
class HedgehogsRepository(BaseRepository):
...省略
async def update_hedgehog(
self, *, id: int, hedgehog_update: HedgehogUpdate, requesting_user: UserInDB
) -> HedgehogInDB:
hedgehog = await self.get_hedgehog_by_id(id=id, requesting_user=requesting_user)
if not hedgehog:
return None
if hedgehog.owner != requesting_user.id:
raise HTTPException(
status_code=status.HTTP_403_FORBIDDEN,
detail="Users are only able to update hedgehogs that they created.",
)
hedgehog_update_params = hedgehog.copy(
update=hedgehog_update.dict(exclude_unset=True)
)
if hedgehog_update_params.color_type is None:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail="Invalid color type. Cannot be None.",
)
updated_hedgehog = await self.db.fetch_one(
query=query.UPDATE_HEDGEHOG_BY_ID_QUERY,
values={
**hedgehog_update_params.dict(exclude={"created_at", "updated_at"}),
"owner": requesting_user.id,
},
)
return HedgehogInDB(**updated_hedgehog)update_hedgehogメソッドのサイズが大きくなっていますが殆どはエッジケースを扱っています。
もし頻繁に肥大化が起こるようであればBaseRepositoryをもう少し構築し多くの定型的な処理を継承するように構築するが好ましいです。
今回は完成間近なのでこのまま行こうと思います。
SQLクエリも更新します:
...省略
UPDATE_HEDGEHOG_BY_ID_QUERY = """
UPDATE hedgehogs
SET name = :name,
description = :description,
age = :age,
color_type = :color_type
WHERE id = :id AND owner = :owner
RETURNING id, name, description, age, color_type, owner, created_at, updated_at;
"""最後にルートを構成します:
...省略
@router.put(
"/{hedgehog_id}/",
response_model=HedgehogPublic,
name="hedgehogs:update-hedgehog-by-id",
)
async def update_hedgehog_by_id(
hedgehog_id: int = Path(..., ge=1),
current_user: UserInDB = Depends(get_current_active_user),
hedgehog_update: HedgehogUpdate = Body(..., embed=True),
hedgehogs_repo: HedgehogsRepository = Depends(get_repository(HedgehogsRepository)),
) -> HedgehogPublic:
updated_hedgehog = await hedgehogs_repo.update_hedgehog(
id=hedgehog_id, hedgehog_update=hedgehog_update, requesting_user=current_user
)
if not updated_hedgehog:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail="No hedgehog found with that id.",
)
return updated_hedgehog
ルートでは殆ど変更点はありません。
他のメソッド同様にget_current_active_userでルートを保護しそれをRepositoryに渡していることです。
ここでもテストを実行し適切に処理されることを確認してください。
DELETE hedgehogの修正
それではラストにDELETEメソッドをリファクタしていきましょう。
テストケースを更新します:
class TestDeleteHedgehog:
async def test_can_delete_hedgehog_successfully(
self, app: FastAPI, authorized_client: AsyncClient, test_hedgehog: HedgehogInDB
) -> None:
res = await authorized_client.delete(
app.url_path_for(
"hedgehogs:delete-hedgehog-by-id", hedgehog_id=test_hedgehog.id
)
)
assert res.status_code == status.HTTP_200_OK
async def test_user_cant_delete_other_users_hedgehog(
self,
app: FastAPI,
authorized_client: AsyncClient,
test_hedgehogs_list: List[HedgehogInDB],
) -> None:
res = await authorized_client.delete(
app.url_path_for(
"hedgehogs:delete-hedgehog-by-id", hedgehog_id=test_hedgehogs_list[0].id
)
)
assert res.status_code == status.HTTP_403_FORBIDDEN
@pytest.mark.parametrize(
"id, status_code",
((5000000, 404), (0, 422), (-1, 422), (None, 422)),
)
async def test_wrong_id_throws_error(
self,
app: FastAPI,
authorized_client: AsyncClient,
test_hedgehog: HedgehogInDB,
id: int,
status_code: int,
) -> None:
res = await authorized_client.delete(
app.url_path_for("hedgehogs:delete-hedgehog-by-id", hedgehog_id=id)
)
assert res.status_code == status_code店舗ユーザーが自分のハリネズミを削除でき、他の店舗ユーザーのハリネズミを削除できないことを確認しています。
Repositoryを更新します:
...省略
class HedgehogsRepository(BaseRepository):
...省略
async def delete_hedgehog_by_id(self, *, id: int, requesting_user: UserInDB) -> int:
hedgehog = await self.get_hedgehog_by_id(id=id, requesting_user=requesting_user)
if not hedgehog:
return None
if hedgehog.owner != requesting_user.id:
raise HTTPException(
status_code=status.HTTP_403_FORBIDDEN,
detail="Users are only able to delete hedgehogs that they created.",
)
deleted_id = await self.db.execute(
query=query.DELETE_HEDGEHOG_BY_ID_QUERY,
values={"id": id, "owner": requesting_user.id},
)
return deleted_id
delete_hedgehog_by_idメソッドでも他と同じことをしています。
requesting_user を渡して自分が所有するハリネズミの削除のみが許可されていることを確認します。
SQLクエリも他と同様にWHERE句にownerを指定します:
...省略
DELETE_HEDGEHOG_BY_ID_QUERY = """
DELETE FROM hedgehogs
WHERE id = :id AND owner = :owner
RETURNING id;
"""最後にルートを書き換えます:
...省略
@router.delete(
"/{hedgehog_id}/", response_model=int, name="hedgehogs:delete-hedgehog-by-id"
)
async def delete_hedgehog_by_id(
hedgehog_id: int = Path(..., ge=1),
current_user: UserInDB = Depends(get_current_active_user),
hedgehogs_repo: HedgehogsRepository = Depends(get_repository(HedgehogsRepository)),
) -> int:
deleted_id = await hedgehogs_repo.delete_hedgehog_by_id(
id=hedgehog_id, requesting_user=current_user
)
if not deleted_id:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail="No hedgehog found with that id.",
)
return deleted_id
お疲れ様でした！
テストを実行し全てがパスするのを確認してください。
しかし現在は例外返却用のコードがあちこちで重複しています。
ハリネズミリソースが既に存在するかチェックし続け存在しない場合は404例外を発生させています。
また、ユーザーがリソースの変更や削除を許可されていない場合は403例外を発生させています。
ここもリファクタリングして綺麗に纏めましょう、これはAuthでも利用したFastAPIの依存性注入機能を使って共通処理として抽象化することができます。
改修用の新しいdependencyを作りましょう:
$ touch /backend/app/api/dependencies/hedgehogs.pyfrom app.api.dependencies.auth import get_current_active_user
from app.api.dependencies.database import get_repository
from app.db.repositories.hedgehogs import HedgehogsRepository
from app.models.hedgehog import HedgehogInDB
from app.models.user import UserInDB
from fastapi import Depends, HTTPException, Path, status
async def get_hedgehog_by_id_from_path(
hedgehog_id: int = Path(..., ge=1),
current_user: UserInDB = Depends(get_current_active_user),
hedgehogs_repo: HedgehogsRepository = Depends(get_repository(HedgehogsRepository)),
) -> HedgehogInDB:
hedgehog = await hedgehogs_repo.get_hedgehog_by_id(
id=hedgehog_id, requesting_user=current_user
)
if not hedgehog:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail="No hedgehog found with that id.",
)
return hedgehog
def check_hedgehog_modification_permissions(
current_user: UserInDB = Depends(get_current_active_user),
hedgehog: HedgehogInDB = Depends(get_hedgehog_by_id_from_path),
) -> None:
if hedgehog.owner != current_user.id:
raise HTTPException(
status_code=status.HTTP_403_FORBIDDEN,
detail="Action forbidden. Users are only able to modify hedgehogs they own",
)
一般的な例外を抽象化しハリネズミリソースへのアクセスを管理できるようにしました。
これで、ルートとリポジトリを大幅に簡素化することができます。
HedgehogsRepositoryに組み込んで重複したコードを掃除しましょう:
from typing import List
import app.db.repositories.queries.hedgehogs as query
from app.db.repositories.base import BaseRepository
from app.models.hedgehog import HedgehogCreate, HedgehogInDB, HedgehogUpdate
from app.models.user import UserInDB
from fastapi import HTTPException, status
class HedgehogsRepository(BaseRepository):
async def create_hedgehog(
self, *, new_hedgehog: HedgehogCreate, requesting_user: UserInDB
) -> HedgehogInDB:
hedgehog = await self.db.fetch_one(
query=query.CREATE_HEDGEHOG_QUERY,
values={**new_hedgehog.dict(), "owner": requesting_user.id},
)
return HedgehogInDB(**hedgehog)
async def get_hedgehog_by_id(
self, *, id: int, requesting_user: UserInDB
) -> HedgehogInDB:
hedgehog = await self.db.fetch_one(
query=query.GET_HEDGEHOG_BY_ID_QUERY, values={"id": id}
)
if not hedgehog:
return None
return HedgehogInDB(**hedgehog)
async def list_all_user_hedgehogs(
self, requesting_user: UserInDB
) -> List[HedgehogInDB]:
hedgehog_records = await self.db.fetch_all(
query=query.LIST_ALL_USER_HEDGEHOGS_QUERY,
values={"owner": requesting_user.id},
)
return [HedgehogInDB(**item) for item in hedgehog_records]
async def update_hedgehog(
self, *, hedgehog: HedgehogInDB, hedgehog_update: HedgehogUpdate
) -> HedgehogInDB:
hedgehog_update_params = hedgehog.copy(
update=hedgehog_update.dict(exclude_unset=True)
)
if hedgehog_update_params.color_type is None:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail="Invalid hedgehog type. Cannot be None.",
)
updated_hedgehog = await self.db.fetch_one(
query=query.UPDATE_HEDGEHOG_BY_ID_QUERY,
values=hedgehog_update_params.dict(
exclude={"owner", "created_at", "updated_at"}
),
)
return HedgehogInDB(**updated_hedgehog)
async def delete_hedgehog_by_id(self, *, hedgehog: HedgehogInDB) -> int:
return await self.db.execute(
query=query.DELETE_HEDGEHOG_BY_ID_QUERY, values={"id": hedgehog.id}
)
update_hedgehogを中心にコードが大幅に少なくなりupdateとdelete両方のアクションでidの代わりにHedgehogInDB モデルを受け入れるようになりました。
依存関係は get_hedgehog_by_id メソッドを使用してすべての404エラーを処理しています。
また、check_hedgehog_modification_permissionsが処理してくれるので、updateメソッドのrequesting_user引数を削除しました。
これを扱うためにルートを変更します:
from typing import List
from app.api.dependencies.auth import get_current_active_user
from app.api.dependencies.database import get_repository
from app.api.dependencies.hedgehogs import (
check_hedgehog_modification_permissions,
get_hedgehog_by_id_from_path,
)
from app.db.repositories.hedgehogs import HedgehogsRepository
from app.models.hedgehog import (
HedgehogCreate,
HedgehogInDB,
HedgehogPublic,
HedgehogUpdate,
)
from app.models.user import UserInDB
from fastapi import APIRouter, Body, Depends, status
router = APIRouter()
@router.post(
"/",
response_model=HedgehogPublic,
name="hedgehogs:create-hedgehog",
status_code=status.HTTP_201_CREATED,
)
async def create_new_hedgehog(
new_hedgehog: HedgehogCreate = Body(..., embed=True),
current_user: UserInDB = Depends(get_current_active_user),
hedgehogs_repo: HedgehogsRepository = Depends(get_repository(HedgehogsRepository)),
) -> HedgehogPublic:
return await hedgehogs_repo.create_hedgehog(
new_hedgehog=new_hedgehog, requesting_user=current_user
)
@router.get(
"/", response_model=List[HedgehogPublic], name="hedgehogs:list-all-user-hedgehogs"
)
async def list_all_user_hedgehogs(
current_user: UserInDB = Depends(get_current_active_user),
hedgehogs_repo: HedgehogsRepository = Depends(get_repository(HedgehogsRepository)),
) -> List[HedgehogPublic]:
return await hedgehogs_repo.list_all_user_hedgehogs(requesting_user=current_user)
@router.get(
"/{hedgehog_id}/",
response_model=HedgehogPublic,
name="hedgehogs:get-hedgehog-by-id",
)
async def get_hedgehog_by_id(
hedgehog: HedgehogInDB = Depends(get_hedgehog_by_id_from_path),
) -> HedgehogPublic:
return hedgehog
@router.put(
"/{hedgehog_id}/",
response_model=HedgehogPublic,
name="hedgehogs:update-hedgehog-by-id",
dependencies=[Depends(check_hedgehog_modification_permissions)],
)
async def update_hedgehog_by_id(
hedgehog: HedgehogInDB = Depends(get_hedgehog_by_id_from_path),
hedgehog_update: HedgehogUpdate = Body(..., embed=True),
hedgehogs_repo: HedgehogsRepository = Depends(get_repository(HedgehogsRepository)),
) -> HedgehogPublic:
return await hedgehogs_repo.update_hedgehog(
hedgehog=hedgehog, hedgehog_update=hedgehog_update
)
@router.delete(
"/{hedgehog_id}/",
response_model=int,
name="hedgehogs:delete-hedgehog-by-id",
dependencies=[Depends(check_hedgehog_modification_permissions)],
)
async def delete_hedgehog_by_id(
hedgehog: HedgehogInDB = Depends(get_hedgehog_by_id_from_path),
hedgehogs_repo: HedgehogsRepository = Depends(get_repository(HedgehogsRepository)),
) -> int:
return await hedgehogs_repo.delete_hedgehog_by_id(hedgehog=hedgehog)
それぞれのルートは宣言的でシンプルなワンライナーになりました、ほとんどの作業はdependencyによって行われています。
このパターンについては、FastAPI のドキュメントで詳しく説明されています。
SQLクエリからも更新と削除の部分でowner指定を外すことができます:
...省略
UPDATE_HEDGEHOG_BY_ID_QUERY = """
UPDATE hedgehogs
SET name = :name,
description = :description,
age = :age,
color_type = :color_type
WHERE id = :id # 更新
RETURNING id, name, description, age, color_type, owner, created_at, updated_at;
"""
DELETE_HEDGEHOG_BY_ID_QUERY = """
DELETE FROM hedgehogs
WHERE id = :id # 更新
RETURNING id;
"""
テストを実行しすべて合格することを確認しましょう。
リファクタリングはTDDの中でも一番楽しくない部分ではありますが必要不可欠です。
このシリーズ投稿が続く間にあと何回か同様のリファクタをするかもしれません。
まとめ
これで最初のリファクタリングが完了しました。
次の投稿ではユーザーがハリネズミを予約しオーナがその予約を受け入れるかどうかを判断できるようにします。
作成したコードはGitHubにアップしています。
part10ブランチが対応しています。
https://github.com/Nmomos/hedgehog-reservation/tree/part10