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

始めに

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

ソースコード

• https://github.com/hirotoKirimaru/fastapi-practice/blob/ec5b37313dafa6b69884b985f6b2092fe34b2c2a/src/api/routers/dependency_injection.py

終わりに

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

参考情報

• SpringのBean定義(Java Config)で型が重複する場合のインジェクション方法
  • https://ik.am/entries/377
• FastAPI公式ヘルプのパラメータ化された依存関係
  • https://fastapi.tiangolo.com/advanced/advanced-dependencies/

FastAPIで任意のHttpStatusと任意の型を返却する(JSONResponse)

始めに

FastAPIではHTTPExceptionを使用すればHttpStatusを200以外でも返却可能です。しかし、エラー詳細は1つしか返却できません。ファイルのエラーハンドリングでは多数のエラーが発生した場合には、複数のエラーを返却したいユースケースがあります。

ExceptionHandlerを使用することでもHTTPStatusや複数エラーを返却するようにできますが、基本的にはファイルバリデーション処理でしか使用しないため、全体に適用するには範囲が大きすぎます。

今回の記事ではファイルのバリデーション結果を返却するために次の仕様を満たせるようにします。

1. バリデーションに失敗したらHTTPStatus200以外を返却
2. エラーのBodyには複数の理由を設定できる
3. ExceptionHandlerを使用しない
4. OpenAPIにもレスポンス構造を反映できる

環境

• Python
  • 3.12.4
• FastAPI
  • 0.112.0

実装

レスポンスの型を定義する

FastAPIのデフォルト機能として、HTTPExceptionをraiseすると次のレスポンス型で返却されます。

{
    "detail": "FILE_INVALID"
}

今回は他でも流用しやすくするため、このレスポンスをベースに型を拡張します。

{
  "detail": "FILE_INVALID",
  // ここから下を追加する形のレスポンスにする
  "errorLists": {
    "indexes": [
      1,
      2,
      3
    ],
    "reason": "Not Found"
  }
}

具体的には次のような型定義をします。後でOpenAPI側でもデータ構造をわかるようにしたいので、examplesまで定義しておきます。

from typing import Any, List, Annotated
from pydantic import BaseModel, Field

class ErrorMessage(BaseModel):
    reason: Annotated[str, Field(description="")]
    indexes: Annotated[List[int], Field(description="エラーが発生したインデックスのリスト")]


class ExceptionResponse(BaseModel):
    detail: Annotated[str, Field(description="Exception detail", examples=["File Invalid"])]
    error_lists: Annotated[List[ErrorMessage], Field(description="Exception details", examples=[{"reason": "Not Found", "indexes": [1, 2, 3]}])]

HttpStatus 200以外で返却する

JSONResponseを返却することで任意のHTTPStatusコードや任意の型定義を返却できます。

今回はファイルの内容が誤っていたことを伝えたいので、HTTPStatus=422で返却します。また今回の記事では省略していますが、Pythonではスネークケース、Responseではキャメルケースにしたいのでby_alias=Trueも付与しておきます。

return JSONResponse(
    status_code=422,
    content=ExceptionResponse(detail="File Invalid", error_lists=[]).model_dump(by_alias=True)
)

OpenAPIにもレスポンスの構造を伝える

APIRouterのresponsesを定義するとOpenAPI側にレスポンスの型を伝えられます。そのため、先ほど定義したExceptionResponseを指定します。

@router.get("/files", responses={422: {"model": ExceptionResponse}})

結果

OpenAPI側に定義を反映できます。

ソースコード

• https://github.com/hirotoKirimaru/fastapi-practice/blob/9028f918c293568032a6789d7eaae75ac939513f/src/api/routers/exception.py

終わりに

ファイルのバリデーションは雑にやるならHTTPExceptionを返却するだけで済むのですが、だいたいユーザビリティが低いのでエラーをまとめなければならず、毎回実装するたびに悩んでいる気がします。

ファイルバリデーションのあるべきハンドリングの記事があればぜひ参考にしたいです。

参考情報

• レスポンスを直接返す - FastAPI
• Additional Responses in OpenAPI - FastAPI

PythonのPydanticでdatetimeのシリアライズフォーマットを決める

始めに

Pythonにてアプリケーション内ではdatetimeとして扱いつつ、APIとしてはYYYY-mm-dd等の特定のフォーマットの文字列で返却したいことがあります。

今回はPydanticを用いて実装する方法を記事にします。

環境

• Python
  • 3.12.4
• Pydantic
  • 2.8.2

実装

field_serializer を使用する

field_serializerを用いることで、シリアライズ時にフォーマットしてくれるようになります。

次のコードでは、特に何もしていなければISO8601のフォーマットで返却し、field_serializerが有効な個所では%Y-%m-%d %H:%M:%Sで返却するテストを記載しています。

from datetime import datetime
from zoneinfo import ZoneInfo
from pydantic import BaseModel, field_serializer, PlainSerializer
import json
from typing import Annotated

class _TestModel(BaseModel):
    dt: datetime
    normal_dt: datetime

    @field_serializer("dt")
    def serialize_datetime(self, dt: datetime, _info):
        return dt.strftime("%Y-%m-%d %H:%M:%S")


class TestFieldSerializer:
    async def test_01(self):
        dt = datetime(2024, 12, 1, 2, 3, 4, tzinfo=ZoneInfo("UTC"))
        model = _TestModel(dt=dt, normal_dt=dt)
        actual_str = model.model_dump_json()
        actual_dict = json.loads(actual_str)

        assert actual_dict.get("dt") == "2024-12-01 02:03:04"
        assert actual_dict.get("normal_dt") == "2024-12-01T02:03:04Z"

PlainSerializer で型を表現する

AnnotatedとPlainSerializerを組み合わせることで型として表現できます。各モデルごとに定義する必要がないので、基本的にはこちらを採用するほうがメリットがあります。

CustomDatetime = Annotated[
    datetime,
    PlainSerializer(lambda dt: dt.strftime("%Y-%m-%d %H:%M:%S"))
]

class _TestModel2(BaseModel):
    dt: CustomDatetime
    normal_dt: datetime


class TestCustomDatetime:
    async def test_serialize(self):
        dt = datetime(2024, 12, 1, 2, 3, 4, tzinfo=ZoneInfo("UTC"))
        model = _TestModel2(dt=dt, normal_dt=dt)
        actual_str = model.model_dump_json()
        actual_dict = json.loads(actual_str)

        assert actual_dict.get("dt") == "2024-12-01 02:03:04"
        assert actual_dict.get("normal_dt") == "2024-12-01T02:03:04Z"

    async def test_deserialize(self):
        # GIVEN
        dt = datetime(2024, 12, 1, 2, 3, 4, tzinfo=ZoneInfo("UTC"))
        model = _TestModel2(dt=dt, normal_dt=dt)
        actual_str = model.model_dump_json()

        # WHEN
        actual = _TestModel2.model_validate_json(actual_str)
        # シリアライズ時にTZが削除されている
        assert actual.dt == dt.replace(tzinfo=None)
        assert actual.normal_dt == dt

ソースコード

• fastapi-practice/tests/unit/schemas/test_datetime_serialize.py at e002680c4f5d880ca39b347fcbfe58de4104ea9b · hirotoKirimaru/fastapi-practice · GitHub

終わりに

基本的にはフロントでフォーマットすればいいのですが、全処理をバックエンドに押し付ける場合にはこういう処理も可能です。

また、CSV作成処理等でも流用できます。

参考情報

• field_serializer
• PlainSerializer