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

2024年1月23日火曜日

Pydantic の CustomTypes を使ってみる

Pydantic の CustomTypes を使ってみる

概要

同じような記述のタイプヒントを複数のモデルで使い回す場合には CustomType で独自の型を定義して使いますと便利です

環境

  • macOS 11.7.10
  • Python 3.11.6
  • pydantic 2.5.3

サンプルコード

from typing import Annotated

from pydantic import BaseModel, Field, TypeAdapter

Name = Annotated[str, Field(min_length=5)]


class Animal(BaseModel):
    name: Name


class User(BaseModel):
    name: Name
    age : int


AdaptedName = TypeAdapter(Name)

if __name__ == "__main__":
    user_data = {
        "name": "hawksnowlog",
        "age": 10
    }
    user = User(**user_data)
    print(user.name)
    animal_data = {
        "name": "snowlog",
        # "name": "hawk", # error
    }
    animal = Animal(**animal_data)
    print(animal.name)
    # TypeAdapter を使えばクラスにしないでもカスタムタイプを検証できる
    name= AdaptedName.validate_python("hawksnowlog")
    # name = AdaptedName.validate_python("hawk")  # error
    print(name)

最後に

継承してもいいですがわざわざ継承するほどでもない共通のフィールドを扱いたい場合には便利かなと思います
型を一箇所で管理できれば一箇所修正するだけで複数のモデルに適用することができます

参考サイト

2024年1月22日月曜日

Pydantic の TypeAdapter を使ってみる

Pydantic の TypeAdapter を使ってみる

概要

使ってみました
簡単に言うと独自のクラスを扱うコレクションクラスを新しいクラスとして定義することができる機能です

環境

  • macOS 11.7.10
  • Python 3.11.6
  • pydantic 2.5.3

サンプルコード

from typing import List

from pydantic import BaseModel, TypeAdapter


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


UserList = TypeAdapter(List[User])


if __name__ == "__main__":
    # 単一データの場合 (TypeAdapterなし)
    data = {
        "name": "hawk",
        "age": 10
    }
    user = User(**data)
    print(user.name)
    # 複数データの場合 (TypeAdapter あり)
    data_list = [
        {
            "name": "snowlog",
            "age": 20,
        },
        {
            "name": "hawksnowlog",
            "age": 30,
        }
    ]
    user_list = UserList.validate_python(data_list)
    for user in user_list:
        print(user.name)
    # 複数データの場合 (TypeAdapter なし)
    user_list: list[User]
    for data in data_list:
        user = User(**data)
        user_list.append(user)
    for user in user_list:
        print(user.name)

解説

サンプルコードを見るのが一番理解できるかなと思います
TypeAdapter を使った場合はシリアライズが 1 行で完了しますが使っていない場合はシリアライズをわざわざループで回して行う必要があるので記述が面倒になります

これが TypeAdapter の一番の機能かなと思います

最後に

データベースのモデルなども使うことができるのか気になりました
CustomTypesという機能もありこれと組み合わせると更に便利に使うことができそうです

参考サイト

2024年1月12日金曜日

flask-pydantic が v2 に対応していたので試してみた

flask-pydantic が v2 に対応していたので試してみた

概要

過去に使ったときは pydantic v2 に対応していませんでした
flask-pydantic 0.12.0 で対応したようなので使ってみました

環境

  • macOS 11.7.10
  • Python 3.11.6
    • flask 3.0.0
    • pydantic 2.5.3
    • flask-pydantic 0.12.0

サンプルコード

from typing import Optional

from flask import Flask, jsonify
from flask_pydantic import ValidationError, validate
from pydantic import BaseModel, field_validator

app = Flask(__name__)
app.config["FLASK_PYDANTIC_VALIDATION_ERROR_RAISE"] = True


class ResponseModel(BaseModel):
    id: int
    age: int
    name: str
    nickname: Optional[str]


class RequestBodyModel(BaseModel):
    name: str
    nickname: Optional[str]

    @field_validator("name")
    def check_name(cls, v, values, **kwargs):
        if v != "hawksnowlog":
            raise ValueError()
        return v


@app.route("/", methods=["POST"])
@validate()
def post(body: RequestBodyModel):
    name = body.name
    nickname = body.nickname
    return ResponseModel(name=name, nickname=nickname, id=0, age=1000)


@app.errorhandler(ValidationError)
def handle_bad_request(_):
    return jsonify({"error": "validation_error"}), 400


app.register_error_handler(400, handle_bad_request)

所感

開発はまだまだ途中という感じかなと思います
エラーハンドリングの部分が微妙で今回 FLASK_PYDANTIC_VALIDATION_ERROR_RAISE というフラグを True にしています
これがない状態で ValueError を投げると json にシリアライズする部分でエラーになり json のエラー構文が返ってきません
なので flask-pydantic 側でハンドリングはせずにアプリ側でハンドリングを実装してあげる必要があります
これに関してはそのうち直る可能性はありそうですがそれ以外でもまだバグがありそうなので v2 対応はまだ様子見なのかもしれません

参考サイト

2024年1月11日木曜日

Flask で pydantic を使う方法を考える

Flask で pydantic を使う方法を考える

概要

過去にflask-pydanticを使った方法を紹介しました
しかし flask-pydantic は V2 に対応していません
今回は flask-pydantic なしで Flask + pydantic を連携する方法を考えます

環境

  • macOS 11.7.10
  • Python 3.11.6
    • flask 3.0.0
    • pydantic 2.5.3

案1: ルーティング内の先頭でモデルを使用する

from flask import Flask, jsonify, request
from pydantic import BaseModel, field_validator

app = Flask(__name__)


class User(BaseModel):
    name: str

    @field_validator("name")
    @classmethod
    def validate_name(cls, v):
        if v != "hawk":
            raise ValueError()
        return v


@app.route("/", methods=["POST"])
def test():
    data = request.json
    if data is None:
        raise ValueError()
    user = User(**data)
    return jsonify(user.model_dump())

メリット

  • 簡単、明瞭、特に何も考えず先頭で変換するだけなので楽

デメリット

  • 変換する部分の実装を忘れそう
  • request.json から変換しているが他にも query や path などをモデルに変換する部分の考慮しなければならない

案2: デコレータを使い引数で受け取る

from flask import Flask, jsonify, request
from pydantic import BaseModel, field_validator

app = Flask(__name__)


class User(BaseModel):
    name: str

    @field_validator("name")
    @classmethod
    def validate_name(cls, v):
        if v != "hawk":
            raise ValueError()
        return v


def validate(func):
    def wrapper(*args, **kwargs):
        data = request.json
        if data is None:
            raise ValueError()
        user = User(**data)
        return func(user, *args, **kwargs)

    return wrapper


@app.route("/", methods=["POST"])
@validate
def test(user: User):
    return jsonify(user.model_dump())

メリット

  • ルーティング側での変換が不要になるのでコードがきれいになる
  • 引数で受け取れてかつ検証済みのオブジェクトなので安全に扱える

デメリット

  • デコレータが肥大化しないように気をつける必要がある
    • モデルごとにデコレータを追加する必要がないようにする必要がある
  • request.json から変換しているが他にも query や path などをモデルに変換する部分の考慮しなければならない

最後に

頑張れば何とかなりそうだけどコードが複雑化しないように工夫する必要がありそうです
そのあたりを吸収してくれてくれているのが flask-pydantic なので最悪ライブラリを自分でメンテするのもありなのかもしれません

2023年11月18日土曜日

PydanticV2 で dict から dataclass を生成する方法

PydanticV2 で dict から dataclass を生成する方法

概要

dict -> obj のデシリアライズを Pydantic V2 を使ってやる方法を紹介します
pydantic の dataclass を使います

過去に JSONWizard を使った方法も紹介しています

環境

  • Python 3.10.2
  • pydantic 2.5.1

サンプルコード

from dataclasses import asdict

from pydantic.dataclasses import dataclass


@dataclass
class Profile:
    lang: str
    framework: str


@dataclass
class User:
    id: int
    name: str
    profile: Profile


if __name__ == "__main__":
    # dict -> obj
    user_dict = {
        "id": 1,
        "name": "hawksnowlog",
        "profile": {
            "lang": "python",
            "framework": "flask",
        },
    }
    user = User(**user_dict)
    print(user.id)
    print(user.name)
    print(user.profile.lang)
    print(user.profile.framework)
    # obj -> dict
    print(asdict(user))

ちょっと解説

  • ネストした構造でもいけます
  • Optional など組み合わせれば任意の値もとれます
  • BaseModel ではこの方法は使えません
  • 各種フィールドは Field として定義することはできません

最後に

BaseModel として使えないのが少し残念ですが単純な dict との変換ならこれが一番簡単な方法かなと思います

ただサードパーティを使うので管理などは少し面倒です
本当に簡単な変換なのであれば純正の json モジュールだけでも何とかなります

参考サイト

2023年11月16日木曜日

Pydantic で Union を使って model_validate する場合は必ず discriminator を使うべきである

Pydantic で Union を使って model_validate する場合は必ず discriminator を使うべきである

概要

model_validate を使用すると pydantic が自動で型を推論します
更に Union を使っている場合は Union に指定している複数のクラスのどれにするかを自動的に推論します
同じフィールドを持つ場合には意図しないクラスに割り当てられることもあります
そういったことを避けるために discriminator という機能を使いましょう

環境

  • Python 3.10.2
  • pydantic 2.5.1

サンプルコード

from typing import Annotated, Literal, Union

from pydantic import BaseModel, Field, field_validator


class Cat(BaseModel):
    # model_validate 時の型推論用フィールド
    animal_type: Literal["cat"] = "cat"
    value: str = Field()

    @field_validator("value")
    @classmethod
    def validate_value(cls, v: str):
        if v != "neko":
            raise ValueError()
        return v


class Dog(BaseModel):
    # model_validate 時の型推論用フィールド
    animal_type: Literal["dog"] = "dog"
    value: str = Field()

    @field_validator("value")
    @classmethod
    def validate_value(cls, v: str):
        if v != "inu":
            raise ValueError()
        return v


Animal = Annotated[
    Union[Cat, Dog],
    Field(discriminator="animal_type"),
]


class Trainer(BaseModel):
    # model_validate 時の型推論用フィールド
    animal: Animal = Field()


if __name__ == "__main__":
    # 成功
    cat = Cat(value="neko")
    dog = Dog(value="inu")
    trainer1 = Trainer(animal=cat)
    trainer2 = Trainer(animal=cat)
    Trainer.model_validate(trainer1.model_dump())
    Trainer.model_validate(trainer2.model_dump())
    # 失敗
    cat = Cat(value="inu")
    trainer1 = Trainer(animal=cat)
    Trainer.model_validate(trainer1.model_dump())

ちょっと解説

一見当たり前のことをしていますが同一のフィールド名をもつクラスを作成しそのクラスに対して model_validate を実行するとフィールドの情報からクラスを推測するため間違ったクラスになる可能性があります

model_dump はクラスの json 情報を返却します
また model_validate で渡す json 情報はすべてのフィールドがプリミティブな値でなければいけないので model_dump される json はすべて文字列だけの情報になります

その状態で discriminator 情報なしで model_validate するとクラスを特定する情報がない状態になってしまいます
なのでクラスを特定する情報として今回 animal_type フィールドを追加し必ずクラスごとに一意になるように値を設定することでプリミティブな値のみを持つ model_dump の情報からでも適切なクラスに変換し model_validate することができるようになります

最後に

Pydantic で Union を使う場合には必ず discriminator を使うようにしましょう

参考サイト

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 を使わざるを得ない場合は仕方ないかなと思います