きり丸の技術日記

技術検証したり、資格等をここに残していきます。

Pythonのenumをintで継承しなくてもIntEnumが標準化されてた

Python 技術

始めに

過去にPythonのenumはint等のプリミティブ型を継承すると便利という記事を投稿していました。

その後、Python3.11にてIntEnumやStrEnumが標準化されていることを知ったので共有します。

環境

  • Python
    • 3.12.4

実装

intEnumの使用方法は次のとおりです。

from enum import IntEnum

class IntInheritEnum(IntEnum):
    ID = 1


# 列挙型ではなく、intとして振舞うため1にアクセスできる
IntInheritEnum.ID

# 本来のアクセス方法
IntInheritEnum.ID.value

ちなみに、intを継承したEnumとの相違点としてはIntEnumがReprEnumを継承していることで、str()で期待する値を出力してくれる点がいいですね。実運用ではあまり大きな違いはないですが。

class IntInheritEnum(IntEnum):
    ID = 1

class IntInheritEnum2(int, Enum):
    ID = 1

# ログで動作確認するときにちょっとうれしい
print(IntInheritEnum.ID)
# 1
print(IntInheritEnum2.ID)
# IntInheritEnum2.ID

from pydantic import BaseModel
class Tmp(BaseModel):
    aaa: IntInheritEnum
    bbb: IntInheritEnum2

# pydanticでインスタンスをログ出力するときには効かないので微妙
# どこかのアップデートでReprEnumが使用されるようになる…？
print(Tmp(aaa=IntInheritEnum.ID, bbb=IntInheritEnum2.ID))
#  {'aaa': <IntInheritEnum.ID: 1>, 'bbb': <IntInheritEnum2.ID: 1>}

注意点

前回の記事でも記載した注意点ですが、Enumの型のままintとしても振舞えるので型ヒントをつけたりすると混乱します。

# intとして振舞うのでエラーならない
result: int = IntInheritEnum.ID + 10

# 型が違うのに計算できてしまうのは確かに違和感がある
def add(num1: int, num2: IntInheritEnum):
    return num1 + num2

ソースコード

終わりに

前の継承も難しい構文ではなかったのですが、ちょっと混乱させてしまう点があったのでPythonの標準としてIntEnumとStrEnumを定義していただいて助かります。堂々と使える点が良いです。

参考記事

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

FastAPI Python 技術

始めに

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時にパラメータを与える方法を教えてくれなくて困りました。最初から公式のヘルプには書いてあったので、隅々まで読んでみることが必要ですね。

参考情報

ランキング参加中
Python
ランキング参加中
プログラミング

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

FastAPI Python 技術

始めに

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側に定義を反映できます。

ソースコード

終わりに

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

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

参考情報