DDDが好きなPHPerがFastAPIを触ってみる

きっかけ

偉大なWeb＋DB Pressの隔月最終号でFastAPIを特集していたので、触ってみました。

gihyo.jp

PHPを4年ほど書いており、動的型付け言語ですがDDDに影響を受けて
・変数・戻り値に型を必ず定義する
・イミュータブルなクラスを使用する
・「依存性の逆転」を利用してDBや外部サービスに対して疎結合にする
などを意識しており、FastAPI(Python)でも可能なのかと思い触ってみました。

GitHub - yuktake/expenditure-management-api: web+db sample fastapi application

Pythonの印象

型が指定できない

型ヒントによって変数や戻り値の型を明示することができるが、異なる型を代入してもエラーが起こらない。

インターフェースがない

「依存性の逆転」ができない？

試したいこと

型エラーを起こす

明示した型ヒントと異なる型を代入した場合にエラーを起こす。

イミュータブルなクラス

インスタンス生成後、内部状態を変えられないようにする。

「依存性の逆転」の実現

任意の条件に応じてDIするクラスを変更する
インターフェースの代替になるものを使用して「依存性の逆転」を実現する

結果

型エラーを起こす→Pydanticの使用

PydanticのBaseModelを継承したクラスを実装することで、異なった型が入力された場合ステータスコード422を返してくれます。
またFastAPIではエンドポイントのRequest/Responseクラスを指定することができ、Pydanticのクラスを指定することでパラメータの型を自動でバリデーションしてくれます。

api/wallets/view.py

@router.get("", response_model=GetWalletsResponse)
async def get_wallets(
    use_case: Annotated[ListWallets, Depends(ListWallets)],
) -> GetWalletsResponse:
    """Walletの一覧取得API"""
    return GetWalletsResponse(
        wallets=[Wallet.model_validate(w) for w in await use_case.execute()]
    )

イミュータブルなクラス→dataclassesモジュールの使用

dataclassesモジュールの@dataclassをクラスの先頭に追記することで、クラスをイミュータブルにすることができます。
これによって一度生成されたインスタンスの変数を変更しようとするとエラーが起きるようになります。
またコンストラクタでself.model_validate(self)を呼び出すことで、不完全なインスタンスを生成できないようにします。

models.py

from datetime import datetime
from enum import StrEnum
from typing import Annotated
from pydantic import BaseModel as _BaseModel
from pydantic import ConfigDict, PositiveInt, Field, validator
from pydantic import WrapSerializer
from utils.datetime import to_utc
from typing import Any
from dataclasses import dataclass

UTCDatetime = Annotated[datetime, WrapSerializer(to_utc)]

class BaseModel(_BaseModel):
    # 変数名を揃えることでORMインスタンスからPydanticインスタンスに変換できる？
    model_config = ConfigDict(from_attributes=True)

class HistoryType(StrEnum):
    INCOME = "INCOME"
    OUTCOME = "OUTCOME"

@dataclass(frozen=True)
class History(BaseModel):
    def __init__(self, **kwargs):
        super().__init__(**kwargs)
        self.model_validate(self)

    history_id: int|None
    name: str
    amount: PositiveInt
    type: HistoryType
    history_at: UTCDatetime
    wallet_id: int

@dataclass(frozen=True)
class Wallet(BaseModel):
    def __init__(self, **kwargs):
        super().__init__(**kwargs)
        self.model_validate(self)

    wallet_id: int|None
    name: str = Field(..., min_length=4, max_length=10)
    histories: list[History]

    @property
    def balance(self) -> int:
        return sum(
            h.amount
            if h.type == HistoryType.INCOME
            else -h.amount
            for h in self.histories
        )

    # 独自関数でのValidationも可能
    # @validator("name", pre=True)
    # def test(cls, v: Any) -> Any:
    #     raise NotImplementedError()

「依存性の逆転」の実現→抽象クラスの使用

Pythonにはインターフェースはありませんが、代わりに抽象クラスで代替します。
抽象クラスの関数でNotImplementedErrorを書いておくことで、継承クラスでの実装を強制できます。

抽象クラス(repositories/abstract_wallet_repository.py)

class AbstractWalletRepository(ABC):

    @abstractmethod
    async def add(self, session: AsyncSession, name: str) -> Wallet:
        raise NotImplementedError()

継承したクラス(repositories/wallet_repository.py)

class WalletRepository(AbstractWalletRepository):

    async def add(self, session: ass, wallet: Wallet) -> Wallet:
        wallet_orm = WalletORM(name=wallet.name, histories=[])
        session.add(wallet_orm)
        await session.flush()
        return wallet_orm.to_entity()

またDependency InjectionもFastAPIのDependsを使用することで設定が可能です。

DIの設定(dependencies/repository.py)

settings = Settings()

if settings.status == "testing":
    WalletRepositoryInterface = Annotated[
        AbstractWalletRepository, Depends(TestWalletRepository)
    ]
else:
    WalletRepositoryInterface = Annotated[
        AbstractWalletRepository, Depends(WalletRepository)
    ]

※PydanticのSettingsConfigDictを使用することで、.envファイルの値によってDIするクラスを変更しています。

DIの使用(api/wallets/use_case.py)

class ListWallets:

    def __init__(
        self,
        session: SessionInterface,
        repo: WalletRepositoryInterface,
    ) -> None:
        self.session = session
        self.repo = repo

    async def execute(self) -> list[Wallet]:
        sess = self.session.get_session()
        async with sess() as s:
            wallets = await self.repo.get_all(s)
        return wallets

参考記事

github.com