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月27日水曜日

SQLAlchemy でクエリキャッシュを使う方法 (dogpile.cache編)

SQLAlchemy でクエリキャッシュを使う方法 (dogpile.cache編)

概要

SQLAlchemy は以前 Beker cache という仕組みを使ってクエリキャッシュしていました
しかし SQLAlchemy 2.0 の現在ではその方法は廃止され dogpile.cache という汎用キャッシュの仕組みを使ってクエリキャッシュするのが王道になっています

前回 dogpile.cache の簡単な使い方を紹介しました
今回は SQLAlchemy と組み合わせて使ってみます

環境

  • macOS 13.5.2
  • Python 3.11.5
    • dogpile.cache 1.2.2
    • SQLAlchemy 2.0.21

サンプルコード

from dogpile.cache import make_region
from sqlalchemy import Column, Integer, String, create_engine
from sqlalchemy.orm import Session, declarative_base, sessionmaker

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)


# Dogpileキャッシュリージョンを作成
cache_region = make_region().configure(
    "dogpile.cache.memory",  # メモリキャッシュを使用しますが、他のバックエンドも利用可能です
    expiration_time=3600,  # キャッシュの有効期限を設定(秒単位)
)


# SQLAlchemyセッションをキャッシュ対象にする
@cache_region.cache_on_arguments()
def get_user_by_id(user_id):
    db: Session = SessionLocal()
    print("Querying database for user:", user_id)
    result = db.query(User).filter_by(id=user_id).first()
    db.close()
    return result


# ユーザーを取得し、キャッシュを利用
cached_user = get_user_by_id(1)
print("User 1:", cached_user.name)

# 同じユーザーを再度取得し、キャッシュを利用
cached_user = get_user_by_id(1)
print("User 1 (cached):", cached_user.name)

ちょっと解説

dogpile.cache は同一パラメータを使った関数呼び出しの場合はキャッシュを参照するようになります
なのでその仕組を使ってクエリを投げる関数を作成しその関数に対して cache_on_arguments アノテーションを付与することでキャッシュ化することができます

2回関数をコールしていますが2回目はキャッシュを参照しているため「Querying database for user: 1」が表示されないのが確認できます

こんな感じで既存のクエリを発行する関数を簡単にキャッシュ化することができます

最後に

dogpile.cache を使って SQLAlchemy のクエリキャッシュをする方法を紹介しました
アノテーション付与だけで使えるのでかなり簡単です

dogpile.cache には他にもキャッシュ戦略やキャッシュを特定するキーの生成方法などいろいろとカスタマイズすることができるので興味があれば参考サイトにある公式ドキュメントを見ることをおすすめします

参考サイト

2023年9月26日火曜日

dogpile.cache 超入門

dogpile.cache 超入門

概要

dogpile.cache は SQLAlchemy のクエリキャッシュのために作られたライブラリでその名の通りクエリキャッシュに使います
しかし dogpile.cache 自体は単なる関数キャッシュとしても使えるので今回は SQLAlchemy なしで簡単に動かせるサンプルコードを紹介します

環境

  • macOS 13.5.2
  • Python 3.11.5
  • dogpile.cache 1.2.2

サンプルコード

from dogpile.cache import make_region

# Dogpileキャッシュリージョンを作成
cache_region = make_region().configure(
    "dogpile.cache.memory",  # メモリキャッシュを使用しますが、他のバックエンドも利用可能です
    expiration_time=3600,  # キャッシュの有効期限を設定(秒単位)
)


# キャッシュする関数を定義
@cache_region.cache_on_arguments()
def expensive_function(x, y):
    print("Function executed!")
    return x + y


# 関数を呼び出してキャッシュを利用
result1 = expensive_function(2, 3)
print("Result 1:", result1)  # この行が実行されると、関数が実際に呼び出されます

# 同じ引数で関数を再度呼び出し
result2 = expensive_function(2, 3)
print("Result 2:", result2)  # この行はキャッシュから結果を取得します

# 別の引数で関数を呼び出し
result3 = expensive_function(4, 5)
print("Result 3:", result3)  # この行が実行されると、新しい結果がキャッシュされます

ちょっと解説

result1 と result3 は関数を実行しています
result2 は関数を実行せずキャッシュ (メモリ) から結果を取得しているため「Function executed!」が表示されないのが確認できます

またdogpile.cache.memoryでキャッシュ先のバックエンドを指定することもできます
今回はメモリを使っていますが他にも redis や memcached などいろいろなバックエンドがキャッシュ先として使えます

最後に

アノテーションを一つ付けるだけで簡単に関数の結果をキャッシュすることができます
大量にコールされる関数でかつ結果が非同期でも問題ない場合は使えるかなと思います
結果が非同期というかどんな結果でも問題ないような関数の場合には使えると思います

次回は SQLAlchemy と組み合わせて使ってみたいと思います

参考サイト

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月21日木曜日

Python の validators でサポートしていないプロトコルの URL をバリデーションする方法

Python の validators でサポートしていないプロトコルの URL をバリデーションする方法

概要

無理やり対応しているプロトコルに変換して validators.url に投げます

環境

  • macOS 13.5.2
  • Python 3.11.3
    • validators 0.22.0

サンプルコード

import validators


def replace_protocol(url):
    if url.startswith("mqtt"):
        return mqtt_url.replace("mqtt", "http")
    elif url.startswith("mqtts"):
        return mqtt_url.replace("mqtts", "https")
    else:
        raise ValueError()
    

mqtt_url = "mqtts://localhost:1884"
http_url = replace_protocol(mqtt_url)

print(validators.url(http_url, simple_host=True, may_have_port=True))

最後に

苦肉の策としてこんな感じで対応していないプロトコルでも URL バリデーションできます
本当は validators の対応を待つのがいいかなと思います

2023年9月20日水曜日

Pythonのvalidatorsを使ってみる

Pythonのvalidatorsを使ってみる

概要

validators ライブラリは複雑な正規表現を使わずにいろいろなフォーマットのバリデーションが行える便利ライブラリです

今回は URL のバリデーションだけですが使って見ました

環境

  • macOS 13.5.2
  • Python 3.11.3
    • validators 0.22.0

インストール

  • pipenv install validators

サンプルコード

成功している場合は True で失敗している場合は False ではなく ValidationError のオブジェクトが返ってきます

まだまだ開発段階ですべての URL スキームに対応しているわけではないので注意してください

import validators

# OK
print(validators.url("https://kakakikikeke.com"))
print(validators.url("http://localhost.dev"))
print(validators.url("http://localhost", simple_host=True))
print(validators.url("http://localhost:8000", simple_host=True, may_have_port=True))

# Error
print(validators.url("localhost"))  # schema がない
print(validators.url("localhost:8000", simple_host=True, may_have_port=True))  # schema がない
print(validators.url("http://localhost:8000"))  # ポートが指定されている (my_have_port=True がない場合
print(validators.url("mqtt://localhost:1883", simple_host=True, may_have_port=True))  # mqtt スキームはサポートしていない https://github.com/python-validators/validators/blob/master/src/validators/url.py#L41

最後に

他にもいろいろなバリデーションに対応しています
メールフォーマットをチェックした IP アドレスフォーマットのチェック、ホスト名チェックなども行えます

正規表現を愚直に書くよりかはこちらに寄せるほうがコードとしては読みやすくなるかなと思います

そのうち uri というメソッドができるのでそっちでいろいろなスキームに対応できるかなと思います 参考

参考サイト