📑

Moto と PynamoDB で DynamoDB を用いたAPIのテストを書く方法

2023/02/04に公開

Python

はじめに

こんにちは。@hayata-yamamotoです。

DynamoDB のようなリソースを用いてプロダクトを開発する際、ローカル用にエミュレーターを起動し、そこにテスト用のデータを詰め込んで利用することがあると思います。よくあるのはテストのために docker-compose.yml を書いて、docker compose up -d しテストを実行するパターンです。しかしこれは、徐々にサービスが大きくなり関連するリソースが増えてきた時に docker-compose.yml 自体のメンテナンスがめんどくさくなってくる問題を抱えていました。

弊社では、この問題を少しでも緩和できればと考え、Moto という AWS のモックアップライブラリを用いてテストを実装しています。もちろん、AWS の進化はとても早く、API の機能も目まぐるしく変わるため、100％モックアップされるわけではありませんが、Moto を用いることによってローカルでエミュレーターを起動する作業が少し軽減されています。Moto については、以前紹介していますのでそちらもご覧ください

今回は、この Moto と DynamoDB の Pythonic なインタフェースである^[1] PynamoDB を用いて、どのように API の定義や実装を行い、テストコードを実装しているかをサンプルコードを交えて紹介します。DynamoDB を Python で扱い、かつ API サービスを提供する際の参考となりましたら幸いです。

なお、今回の記事に用いたコードは以下のリポジトリで公開されています。「記事ではちょっとわかりにくいな」と思った場合はそちらも合わせてご覧ください。

FastAPI と PynamoDB で API を構築する

FastAPI のドキュメントで紹介されているサンプルコードを参考に、あるアイテムを作成・更新するエンドポイントを FastAPI と DynamoDB で提供したいとします。

ひとおもいに実装してみると以下のような感じでしょうか。UI の関係で、単一ファイルに実装する想定としていますが、もちろんファイルは分割してもらって問題ありません。

main.py

from uuid import uuid4

from fastapi import Depends, FastAPI, HTTPException
from pydantic import BaseModel
from pynamodb import attributes, models
from pynamodb.exceptions import DoesNotExist


class Item(models.Model):
    class Meta:
        region = "ap-northeast-1"
        table_name = "item"

    id = attributes.UnicodeAttribute(hash_key=True)
    name = attributes.UnicodeAttribute()
    description = attributes.UnicodeAttribute(null=True, default=None)
    price = attributes.NumberAttribute()
    tax = attributes.NumberAttribute(null=True, default=None)
    version = attributes.VersionAttribute()


class PostRequest(BaseModel):
    name: str
    description: str | None = None
    price: float
    tax: float | None = None


class PutRequest(PostRequest):
    version: int


class ItemResponse(BaseModel):
    id: str
    name: str
    description: str | None
    price: float
    tax: float | None
    version: int


app = FastAPI()


@app.post("/items/", response_model=ItemResponse, response_model_exclude_none=True)
def create_item(request: PostRequest) -> ItemResponse:
    item = Item(
        id=str(uuid4()),
        name=request.name,
        description=request.description,
        price=request.price,
        tax=request.tax,
    )
    item.save()
    return ItemResponse.parse_raw(item.to_json())


def get_valid_item(item_id: str) -> Item:
    try:
        return Item.get(item_id)
    except DoesNotExist:
        raise HTTPException(status_code=404, detail="Item not found")


@app.put(
    "/items/{item_id}", response_model=ItemResponse, response_model_exclude_none=True
)
def put_item(request: PutRequest, item: Item = Depends(get_valid_item)) -> ItemResponse:
    actions = [
        Item.name.set(request.name),
        Item.description.set(request.description),
        Item.price.set(request.price),
        Item.tax.set(request.tax),
    ]
    item.update(actions=actions)
    return ItemResponse.parse_raw(item.to_json())

Pydantic の BaseModel でリクエストとレスポンスの型を定義し、それらを元に PynamoDB で作成したモデルで Item を作成して、結果を返却する、とてもシンプルなものになっています。楽観ロックをかけたい時のバージョンは PynamoDB がよしなにやってくれるため、 update を実行する際にバージョン不整合があれば自動でエラーが Raise されるようになっています。

PynamoDB と Moto, Starlette Test Client を用いて API の振る舞いをテストする

では、上記の API に対してテストを書いてみます。具体的には以下のような形となります。

test_main.py

from unittest import TestCase
from uuid import uuid4

from fastapi.exceptions import HTTPException
from moto import mock_dynamodb
from starlette.testclient import TestClient

from main import Item, app, get_valid_item


class TestApp(TestCase):
    def setUp(self) -> None:
        self.mock_dynamodb = mock_dynamodb()
        self.mock_dynamodb.start()

        self.test_client = TestClient(app)
        Item.create_table()

        self.test_item = Item(
            id=str(uuid4()), name="test", description="test", price=100, tax=0.1
        )
        self.test_item.save()

    def tearDown(self) -> None:
        self.mock_dynamodb.stop()

    def test_create_item(self) -> None:
        payload = {"name": "item1", "description": "item1 description", "price": 100}
        r = self.test_client.post(url="/items", json=payload)
        self.assertEqual(200, r.status_code)

        item = Item.get(r.json()["id"])
        self.assertTrue(item.exists())

    def test_get_valid_item(self) -> None:
        item = get_valid_item(self.test_item.id)
        self.assertTrue(item.exists())

        with self.assertRaises(HTTPException):
            get_valid_item("dummy-id")

    def test_put_item(self) -> None:
        payload = {
            "name": "item1",
            "description": "item1 description",
            "price": 100,
            "version": self.test_item.version,
        }
        r = self.test_client.put(url=f"/items/{self.test_item.id}", json=payload)
        self.assertEqual(200, r.status_code)
        self.assertEqual(self.test_item.version + 1, r.json()["version"])

        after_item = Item.get(self.test_item.id)
        self.assertNotEqual(after_item.serialize(), self.test_item.serialize())

Moto で DynamoDB に対する API コールをモックできるようになっています。テストの際は、モック用の DynamoDB を起動したのち、PynamoDB が用意している create_table のラップメソッドを直接実行してしまって問題ありません。開発環境や本番環境に提供するのと同じコードでテストを実施することができ、実装時も docker コンテナを起動したりする手間がなくなります。

ここまで書いたら最後に

$ poetry run python -m unittest test_main.py

のようにテストを実行し、全てのテストケースが OK になることを確認すれば一連の開発は完了となります。

おわりに

今回は、簡単ではありますが Moto と PynamoDB を用いて FastAPI で実装された API のテスト方法を紹介しました。実際には、もう少し複雑な処理を API に書き、それを DynamoDB だけではなく、AWS のさまざまなリソースをモックしながら API の挙動を確認しています。

最後に、弊社ではトドケールの開発を牽引してくださるエンジニアのみなさんを募集しています！募集中のロールは以下のリンクよりご確認いただけます。気になった方はいつでもご連絡くださいませ！

脚注

https://pynamodb.readthedocs.io/en/stable/#welcome-to-pynamodb-s-documentation ↩︎

はじめに

FastAPI と PynamoDB で API を構築する

PynamoDB と Moto, Starlette Test Client を用いて API の振る舞いをテストする

おわりに

Discussion