SQLalchemy 2.0 listen events and handle column changing

Question

Всем привет, у меня есть модель таблица:

class User(Base):
    username: Mapped[str] = mapped_column(String(1_000), unique=True)
    email: Mapped[str] = mapped_column(unique=True, nullable=True)
    password: Mapped[str]
    is_active: Mapped[bool] = mapped_column(default=True)
    is_verified: Mapped[bool] = mapped_column(default=False)

Мне нужно сделать так, ЧТОБЫ при изменении поля email, автоматически is_verified поле менялось на False

Answer 1

Можно воспользоваться декоратором validates для поля email и при обновлении поля, is_verified будет выставлять в False

Создаем модель, тестовую базу и подключение:

from sqlalchemy import String, create_engine
from sqlalchemy.orm import (
    sessionmaker, validates, Mapped, mapped_column, declarative_base
)

Base = declarative_base()

class User(Base):
    __tablename__ = 'users'

    id: Mapped[int] = mapped_column(primary_key=True, autoincrement=True)
    username: Mapped[str] = mapped_column(String(1_000), unique=True)
    email: Mapped[str] = mapped_column(unique=True, nullable=True)
    password: Mapped[str]
    is_active: Mapped[bool] = mapped_column(default=True)
    is_verified: Mapped[bool] = mapped_column(default=False)

    @validates("email")
    def validate_email(self, _: str, value: str) -> str:
        if value != self.email:
            self.is_verified = False
        return value


engine = create_engine('sqlite:///example.db')
Base.metadata.create_all(engine)

Session = sessionmaker(bind=engine)
session = Session()

Создаем тестового user-a:

new_user = User(
    username="John Doe",
    email="[email protected]",
    password="qwerty",
    is_verified=True
)

session.add(new_user)
session.commit()

Обновляем поле email (необходимо закомментировать строки создания user-a)

user_to_update = session.query(User).filter_by(username="John Doe").first()
print("User before update email:")
print(f"Email = {user_to_update.email}, is_verified = {user_to_update.is_verified}")
user_to_update.email = "[email protected]"
print()
print("User after update email")
print(f"Email = {user_to_update.email}, is_verified = {user_to_update.is_verified}")

Вывод:

User before update email:
Email = [email protected], is_verified = True

User after update email
Email = [email protected], is_verified = False

Update

Также можно настроить событие - events которое срабатывает на изменение поля email

from sqlalchemy import event
from sqlalchemy.orm.mapper import Mapper
from sqlalchemy.engine.base import Connection

@event.listens_for(User, 'before_update')
def before_update(mapper: Mapper, connection: Connection, target: User) -> None:
    state = target.__dict__.get('_sa_instance_state')
    history = state.attrs.email.history
    if history.has_changes():
        target.is_verified = False

БЛОГ НА HUSL

SQLalchemy 2.0 listen events and handle column changing

Ответы (1 шт):