きっかけ
偉大なWeb＋DB Pressの隔月最終号でFastAPIを特集していたので、触ってみました。
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
