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 できるオブジェクトじゃないとダメかなと思ったいましたがそんなことはありませんでした

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