主页 / 问题 / 1587821
Accepted
Kamran Pulatov
Asked:2024-07-21 05:26:41 +0000 UTC2024-07-21 05:26:41 +0000 UTC

SQLalchemy 2.0 监听事件并处理列更改

  • 772

大家好，我有一个模型表：

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)

我需要确保当电子邮件字段更改时，is_verified 字段自动更改为 False

python

1 个回答

  1. Best Answer

    您可以对字段使用validatesemail装饰器，并且在更新字段时，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()
    • 创建测试用户：
    new_user = User(
    username="John Doe",
    email="[email protected]",
    password="qwerty",
    is_verified=True
)

session.add(new_user)
session.commit()
    • 更新字段email（需要注释掉用户创建行）
    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

    更新

    • 您还可以设置事件 -当字段更改时触发的事件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
    • 2

相关问题