FastAPI を使ってWEBアプリを作ってみる その5
前回の投稿ではユニットテストを行うための土台作りをし、いくつかのテストケースを実装しました。
この投稿ではRESTの標準仕様に従い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 | 今ここ |
作成するAPIエンドポイント
今回作成するハリネズミさんを管理するリソース構造は次のとおりです。
| エンドポイント | メソッド | 概要 |
|---|---|---|
| /hedgehogs/ | POST | 新しくハリネズミさんを登録する |
| /hedgehogs/{id}/ | GET | IDでハリネズミさんを取得する |
| /hedgehogs/ | GET | 全てのハリネズミさんを取得する |
| /hedgehogs/{id}/ | PUT | IDで指定したハリネズミさんを更新する |
| /hedgehogs/{id}/ | DELETE | IDで指定したハリネズミさんを削除する |
最初の2つのエンドポイントはこれまでの投稿で実装しています。
まず最初にすべてのハリネズミさんを読み取るためのGETエンドポイントを構築しましょう。
今回もテスト駆動開発を意識し最初にテストを作成し失敗させてからコードを書いていきます。
一覧取得のエンドポイントを作成
テストケースをTestGetHedgehogに追加します。
class TestGetHedgehog:
async def test_get_hedgehog_by_id(
self,
app: FastAPI,
client: AsyncClient,
test_hedgehog: HedgehogInDB
) -> None:
res = await client.get(app.url_path_for(
"hedgehogs:get-hedgehog-by-id",
id=test_hedgehog.id
))
assert res.status_code == HTTP_200_OK
hedgehog = HedgehogInDB(**res.json())
assert hedgehog == test_hedgehog
@pytest.mark.parametrize(
"id, status_code",
(
(500, 404),
(-1, 404),
(None, 422),
),
)
async def test_wrong_id_returns_error(
self, app: FastAPI, client: AsyncClient, id: int, status_code: int
) -> None:
res = await client.get(app.url_path_for("hedgehogs:get-hedgehog-by-id", id=id))
assert res.status_code == status_code
# ここから追加
async def test_get_all_hedgehogs_returns_valid_response(
self, app: FastAPI, client: AsyncClient, test_hedgehog: HedgehogInDB
) -> None:
res = await client.get(app.url_path_for("hedgehogs:get-all-hedgehogs"))
assert res.status_code == 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この新しいテストは hedgehogs:get-all-hedgehogs にアクセスし200応答が返ることを確認し、応答がListでかつ空のリストではないことをチェックした後に HedgehogInDB モデルに強制します。
最後に、test_hedgehog fixture で作成しているレコードが応答に存在することを確認します。
この状態でテストを実行すると、おなじみの starlette.routing.NoMatchFound エラーが表示されます。
まだ hedgehogs:get-all-hedgehogs を作成していないので当然ですね。
ではこのルーティングを設定していきます。
...省略
@router.get('/',
response_model=List[HedgehogPublic],
name='hedgehogs:get-all-hedgehogs')
async def get_all_hedgehogs() -> List[HedgehogPublic]:
return None
...省略一番最初の投稿で作成していたダミーのエンドポイントを編集しました。
name にテストで宣言した get-all-hedgehogs を割り当て、固定値の値を削除しています。
再度テスト実行すると次のようなエラーに変わります。
先ほど作成したテストケースではリストが返ることを期待していますがNoneが返ってきたためエラーになっていることがわかります。
コードを次のように修正してみます。
...省略
@router.get('/',
response_model=List[HedgehogPublic],
name='hedgehogs:get-all-hedgehogs')
async def get_all_hedgehogs() -> List[HedgehogPublic]:
return [] # 変更
...省略
今度は応答のリストが空ではないことを期待していたことによるエラーへ変わりました。
リストにアイテムを含めた結果が返るようにコードを変えてみましょう。
...省略
@router.get('/',
response_model=List[HedgehogPublic],
name='hedgehogs:get-all-hedgehogs')
async def get_all_hedgehogs() -> List[HedgehogPublic]:
return [{"id": 1, "name": "test hedgehog", "age": 1}]
...省略
テストの結果が再度変わりました。
最後のアサーションでエラーが発生しています、今回応答に返したデータはダミーデータだったため test_hedgehog Fixtureで作成しているリソースが含まれていないことによるためです。
実際にデータベースにアクセスし取得した結果を応答に含めるようにする必要があります。
...省略
@router.get('/',
response_model=List[HedgehogPublic],
name='hedgehogs:get-all-hedgehogs')
async def get_all_hedgehogs(
hedgehogs_repo: HedgehogsRepository = Depends(get_repository(HedgehogsRepository))
) -> List[HedgehogPublic]:
return await hedgehogs_repo.get_all_hedgehogs()
...省略Repositoryへの依存性注入をおこない、 get_all_hedgehogs() メソッドからの戻り値を返却するように定義しました。
この状態でテストをすると AttributeError: 'HedgehogsRepository' object has no attribute 'get_all_cleanings' という理にかなった指摘が入るので早速このメソッドを作成します。
from typing import List
from app.db.repositories.base import BaseRepository
from app.models.hedgehog import HedgehogCreate, HedgehogInDB
CREATE_HEDGEHOG_QUERY = """
INSERT INTO hedgehogs (name, description, age, color_type)
VALUES (:name, :description, :age, :color_type)
RETURNING id, name, description, age, color_type;
"""
GET_HEDGEHOG_BY_ID_QUERY = """
SELECT id, name, description, age, color_type
FROM hedgehogs
WHERE id = :id;
"""
GET_ALL_HEDGEHOGS_QUERY = """
SELECT id, name, description, age, color_type
FROM hedgehogs;
"""
class HedgehogsRepository(BaseRepository):
async def create_hedgehog(self, *, new_hedgehog: HedgehogCreate) -> HedgehogInDB:
query_values = new_hedgehog.dict()
hedgehog = await self.db.fetch_one(
query=CREATE_HEDGEHOG_QUERY,
values=query_values
)
return HedgehogInDB(**hedgehog)
async def get_hedgehog_by_id(self, *, id: int) -> HedgehogInDB:
hedgehog = await self.db.fetch_one(
query=GET_HEDGEHOG_BY_ID_QUERY,
values={"id": id}
)
if not hedgehog:
return None
return HedgehogInDB(**hedgehog)
async def get_all_hedgehogs(self) -> List[HedgehogInDB]:
hedgehog_records = await self.db.fetch_all(query=GET_ALL_HEDGEHOGS_QUERY)
return [HedgehogInDB(**item) for item in hedgehog_records]
GET_ALL_HEDGEHOGS_QUERY を定義し、これを利用してレコードをモデルに変換する get_all_hedgehogs() を作成しました。
これでテストを実行すると全てのテストをパスすることができます！
PUTメソッドでの更新用エンドポイント
次はIDを指定し、対象のハリネズミさんが存在していた場合はデータを更新するエンドポイントを実装します。
おなじみテストケースの追加をまずおこないましょう。
...省略
class TestUpdateHedgehog:
@pytest.mark.parametrize("attrs_to_change, values",
((["name"],
["new fake hedgehog name"]),
(["description"],
["new fake hedgehog description"]),
(["age"],
[3.14]),
(["color_type"],
["SOLT & PEPPER"]),
(["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,
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 client.put(
app.url_path_for("hedgehogs:update-hedgehog-by-id", id=test_hedgehog.id),
json=hedgehog_update
)
assert res.status_code == 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:
assert getattr(test_hedgehog, attr) == value
@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 hedgehog type"}, 422),
(1, {"color_type": None}, 400),
),
)
async def test_update_hedgehog_with_invalid_input_throws_error(
self,
app: FastAPI,
client: AsyncClient,
id: int,
payload: dict,
status_code: int,
) -> None:
hedgehog_update = {"hedgehog_update": payload}
res = await client.put(
app.url_path_for("hedgehogs:update-hedgehog-by-id", id=id),
json=hedgehog_update
)
assert res.status_code == status_code
最初のテストケースでは、ハリネズミさんの属性ごとにいくつかのパラメーターを組み合わせで定義し更新処理をかけ、元のモデルインスタンスの保持している各属性の値が実際に書き変わっていることを確認します。
2番目のテストケースでは、さまざまな無効なペイロードとIDの組み合わせを使用し、それぞれ適切に400系ステータスコードが返ることを確認します。
このままテストを再度実行するとすべて同じ理由で失敗しますが、これを解消するためにどんな作業をすれば良いのかはもう皆さんわかっていると思います！
先ほどのGETメソッドでは、TDD思想に乗っ取り各行を一つ一つクリアしエラーを繰り返して進めましたが、今回は直接実装していきましょう。
from app.models.hedgehog import HedgehogCreate, HedgehogPublic, HedgehogUpdate
from fastapi import APIRouter, Body, Depends, HTTPException, Path
...省略
@router.put("/{id}/", response_model=HedgehogPublic,
name="hedgehogs:update-hedgehog-by-id")
async def update_hedgehog_by_id(
id: int = Path(..., ge=1, title="The ID of the hedgehog to update."),
hedgehog_update: HedgehogUpdate = Body(..., embed=True),
hedgehogs_repo: HedgehogsRepository = Depends(get_repository(HedgehogsRepository)),
) -> HedgehogPublic:
updated_hedgehog = await hedgehogs_repo.update_hedgehog(
id=id,
hedgehog_update=hedgehog_update)
if not updated_hedgehog:
raise HTTPException(
status_code=HTTP_404_NOT_FOUND,
detail="No hedgehog found with that id.")
return updated_hedgehogパス検証のためにFastAPIから Path をインポートしました。
エンドポイントの引数に ge=1 を追加したことで、idは1以上の整数でなければならないことをFastAPIに伝えています。
この制約が満たされないリクエストがきた場合、FastAPIはクライアントに HTTP_422_UNPROCESSABLE_ENTITY を返します。
エンドポイントは指定されたIDと、その更新内容を HedgehogsRepository の update_hedgehog メソッドに引き渡します。
メソッドから有効なハリネズミさんが返されなかった場合は、404 エラーを返却しリクエストで指定されたIDがデータベースに存在していなかったことを示します。
それでは update_hedgehog を書いていきます。
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
...省略
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;
"""
class HedgehogsRepository(BaseRepository):
...省略
async def update_hedgehog(
self, *, id: int, hedgehog_update: HedgehogUpdate
) -> HedgehogInDB:
hedgehog = await self.get_hedgehog_by_id(id=id)
if not hedgehog:
return None
hedgehog_update_params = hedgehog.copy(
update=hedgehog_update.dict(exclude_unset=True))
if hedgehog_update_params.color_type is None:
raise HTTPException(
status_code=HTTP_400_BAD_REQUEST,
detail='Invalid color type. Cannot be None.')
try:
updated_hedgehog = await self.db.fetch_one(
query=UPDATE_HEDGEHOG_BY_ID_QUERY,
values=hedgehog_update_params.dict()
)
return HedgehogInDB(**updated_hedgehog)
except Exception as e:
print(e)
raise HTTPException(
status_code=HTTP_400_BAD_REQUEST,
detail='Invalid update params.')
ここにはいくつかポイントがあります。
以前の投稿で定義した get_hedgehog_by_id メソッドを最初に呼び出し、IDに対応するハリネズミさんを見つけられない場合は404例外を発生させます。
この関数を経由することで HedghehogInDB モデルを返すためモデルの変換が楽になります。
pydanticのドキュメントで指定されているように、モデル上で .copy() メソッドを呼び出して、updateパラメータに変更したい内容を渡すことができます。
Pydanticは、updateは「コピーされたモデルを作成する際に変更する値のDict」であるべきであることを示しており、PUTルートで受け取ったHedgehogUpdateモデルで.dict()メソッドを呼び出すことでそれを取得しています。
また、exclude_unset=True を指定することでPydanticはモデル作成時に明示的に設定していない属性がリクエストに含まれていた場合は除外してくれるようになります。
これでテストを実行し最後までパスすることを確認してください。
DELETEメソッドでの削除用エンドポイント
最後に指定したIDのハリネズミさんを削除しましょう。
例の如くまずはテストケースです。
...省略
class TestDeleteHedgehog:
async def test_can_delete_hedgehog_successfully(
self, app: FastAPI, client: AsyncClient, test_hedgehog: HedgehogInDB
) -> None:
res = await client.delete(app.url_path_for(
'hedgehogs:delete-hedgehog-by-id', id=test_hedgehog.id))
assert res.status_code == HTTP_200_OK
res = await client.get(app.url_path_for(
'hedgehogs:get-hedgehog-by-id', id=test_hedgehog.id))
assert res.status_code == HTTP_404_NOT_FOUND
@pytest.mark.parametrize(
'id, status_code',
(
(500, 404),
(0, 422),
(-1, 422),
(None, 422),
),
)
async def test_delete_invalid_input_throws_error(
self,
app: FastAPI,
client: AsyncClient,
id: int,
status_code: int
) -> None:
res = await client.delete(app.url_path_for(
'hedgehogs:delete-hedgehog-by-id', id=id))
assert res.status_code == status_codeこれまでのテストアプローチと同じです。
まずはハリネズミさんの削除をおこない、削除後に同じIDでGETを行うことでデータベースから確実に削除されたことを確認します。
その後はいくつかの無効なIDをパラメータに指定し、それぞれ適切なステータスが返ることを確認します。
それではロジック側を実装しましょう。
...省略
@router.delete('/{id}/', response_model=int, name='hedgehogs:delete-hedgehog-by-id')
async def delete_hedgehog_by_id(
id: int = Path(..., ge=1, title='The ID of the hedgehog to delete.'),
hedgehogs_repo: HedgehogsRepository = Depends(get_repository(HedgehogsRepository))
) -> int:
deleted_id = await hedgehogs_repo.delete_hedgehog_by_id(id=id)
if not deleted_id:
raise HTTPException(
status_code=HTTP_404_NOT_FOUND,
detail='No hedgehog found with that id.')
return deleted_idこれでテストを実行するといつものエラーが発生します。
これに対処するためRepositoryにメソッドを実装しましょう。
...省略
DELETE_HEDGEHOG_BY_ID_QUERY = '''
DELETE FROM hedgehogs
WHERE id = :id
RETURNING id;
'''
...省略
async def delete_hedgehog_by_id(self, *, id: int) -> int:
hedgehog = await self.get_hedgehog_by_id(id=id)
if not hedgehog:
return None
deleted_id = await self.db.execute(
query=query.DELETE_HEDGEHOG_BY_ID_QUERY, values={'id': id})
return deleted_idまとめ
今回の投稿では前回の投稿で作成したテスト環境を使用してRESTfulエンドポイントを実装しました。
投稿の中では pytest を使用しテストを繰り返していきましたが、FastAPIが提供するOpenAPIドキュメントを使用し手動でのテストも行うことを推奨します。
次回の投稿ではユーザーモデルを作成し、認証機構の準備をしていきます。
作成したコードはGitHubにアップしています。
part5ブランチが対応しています。
https://github.com/Nmomos/hedgehog-reservation/tree/part5