まとめ(結論だけ知りたい人用)

単一の値に対してのvalidation
@￰validatesデコレータを利用する
複数の値に対してのvalidation
- eventを使う(明示的flush使用)
  - バリデーションを行うタイミングが明確
  - 無駄なクエリ発行を行いやすい
- eventを使う(autoflush使用)
  - バリデーションを行うタイミングが明確ではない
  - 無駄なクエリ発行を抑制できる
- session.add()にvalidationを埋め込む
  - バリデーションを行うタイミングが明確
  - 無駄なクエリ発行を抑制できる
  - メタプログラミングチックなので注意が必要

具体的なコードを確認したい方は記事本文をどうぞ

概要

SQLAlchemyのmodelオブジェクトのattributeに対して、
「必ず0以上である」などの制約を付けたい時があります。
また、2つ以上の値の関係に対しても制約を付けたい場合があります。

今回は「単一の値に対してのvalidation」と「複数の値に対してのvalidation」の
2つのvalidationについて見ていきます。

単一の値に対してのvalidation

SQLAlchemyには@￰validatesというデコレーターが用意されています。
このデコレーターを使うことで単一の値に対してのvalidationを設定することが出来るようになります。
以下に使用例を示します。

from sqlalchemy.orm import declarative_base
from sqlalchemy.orm import validates
from sqlalchemy.schema import Column
from sqlalchemy.types import VARCHAR


class User(Base):
    __tablename__ = "user"
    user_id = Column(VARCHAR(26), name="id", primary_key=True)
    name = Column(VARCHAR(255), name="name", nullable=False)
    email = Column(VARCHAR(255), name="email")

    @validates("email")
    def validate_email(self, key, value):
        if "@" not in value:
            raise ValueError("Invalid email address")

        return value # valueを返す必要があります。

    @validates("name", "user_id") # 複数のフィールドを指定できます。
    def validate_name(self, key, value):
        if value is None:
            raise ValueError("Value must not be None")

        return value


user = User(user_id="1", name=None, email="john.example.com")

このコードを実行するとValueError: Value must not be Noneがエラーとして出力されます。

複数の値に対してのvalidation

@￰validatesを用いたバリデーションは値の代入が行われるたびに実行されるため、
複数の値の関係をバリデーションするという用途にはあまり向いていません。

複数の値をバリデーションする場合、以下の複数の方法が考えられます。

eventを使う(明示的flush使用)

以下のコードを追加すると使用可能になります。

@event.listens_for(User, "before_update")
@event.listens_for(User, "before_insert")
def validate_before_update(mapper, connection, target):
    if target["user_id"] == target["name"]:
        raise ValueError

このイベントリスナーはflush()を実行した際に呼び出されます。

SessionClass = sessionmaker(engine, autoflush=True)
session = SessionClass()

user = User(user_id="1", name="Johnz", email="john@example.com")
session.add(user)
user2 = User(user_id="2", name="Johnz", email="john@example.com")
session.add(user2)
session.flush() # ここでvalidationが呼び出される。

user3 = User(user_id="3", name="Johnz", email="john@example.com")
session.add(user3)
session.flush() # ここでvalidationが呼び出される。

session.query(User).all()

session.commit()

明示的にflush()を使用する場合は、バリデーションが実行されるタイミングが
分かるため、バリデーション後のエラーに対するcatchを書きやすくなります。

ただし、flush()の実行をライブラリに委譲しないため、無駄なクエリが発行される可能性があります。

eventを使う(autoflush使用)

イベントリスナーのコードは上のeventを使う(明示的flush使用)と同様です。

@event.listens_for(User, "before_update")
@event.listens_for(User, "before_insert")
def validate_before_update(mapper, connection, target):
    if target["user_id"] == target["name"]:
        raise ValueError

しかし、実行タイミングが異なります。

SessionClass = sessionmaker(engine, autoflush=True)
session = SessionClass()

user = User(user_id="1", name="Johnz", email="john@example.com")
session.add(user)
user2 = User(user_id="2", name="Johnz", email="john@example.com")
session.add(user2)
user3 = User(user_id="3", name="Johnz", email="john@example.com")
session.add(user3)

session.query(User).all() # ここでvalidationが呼び出される。

session.commit()

insert/updateクエリが実行される直前にイベントリスナーが呼び出されるため、
autoflushを有効にしている場合は予期せぬ場所でバリデーションが走る可能性があります。

session.add()にvalidationを埋め込む

insertとupdateを実行する前にsession.add()を明示的に実行する必要があるため、
session.add()にvalidatorを埋め込むことでバリデーションを行うという手法です。
この方法はautoflush=Trueであっても実行タイミングが明確になります。
ただし、メタプログラミングチックなので注意が必要。

コードは以下の通りです。

SessionClass = sessionmaker(engine, autoflush=True)
session = SessionClass()

original_add = session.add

def new_add(*args, **kwargs):
    # この部分にバリデーションを書く
    return original_add(*args, **kwargs)

session.add = new_add

後はいつも通りにsession.add()をupdateやinsertを行う部分で実行するだけでOKです。

SessionClass = sessionmaker(engine, autoflush=True)
session = SessionClass()

user = User(user_id="1", name="Johnz", email="john@example.com")
session.add(user)　# ここでvalidationが呼び出される。
user2 = User(user_id="2", name="Johnz", email="john@example.com")
session.add(user2)　# ここでvalidationが呼び出される。
user3 = User(user_id="3", name="Johnz", email="john@example.com")
session.add(user3)　# ここでvalidationが呼び出される。

session.query(User).all()

session.commit()