[FIXED] Skip the default on onupdate defined, for specific update queries in SQLAlchemy

Issue

If I have a list of posts, which have created and updated dates with a default attached onupdate callback.

Sometimes I need to flag the post, for inappropriate reports or similar actions. I do not want the created and updated dates to be modified.

How can I skip the defined onupdate, while making an update action?

Solution

SQLAlchemy will apply a default when

no value was provided to the INSERT or UPDATE statement for that column

however the obvious workaround - explicitly setting the column to its current value - won't work because the session checks whether the value has actually changed, and does not pass a value if it hasn't. Here are two possible solutions, assuming SQLAlchemy 1.4+ and this model:

class Post(db.Model):
    flag = db.Column(db.Boolean, default=False)
    last_updated = db.Column(db.DateTime, default=some_func, onupdate=some_func)

Use an event listener

Add a before update listener that detects when the flag column is being modified, and mark the timestamp column as modified, even though its value is unchanged. This will make SQLAlchemy add the current value to the update, and so the onupdate function will not be called.

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')

Use SQLAlchemy core instead of the ORM

SQLAlchemy core doesn't need a session, so the current timestamp value can be passed to an update to avoid triggering the onupdate function. The ORM will be unaware of any changes made in this way, so if done within the context of a session affected objects should be refreshed or expired. This is a "quick and dirty" solution, but might be good enough if flagging happens outside of the normal application flow.

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)

Answered By - snakecharmerb