Zenn
🐍

FastAPIでDIのメソッドにパラメータを渡す(同期も非同期も)

2024/08/15に公開
Python
FastAPI
tech

始めに

FastAPIにてメソッドやクラスをDIできますが、DI時に部分的に処理を差し替えたい時があります。その時にパラメータを渡せば処理を差し替えられますが、少々ハマったのでそれをブログにします。

環境

  • Python
    • 3.12.4
  • FastAPI
    • 0.112.0

実装

同期処理の場合

Dependsで呼び出す際にlambdaでパラメータを使用すれば呼び出せます。ただし、非同期処理は呼び出せないので注意してください。コンパイルエラーは発生しませんが何も起こりません。

async def async_printer(msg: str) -> None:
    print(msg)

def printer(msg: str) -> None:
    print(msg)

@router.post("/run", response_model=bool)
async def di_test(
    *,
    _: None = Depends(lambda: printer(msg="SYNC")),
    __: None = Depends(lambda: async_printer(msg="ASYNC")),
) -> Any:


# 呼び出すと printer だけしか呼ばれていないことがわかる
# SYNC

非同期処理の場合

classにしたうえで、async def __call__に定義することで処理が呼ばれます。Annotatedを使用してももちろん呼ばれます。

class DIClass:
    def __init__(self, msg: str):
        self.msg = msg

    async def __call__(
        self,
    ):
        print(self.msg)

AsyncDi = Annotated[None, Depends(DIClass(msg="ASYNC THREE"))]


@router.post("/run", response_model=bool)
async def di_test(
    *,
    ___: None = Depends(DIClass(msg="ASYNC TWO")),
    ____: AsyncDi,
) -> Any:

# 呼び出すと、次の順でログが出力される
# ASYNC TWO
# ASYNC THREE

ユースケース

  • バリデーション処理を部分的に差し替える
    • CSV処理で使用しようとしていました

処理が複雑になるので、微妙といえば微妙かもしれません。ただ、次のメリットがあるので一考の余地があるかもしれません。

  • 本処理をWriterインスタンスを使用しつつ、ValidationをReaderインスタンスにする
  • 本処理側はすでにバリデーションがかかったファイルを扱うことで、本処理を整理できる
class CsvFileValidator:
    """
    処理に必要なものをDIで取得する共通部品
    """
    def __init__(self, validator: Callable):
        self.validator = validator

    async def __call__(
        self,
        session: SessionReaderDep, # Validate処理でReaderインスタンスを取得する
        file: UploadFile=File(...),
        current_user=Security(get_current_user),
    ):
        return await self.validator(session, file, current_user)

async def csv_validation(session: AsyncSession, file: UploadFile, current_user: User) -> None:
    """
    具体的なチェック処理
    """
    pass

FileValidated = Annotated[None, Depends(CsvFileValidator(csv_validation))]

@router.post("/run", response_model=bool)
async def di_test(
    *,
    sesison: SessionWriterDep, # 本処理ではWriterインスタンスを使用する
    _: FileValidated # DI時にバリデーションをしてくれる
) -> Any:
    return True

ソースコード

終わりに

ChatGPTに確認しても、非同期処理でDI時にパラメータを与える方法を教えてくれなくて困りました。最初から公式のヘルプには書いてあったので、隅々まで読んでみることが必要ですね。

参考情報

Discussion