跳过 onupdate 定义的默认值,用于 SQLAlchemy 中的特定更新查询
Skip the default on onupdate defined, for specific update queries in SQLAlchemy
如果我有一个 post 的列表,其中 created
和 updated
日期带有默认附加 onupdate
回调。
有时我需要为 inappropriate reports
或类似操作标记 post。我不想修改 created
和 updated
日期。
如何在执行 update
操作时跳过定义的 onupdate
?
SQLAlchemy 将应用默认 when
no value was provided to the INSERT or UPDATE statement for that column
然而,明显的解决方法 - 显式将列设置为其当前值 - 将不起作用,因为会话 checks 值是否已实际更改,如果没有,则不会传递值。这里有两个可能的解决方案,假设 SQLAlchemy 1.4+ 和这个模型:
class Post(db.Model):
flag = db.Column(db.Boolean, default=False)
last_updated = db.Column(db.DateTime, default=some_func, onupdate=some_func)
使用事件侦听器
添加一个 before update listener that detects when the flag column is being modified, and mark 修改后的时间戳列,即使其值未更改。这将使 SQLAlchemy 将当前值添加到更新中,因此不会调用 onupdate 函数。
import sqlalchemy as sa
...
@sa.event.listens_for(Post, 'before_update')
def receive_before_update(mapper, connection, target):
insp = sa.inspect(target)
flag_changed, _, _ = insp.attrs.flag.history
if flag_changed:
orm.attributes.flag_modified(target, 'last_updated')
使用 SQLAlchemy 核心而不是 ORM
SQLAlchemy 核心不需要会话,因此可以将当前时间戳值传递给更新以避免触发 onupdate 函数。 ORM 不会意识到以这种方式所做的任何更改,因此如果在会话的上下文中完成,受影响的对象应该被刷新或过期。这是一个“快速而肮脏”的解决方案,但如果标记发生在正常应用程序流程之外,可能就足够了。
with db.engine.begin() as conn:
posts = Post.__table__
update = sa.update(posts).where(...).values(flag=True, last_updated=posts.c.last_updated)
conn.execute(update)
如果我有一个 post 的列表,其中 created
和 updated
日期带有默认附加 onupdate
回调。
有时我需要为 inappropriate reports
或类似操作标记 post。我不想修改 created
和 updated
日期。
如何在执行 update
操作时跳过定义的 onupdate
?
SQLAlchemy 将应用默认 when
no value was provided to the INSERT or UPDATE statement for that column
然而,明显的解决方法 - 显式将列设置为其当前值 - 将不起作用,因为会话 checks 值是否已实际更改,如果没有,则不会传递值。这里有两个可能的解决方案,假设 SQLAlchemy 1.4+ 和这个模型:
class Post(db.Model):
flag = db.Column(db.Boolean, default=False)
last_updated = db.Column(db.DateTime, default=some_func, onupdate=some_func)
使用事件侦听器
添加一个 before update listener that detects when the flag column is being modified, and mark 修改后的时间戳列,即使其值未更改。这将使 SQLAlchemy 将当前值添加到更新中,因此不会调用 onupdate 函数。
import sqlalchemy as sa
...
@sa.event.listens_for(Post, 'before_update')
def receive_before_update(mapper, connection, target):
insp = sa.inspect(target)
flag_changed, _, _ = insp.attrs.flag.history
if flag_changed:
orm.attributes.flag_modified(target, 'last_updated')
使用 SQLAlchemy 核心而不是 ORM
SQLAlchemy 核心不需要会话,因此可以将当前时间戳值传递给更新以避免触发 onupdate 函数。 ORM 不会意识到以这种方式所做的任何更改,因此如果在会话的上下文中完成,受影响的对象应该被刷新或过期。这是一个“快速而肮脏”的解决方案,但如果标记发生在正常应用程序流程之外,可能就足够了。
with db.engine.begin() as conn:
posts = Post.__table__
update = sa.update(posts).where(...).values(flag=True, last_updated=posts.c.last_updated)
conn.execute(update)