ラベル FastAPI の投稿を表示しています。 すべての投稿を表示
ラベル FastAPI の投稿を表示しています。 すべての投稿を表示

2024年6月19日水曜日

FastAPI のデフォルトのバリデーションエラーをハンドリングするサンプルコード

FastAPI のデフォルトのバリデーションエラーをハンドリングするサンプルコード

概要

RequestValidationError をハンドリングします

環境

  • macOS 11.7.10
  • Python 3.11.6
  • fastapi 0.111.0

サンプルコード

Request と RequestValidationError のリファレンスは参考サイトに記載しています

from fastapi import FastAPI
from fastapi.exceptions import RequestValidationError
from pydantic import BaseModel, Field
from starlette.requests import Request
from starlette.responses import JSONResponse

app = FastAPI()


class Item(BaseModel):
    message: str = Field(max_length=10)


@app.exception_handler(RequestValidationError)
async def validation_exception_handler(request: Request, exc: RequestValidationError):
    # print(request)
    errs = exc.errors()
    print(errs)
    return JSONResponse(
        {"error_field": errs[0]["loc"][1], "message": errs[0]["msg"]}, status_code=400
    )


@app.post("/")
async def post_msg(item: Item):
    return {
        "message": item.message,
    }

最後に

errors で取得した情報は配列なので扱いには注意しましょう

参考サイト

2024年4月8日月曜日

fastapi にリクエストIDを自動で埋め込む方法 (RequestIdPlugin編)

fastapi にリクエストIDを自動で埋め込む方法 (RequestIdPlugin編)

概要

starlette には RequestIdPlugin という仕組みがありこれを使うと context 経由でリクエストIDを取得できます
アプリコンテキスト配下であれば同一のリクエストIDが取得できるのでロギングなどに使えます

環境

  • macOS 11.7.10
  • Python 3.11.6
    • fastapi 0.110.1
    • starlette 0.37.2
    • starlette-context 0.3.6

サンプルコード

from fastapi import FastAPI
from pydantic import BaseModel
from pydantic_settings import BaseSettings
from starlette.middleware import Middleware
from starlette_context import context
from starlette_context.middleware import RawContextMiddleware
from starlette_context.plugins.correlation_id import CorrelationIdPlugin
from starlette_context.plugins.request_id import RequestIdPlugin

# ミドルウェア定義
# これを定義することで context にリクエスト ID が自動で保持される
middleware = [
    Middleware(RawContextMiddleware, plugins=(RequestIdPlugin(), CorrelationIdPlugin()))
]
# ミドルウェア追加
app = FastAPI(middleware=middleware)


class Settings(BaseSettings):
    message: str = ""


settings = Settings()


class Item(BaseModel):
    message: str


@app.get("/")
async def get_msg():
    # コンテキスト情報からリクエストIDの取得
    return {
        "message": settings.message,
        "context": context,
    }


@app.post("/")
def set_msg(item: Item):
    settings.message = item.message
    return {"message": item.message}

最後に

RequestIdPlugin を使って fastapi にリクエストIDを自動生成する方法を紹介しました
context を使うのでアプリコンテキスト配下のモジュール出ないと参照できないのでそこは注意しましょう

参考サイト

2023年9月29日金曜日

pydanticV2 で自作クラス側の field_validator に info.data が渡らない問題をどう解決するか

pydanticV2 で自作クラス側の field_validator に info.data が渡らない問題をどう解決するか

概要

以下にサンプルコードを掲載しますが pydanticV2 を使っていて自作クラスを型に指定しその自作クラス側で FieldValidationInfo を使ってバリデーションした場合に info.data が空になるという問題があります

解決する方法としてクラスをまとめるという方法がありますがそれだとクラスが膨大になってしまいます
今回はその場合の解決策を紹介します

環境

  • Python 3.11.3
  • fastapi 0.100.1
  • pydantic 2.1.1

サンプルコード (info.data が空になる)

以下の場合 name -> age という順番でバリデーションされます
pydantic V2 の場合、バリデーションを通過した値はすべて FieldValidationInfo の info.data というオブジェクトに格納されます

しかし以下のように独自クラス側に age を分けている場合に info.data の値が空になるためバリデーションに成功した name を参照することができません

これを解決する方法として Age クラスを使わずに User 側にバリデーションを寄せる方法がありますがそれだとクラスが大きくなり微妙です

from pydantic import BaseModel, Field, FieldValidationInfo, field_validator


class Age(BaseModel):
    age: int = Field()

    @field_validator("age")
    @classmethod
    def validate_age(cls, v: int, info: FieldValidationInfo):
        # print(info.data["name"]) ## KeyError になる
        if info.context is None:
            return v
        if "age_limit" not in info.context or not isinstance(
            info.context["age_limit"], int
        ):
            raise ValueError("info.context.age_limit が正しく設定されていません")
        print("validate_age_with_context")
        limit = int(info.context["age_limit"])
        if v > limit:
            raise ValueError(f"age が limit={limit} を超えています")
        return v


class User(BaseModel):
    name: str = Field()
    age: Age = Field()

    @field_validator("name")
    @classmethod
    def validate_name(cls, v: str, info: FieldValidationInfo):
        if info.context is None:
            return v
        if "allowed_name" not in info.context or not isinstance(
            info.context["allowed_name"], str
        ):
            raise ValueError("info.context.allowed_name が正しく設定されていません")
        print("validate_name")
        if v not in [info.context["allowed_name"]]:
            raise ValueError()
        return v


if __name__ == "__main__":
    user = User.model_validate(
        {"name": "hawksnowlog", "age": {"age": 0}}, context={"allowed_name": "hawksnowlog", "age_limit": 9}
    )

解決方法 (context を無理やり使う

自分が発見した方法は info.context を使う方法です
これをすることで info.data ではなく info.context を Age 側で参照することで検証済みの name を使うことができます

from pydantic import BaseModel, Field, FieldValidationInfo, field_validator


class Age(BaseModel):
    age: int = Field()

    @field_validator("age")
    @classmethod
    def validate_age(cls, v: int, info: FieldValidationInfo):
        # info.context を使うように変更
        if info.context is None:
            raise ValueError()
        print(info.context["validated_name"])
        if info.context is None:
            return v
        if "age_limit" not in info.context or not isinstance(
            info.context["age_limit"], int
        ):
            raise ValueError("info.context.age_limit が正しく設定されていません")
        print("validate_age_with_context")
        limit = int(info.context["age_limit"])
        if v > limit:
            raise ValueError(f"age が limit={limit} を超えています")
        return v


class User(BaseModel):
    name: str = Field()
    age: Age = Field()

    @field_validator("name")
    @classmethod
    def validate_name(cls, v: str, info: FieldValidationInfo):
        if info.context is None:
            return v
        if "allowed_name" not in info.context or not isinstance(
            info.context["allowed_name"], str
        ):
            raise ValueError("info.context.allowed_name が正しく設定されていません")
        print("validate_name")
        if v not in [info.context["allowed_name"]]:
            raise ValueError()
        return v

    # mode=before にすることで Field() に記載されている評価よりも前にこっちを実行させる
    @field_validator("age", mode="before")
    @classmethod
    def validate_age(cls, v: Age, info: FieldValidationInfo):
        if info.context is None:
            raise ValueError()
        # 同じクラス内では info.data が引き継がれるので検証済みの name が参照できる
        info.context["validated_name"] = info.data["name"]
        Age.model_validate(v, context=info.context)
        return v


if __name__ == "__main__":
    user = User.model_validate(
        {"name": "hawksnowlog", "age": {"age": 0}},
        context={"allowed_name": "hawksnowlog", "age_limit": 9},
    )

最後に

他に自作クラス側に info.data を渡す方法はあるのだろうか

info.data の説明を見ると同一クラス内で参照可能とあるので Age クラスと User が継承するようにすれば参照できるようになるはず

なので継承しない独自クラスを作る場合には別のモデルクラスとフィールドが相互参照するようなモデルの設計をしないほうが無難な気がする

2023年9月28日木曜日

pydanticV2 の model_validate は逐次評価ではなくすべてを評価してから ValidationError が raise される

pydanticV2 の model_validate は逐次評価ではなくすべてを評価してから ValidationError が raise される

概要

特に context (FiledInfoValidator) と組み合わせる場合には注意が必要かなと思います

結論としてはエラーとなりそうなところがあればちゃんと事前に None チェックをしましょうという話です

環境

  • Python 3.11.3
  • fastapi 0.100.1
  • pydantic 2.1.1

サンプルコード

from fastapi import FastAPI
from pydantic import BaseModel, Field, FieldValidationInfo, field_validator

app = FastAPI()


class User(BaseModel):
    name: str = Field()
    age: int = Field()

    @field_validator("name")
    @classmethod
    def validate_name(cls, v: str, info: FieldValidationInfo):
        if info.context is None:
            return v
        print("validate_name")
        if v not in ["hawksnowlog"]:
            raise ValueError()
        return v

    @field_validator("age")
    @classmethod
    def validate_age_with_context(cls, v: int, info: FieldValidationInfo):
        if info.context is None:
            return v
        if "age_limit" not in info.context or not isinstance(
            info.context["age_limit"], int
        ):
            raise ValueError("info.context.age_limit が正しく設定されていません")
        # ここがポイント
        # 前の validate_name で ValueError になると raise されない
        # 更に info.data["name"] には何も設定されずに来るのでここで KeyError が発生してしまう
        if info.data["name"] not in ["hawksnowlog"]:
            raise ValueError()
        # もしやるなら以下のように None チェックしてから info.data["name"] を扱うようにする
        if info.data.get("name") is None:
            raise ValueError()
        if info.data["name"] not in ["hawksnowlog"]:
            raise ValueError()
        print("validate_age_with_context")
        limit = int(info.context["age_limit"])
        if v > limit:
            raise ValueError(f"age が limit={limit} を超えています")
        return v


@app.post("/")
async def test_context(user: User):
    # ルーティング時の User 生成時にコールされるバリデーションはどこかで raise すればエラーになる
    # model_validate の場合は raise 時にエラーとならずすべてが評価されたあとで raise される
    user = User.model_validate(
        {"name": "hawksnowlog2", "age": 0}, context={"age_limit": 9}
    )
    return user

ちょっと解説

基本はコードのコメント部分に書いてあるとおりです

model_validate と context を使ってバリデーションした場合には途中で raise ValueError したものが後ろに回されてまとめて raise されます
なのでバリデーションのコードの中で KeyError など予期せぬエラーが発生した場合にはそこでバリデーションが止まってしまい本当に取得したい ValidationError が取得できなくなってしまいます

なので None 参照や Key 参照をする場合はちゃんと存在するかチェックした上でバリデーションを組む必要があります

2023年9月25日月曜日

PydanticV2 の FieldValidationInfo にはデータベースへのセッションも渡すことができる

PydanticV2 の FieldValidationInfo にはデータベースへのセッションも渡すことができる

概要

前回 FieldValidationInfo の基本的な使い方を紹介しました

今回は db のセッションが渡せることが判明したのでそのサンプルコードを紹介します

環境

  • Python 3.11.3
  • fastapi 0.100.1
  • pydantic 2.1.1

サンプルコード

from fastapi import Depends, FastAPI
from pydantic import BaseModel, Field, FieldValidationInfo, field_validator
from sqlalchemy import Column, Integer, String, create_engine
from sqlalchemy.orm import Session, declarative_base, sessionmaker

app = FastAPI()

engine = create_engine("mysql+pymysql://root@localhost/test?charset=utf8mb4")
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
Base = declarative_base()


class UserTable(Base):
    __tablename__ = "user"

    id = Column(Integer, primary_key=True)
    name = Column(String(50))
    age = Column(Integer)


def get_db():
    db: Session = SessionLocal()
    try:
        yield db
        db.commit()
    except Exception:
        db.rollback()
    finally:
        db.close()


class User(BaseModel):
    name: str = Field()
    age: int = Field()

    @field_validator("name")
    @classmethod
    def validate_name(cls, v: str, info: FieldValidationInfo):
        if info.context is not None:
            # context がある場合のバリデーションは無視する
            return v
        print("validate_name")
        if v not in ["hawksnowlog"]:
            raise ValueError()
        return v

    @field_validator("age")
    @classmethod
    def validate_age_with_context(cls, v: int, info: FieldValidationInfo):
        if info.context is None:
            # context がない場合のバリデーションは無視する
            return v
        if "age_limit" not in info.context or not isinstance(
            info.context["age_limit"], int
        ):
            # context の値をチェック、正しくセットされていない場合はエラー
            raise ValueError("info.context.age_limit が正しく設定されていません")
        if "db" not in info.context or not isinstance(info.context["db"], Session):
            # context の値をチェック、正しくセットされていない場合はエラー
            raise ValueError("info.context.db が正しく設定されていません")
        print("validate_age_with_context")
        # db オブジェクトが使えるかテスト
        users = info.context["db"].query(UserTable).all()
        for user in users:
            print(user.name)
        limit = int(info.context["age_limit"])
        if v > limit:
            raise ValueError(f"age が limit={limit} を超えています")
        return v


@app.post("/")
async def test_context(user: User, db: Session = Depends(get_db)):
    # ルーティングで一度 User のバリデーションをコールする
    user_dict = user.model_dump()
    # ここで context を付与して再度バリデーションをコールする
    user = User.model_validate(user_dict, context={"age_limit": 9, "db": db})
    return user

ポイント

ルーティングの箇所で Depdends + get_db でセッションを取得します
取得したセッションは model_validate の context 引数にそのまま渡します
あとは FieldValidationInfo を使う際に存在チェックや型チェックを行ったあとで普通にクエリを発行すればデータベースにアクセスすることができます

最後に

pickle できるオブジェクトじゃないとダメかなと思ったいましたがそんなことはありませんでした

基本的には何でも渡せるのかなと思います

2023年9月12日火曜日

fastapi で pydantic V2 の FieldValidationInfo (context) を使う方法

fastapi で pydantic V2 の FieldValidationInfo (context) を使う方法

概要

fastapi の BaseModel を使ってリクエストモデルを定義している場合に field_validator は自動でコールされます
ただ context (FieldValidationInfo) を使った場合には context が必ず None になってしまうので fastapi では context が使えません

今回は fastapi で context を使ったバリデーションを実現する方法を考えます

環境

  • Python 3.11.3
  • fastapi 0.100.1
  • pydantic 2.1.1

サンプルコード

from fastapi import FastAPI
from pydantic import BaseModel, Field, FieldValidationInfo, field_validator

app = FastAPI()


class User(BaseModel):
    name: str = Field()
    age: int = Field()

    @field_validator("name")
    @classmethod
    def validate_name(cls, v: str, info: FieldValidationInfo):
        if info.context is not None:
            # context がある場合のバリデーションは無視する
            return v
        print("validate_name")
        if v not in ["hawksnowlog"]:
            raise ValueError()
        return v

    @field_validator("age")
    @classmethod
    def validate_age_with_context(cls, v: int, info: FieldValidationInfo):
        if info.context is None:
            # context がない場合のバリデーションは無視する
            return v
        if "age_limit" not in info.context or not isinstance(
            info.context["age_limit"], int
        ):
            # context の値をチェック、正しくセットされていない場合はエラー
            raise ValueError("info.context.age_limit が正しく設定されていません")
        print("validate_age_with_context")
        limit = int(info.context["age_limit"])
        if v > limit:
            raise ValueError(f"age が limit={limit} を超えています")
        return v


@app.post("/")
async def test_context(user: User):
    # ルーティングで一度 User のバリデーションをコールする
    user_dict = user.model_dump()
    # ここで context を付与して再度バリデーションをコールする
    user = User.model_validate(user_dict, context={"age_limit": 9})
    return user

ちょっと解説

ルーティングのリクエストモデルを受け取るところで一度各種バリデーションがコールされます
そしてルーティング内で model_validate することで再度バリデーションをコールします
そしてここで context を付与してコールします

バリデーション側では context が設定されているかどうかでバリデーションを実行するか判断しています
context が不要なバリデーションを二度コールしても一度通っているので必ず成功するため実行しても問題ないですが処理効率的に二回実行しても意味がないのでスルーするようにしています

最後に

これ以外に context ありで fastapi からバリデーションを実行する方法があるのだろうか

参考サイト

2023年8月31日木曜日

fastapiでリスト内にデータ構造が含まれるようなフォームデータを扱う方法

fastapiでリスト内にデータ構造が含まれるようなフォームデータを扱う方法

概要

リクエストでいうと以下のようなリクエストが来ることを想定して作ります

  • curl -XPOST -H "Action: Hoge" localhost:8000/submit/ -H "content-type: application/x-www-form-urlencoded" -d "data[0][lang]=ruby&data[0][framework]=sinatra&data[1][lang]=python&data[1][framework]=fastapi"

data はリストになっていてその中に lang, framework というデータ構造を持つオブジェクトが入っているイメージです

環境

  • Python 3.11.3
  • fastapi 0.100.1
  • pydantic 2.1.1

サンプルコード

from fastapi import Depends, FastAPI, Request
from pydantic import BaseModel, field_validator

app = FastAPI()


# リスト内のデータ構造を管理するクラス
class Profile(BaseModel):
    lang: str
    framework: str

    @field_validator("lang")
    @classmethod
    def validate_lang(cls, v: str):
        if v not in ["ruby", "python"]:
            raise ValueError()
        return v


# リクエストのリスト部分を管理するクラス
class RequestData(BaseModel):
    profile: list[Profile]


# 独自のフォームパーサ
async def parse_form_data(request: Request) -> RequestData:
    form_data = await request.form()
    print(form_data)
    request_data: list[Profile] = []
    # とりあえず 100 リストまで対応
    for i in range(100):
        # lang が含まれているか確認、0から始まっていないもしくは指定のデータ構造がない場合などは終了
        lang = form_data.get(f"data[{i}][lang]")
        if lang is None:
            break
        # framework が含まれているか確認、0から始まっていないもしくは指定のデータ構造がない場合などは終了
        framework = form_data.get(f"data[{i}][framework]")
        if framework is None:
            break
        # データ構造の作成
        profile = Profile(lang=str(lang), framework=str(framework))
        # リストへの追加
        request_data.append(profile)
    # Depends の返り値を指定
    return RequestData(profile=request_data)


@app.post("/submit/")
async def submit_form(form_data: RequestData = Depends(parse_form_data)):
    # Depends で parse_form_data を指定することでリクエストデータを独自でパースする
    return form_data

動作確認

  • curl -XPOST -H "Action: Hoge" localhost:8000/submit/ -H "content-type: application/x-www-form-urlencoded" -d "data[0][lang]=ruby&data[0][framework]=sinatra&data[1][lang]=python&data[1][framework]=fastapi"

は通ります

  • curl -XPOST -H "Action: Hoge" localhost:8000/submit/ -H "content-type: application/x-www-form-urlencoded" -d "data[0][lang]=ruby&data[0][framework]=sinatra&data[1][lang]=python3&data[1][framework]=fastapi"

はエラーになります

最後に

そもそもこういったリクエストは form ではなく json で受け取るべきだと思います
ファイルを送信したくどうしても form を使わざるを得ない場合は仕方ないかなと思います

2023年8月30日水曜日

fastapiでヘッダのアクション名に応じてコールするルーティングを変更する方法

fastapiでヘッダのアクション名に応じてコールするルーティングを変更する方法

概要

前回の続きです
ヘッダの情報に合わせて適切なルーティングをコールするようにしてみました

環境

  • Python 3.11.3
  • fastapi 0.100.1
  • pydantic 2.1.1

サンプルコード

from typing import Self

from fastapi import Depends, FastAPI, Form, Request
from pydantic import BaseModel, field_validator

app = FastAPI()


async def get_action(request: Request) -> str:
    action = request.headers.get("Action", "default")
    return action


class UserForm(BaseModel):
    username: str
    password: str

    @classmethod
    def as_form(cls, username: str = Form(...), password: str = Form(...)) -> Self:
        return cls(username=username, password=password)

    @field_validator("username")
    def validate_username(cls, v):
        if v != "hawksnowlog":
            raise ValueError("Invalid username.")
        return v


@app.get("/")
async def root(request: Request, action: str = Depends(get_action)):
    form_data = await request.form()
    if action == "Hoge":
        # FormData は immutableDict なのでアンパックできる
        return await hoge(action, UserForm(**form_data))  # type: ignore
    return action


@app.post("/hoge")
async def hoge(
    action: str = Depends(get_action), form_data: UserForm = Depends(UserForm.as_form)
):
    return action, form_data.username

動作確認

  • pipenv run uvicorn app:app --reload

POST で送信できる

  • curl -XPOST -H "Action: Hoge" localhost:8000/hoge -H "content-type: application/x-www-form-urlencoded" -d "username=hawksnowlog&password=pass"

GET からも送信できる

  • curl -XGET -H "Action: Hoge" localhost:8000 -H "content-type: application/x-www-form-urlencoded" -d "username=hawksnowlog&password=pass"

どちらも username のバリデーションが動作していることも確認できると思います

最後に

同一パスに対する異なる処理を書きたい場合には使えるテクニックかなと思います

2023年8月29日火曜日

fastapiでヘッダ情報とフォームデータを同時に扱う方法

fastapiでヘッダ情報とフォームデータを同時に扱う方法

概要

ヘッダは Depends で処理してフォームデータは BaseModel でリクエストとして処理します

環境

  • Python 3.11.3
  • fastapi 0.100.1
  • pydantic 2.1.1

サンプルコード

from typing import Self

from fastapi import Depends, FastAPI, Form, Request
from pydantic import BaseModel, field_validator

app = FastAPI()


async def get_action(request: Request) -> str:
    action = request.headers.get("Action", "default")
    return action


class UserForm(BaseModel):
    username: str
    password: str

    @classmethod
    def as_form(cls, username: str = Form(...), password: str = Form(...)) -> Self:
        return cls(username=username, password=password)

    @field_validator("username")
    def validate_username(cls, v):
        if v != "hawksnowlog":
            raise ValueError("Invalid username.")
        return v


@app.get("/")
async def root(
    action: str = Depends(get_action), form_data: UserForm = Depends(UserForm.as_form)
):
    return action, form_data

動作確認

  • pipenv run uvicorn app:app --reload
  • curl -H "Action: Hoge" localhost:8000 -XGET -H "content-type: application/x-www-form-urlencoded" -d "username=hawksnowlog&password=pass"

=> ["Hoge",{"username":"hawksnowlog","password":"pass"}]

最後に

更にアクションに応じて処理する場合は一度ヘッダを取得したあとに別のルーティングをコールするなどの処理が必要なのかもしれない

2023年8月21日月曜日

PydanticV2 で Xml なレスポンスを扱う方法を考える

PydanticV2 で Xml なレスポンスを扱う方法を考える

概要

過去 に Pydantic V1 で検討しましたが V2 でも検討してみます
ydantic-xml が V1 にしか対応していないので pydantic-xml を使わない方針で作成してみます

環境

  • Python 3.11.3
  • fastapi 0.100.1
  • pypdantic 2.1.1

方針

  • Xml レスポンスは fastapi のレスポンスクラスを継承して作成
  • Xml 用のモデル定義は pydantic の BaseModel を継承する
    • レスポンスを Xml に変換するためのシリアライザを持たせる
    • Xml フィールドのキーはパスカルケースに変換する

サンプルコード

import textwrap
from typing import LiteralString, Union

import dicttoxml
from fastapi import FastAPI, Response
from pydantic import BaseModel, ConfigDict, Field
from pydantic.alias_generators import to_pascal

app = FastAPI()


class User(BaseModel):
    # alias_generator を指定することで Xml のキー情報をパスカルケースにする
    model_config = ConfigDict(populate_by_name=True, alias_generator=to_pascal)

    name: str = Field()
    age: int = Field()

    # model_dump (dict) を xml (str) に変換する
    def to_xml(self) -> LiteralString | bytes:
        return dicttoxml.dicttoxml(
            self.model_dump(by_alias=True),
            custom_root=to_pascal("some_custom_root"),
            attr_type=False,
        )


# fastapi で Xml レスポンスを生成するためのクラス
class XmlResponse(Response):
    def __init__(self, content: Union[str, bytes], status_code: int = 200):
        super().__init__(content, status_code=status_code, media_type="application/xml")


@app.get(
    "/",
    response_class=XmlResponse,
    response_model=User,
    # openapi-doc はここで生成する
    responses={
        200: {
            "description": "Success",
            "content": {
                "application/xml": {
                    "schema": {"$ref": "#/components/schemas/User"},
                    "example": textwrap.dedent(
                        """
                        <?xml version="1.0" encoding="UTF-8"?>
                        <User>
                          <Name>hawk</Name>
                          <Age>20</Age>
                        </User>
                    """
                    )[1:-1],
                },
                "application/json": None,
            },
        },
    },
)
def xml():
    result = User(name="hawksnowlog", age=10)
    return XmlResponse(content=result.to_xml())

動作確認

  • curl localhost:8000
<?xml version="1.0" encoding="UTF-8" ?><SomeCustomRoot><Name>hawksnowlog</Name><Age>10</Age></SomeCustomRoot>

localhost:8000/docs

最後に

pydantic V2 で Xml レスポンスを表現する方法を検討してみました
to_xml などはシリアライザを指定して自動で呼び出してくれるようになるともっといい感じなるかなと思います

参考サイト

2023年8月10日木曜日

pydantic の BaseModel でデータベースに接続してバリデーションを行うケースを考える

pydantic の BaseModel でデータベースに接続してバリデーションを行うケースを考える

概要

結論からすると pydantic の BaseModel 内で DB を参照するのは好ましくないので context を使いましょう

コードは前回のものを流用します

環境

  • Python 3.11.3
  • fastapi 0.100.1
  • pypdantic 2.1.1

サンプルコード

from enum import Enum

from fastapi import Depends, FastAPI, Query, Request
from fastapi.datastructures import QueryParams
from pydantic import BaseModel, Field, FieldValidationInfo, field_validator
from sqlalchemy import Column, Integer, String, create_engine
from sqlalchemy.orm import Session, declarative_base, sessionmaker

app = FastAPI()


engine = create_engine("mysql+pymysql://root@localhost/test?charset=utf8mb4")
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
Base = declarative_base()


class User(Base):
    __tablename__ = "user"

    id = Column(Integer, primary_key=True)
    name = Column(String(50))
    age = Column(Integer)


def get_db():
    db: Session = SessionLocal()
    try:
        yield db
        db.commit()
    except Exception:
        db.rollback()
    finally:
        db.close()


# 各種アクション名を管理する列挙型のクラス
class ActionName(Enum):
    CREATE_INSTANCE = "CreateInstance"


# CreateInstance(例) という名前のアクションのバリデーションを管理するモデル
class CreateInstance(BaseModel):
    instance_name: str = Field(alias="InstanceName")

    @field_validator("instance_name")
    @classmethod
    def validate_instance_name(cls, v: str, info: FieldValidationInfo) -> str:
        # context は FieldValidationInfo に格納されているので使用する場合は引数に追加する
        context = info.context
        # データベースから取得した値が取得できていることが確認できる
        print(context)
        if v != "ins01":
            raise ValueError()
        return v


# 起点となる Action 名のバリデーションを管理するモデル
class Action(BaseModel):
    action: str = Field(Query(alias="Action"))

    @field_validator("action")
    @classmethod
    def validate_action(cls, v: str) -> str:
        # ここでは許可するバリデーション名など基本的なバリデーションのみを行う
        if v not in [ActionName.CREATE_INSTANCE.value]:
            raise ValueError()
        return v

    # アクション名に応じて各種バリデーションをコールする
    def validate(self, query_params: QueryParams, context: dict):
        if self.action == ActionName.CREATE_INSTANCE.value:
            CreateInstance.model_validate(query_params, context=context)


@app.get("/")
async def root(
    request: Request, action: Action = Depends(), db: Session = Depends(get_db)
):
    # データベースの値はここで取得してバリデーション対応の値を context として渡す
    users = db.query(User).all()
    names = [user.name for user in users]
    # ルーティングの引数で Request オブジェクトを参照できるのでこれを使って各種アクションのバリデーションをコールする
    action.validate(request.query_params, context={"names": names})
    return action

解説

ポイントは context です
データベースへの参照はルーティング内でのみ行います
そして必要な値をデータベース側で渡すようにしましょう

このケースの場合は model_validate を使うことが前提となっているのでルーティングの引数でモデルを指定して自動でバリデーションしているような場合は少し改修が必要になります

最後に

一応データベースを参照することもできますがその場合はグローバルにアクセスできる関数やデータベースのコンテキストを考える必要があるので少し面倒なのとコードの可読性が下がるかなと思います

参考サイト