问题实际上是如何更新SQLAlchemy声明性模型,以便它运行验证器。就我而言,使用像User.name = name这样的setters并不是一个选择。

以下是我的意思的可运行示例

from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker

from sqlalchemy import Column, String, Integer
from sqlalchemy.orm import validates
from sqlalchemy.ext.declarative import declarative_base

some_engine = create_engine('sqlite://')
Session = sessionmaker(bind=some_engine)
session = Session()


Base = declarative_base()


class User(Base):
    __tablename__ = 'user'

    id = Column(Integer, primary_key=True)
    name = Column(String, nullable=False)

    @validates('name')
    def validate_name(self, key, value):
        if value != 'asd':
            raise ValueError('not asd')
        return value

Base.metadata.create_all(bind=some_engine)

user = User(id=1, name='qwe')
# >>> ValueError: not asd
user = User(id=1, name='asd')

session.add(user)
session.commit()

session.query(User).filter(User.id=1).update({'name': 'qwe'})
session.query(User).filter(User.id==1)[0].name
# >>> 'qwe'

最佳答案

您可以在模型中添加一个mixin,以提供一种非常简单的更新方法,该方法仅使用setattr()设置实例的属性。

class UpdateMixin:
    """
    Add a simple update() method to instances that accepts
    a dictionary of updates.
    """
    def update(self, values):
        for k, v in values.items():
            setattr(self, k, v)


然后,将用户类别定义为

class User(UpdateMixin, Base):
    ...


要从给定的字典更新单个实例,您可以运行

session.query(User).get(1).update({ 'name': 'qwe' })
# or since you have the user instance from before
user.update({ 'name': 'qwe' })


注意Query.get()的使用。如果没有具有给定id的用户,它将返回None,并尝试在其上调用方法更新。另一个警告是,如果在引发任何异常时不回滚,则由于字典没有排序,因此无法预测发生了什么更新(已添加到会话中)(如果有的话)。因此,请始终回滚任何错误。

我还建议实际命名方法updateSelf或类似的名称,以防止将其与Query.update()混淆的风险。

10-08 05:01