How to set DEFAULT ON UPDATE CURRENT_TIMESTAMP in Postgres with SQLAlchemy?

Question

This is a sister question to How to set DEFAULT ON UPDATE CURRENT_TIMESTAMP in mysql with sqlalchemy?, but focused on Postgres instead of MySQL.

Say we want to create a table users with a column datemodified that updates by default to the current timestamp whenever a row is updated. The solution given in the sister PR for MySQL is:

user = Table(
    "users",
    Metadata,
    Column(
        "datemodified",
        TIMESTAMP,
        server_default=text("CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP"),
    ),
)

How can I get the same functionality with a Postgres backend?

Answer 1

Eventually I implemented this using triggers as suggested by a_horse_with_no_name in the comments. Full SQLAlchemy implementation and integration with Alembic follow.

SQLAlchemy implementation

# models.py

class User(Base):
    __tablename__ = "user"
    id = Column(Integer, primary_key=True)
    name = Column(Text)
    created_at = Column(DateTime, server_default=sqlalchemy.func.now(), nullable=False)
    updated_at = Column(DateTime)

# your_application_code.py

import sqlalchemy as sa

create_refresh_updated_at_func = """
    CREATE FUNCTION {schema}.refresh_updated_at()
    RETURNS TRIGGER
    LANGUAGE plpgsql AS
    $func$
    BEGIN
       NEW.updated_at := now();
       RETURN NEW;
    END
    $func$;
    """

create_trigger = """
    CREATE TRIGGER trig_{table}_updated BEFORE UPDATE ON {schema}.{table}
    FOR EACH ROW EXECUTE PROCEDURE {schema}.refresh_updated_at();
    """

my_schema = "foo"
engine.execute(sa.text(create_refresh_updated_at_func.format(schema=my_schema)))
engine.execute(sa.text(create_trigger.format(schema=my_schema, table="user")))

Integration with Alembic

In my case it was important to integrate the trigger creation with Alembic, and to add the trigger to n dimension tables (all of them having an updated_at column).

# alembic/versions/your_version.py

import sqlalchemy as sa

create_refresh_updated_at_func = """
    CREATE FUNCTION {schema}.refresh_updated_at()
    RETURNS TRIGGER
    LANGUAGE plpgsql AS
    $func$
    BEGIN
       NEW.updated_at := now();
       RETURN NEW;
    END
    $func$;
    """

create_trigger = """
    CREATE TRIGGER trig_{table}_updated BEFORE UPDATE ON {schema}.{table}
    FOR EACH ROW EXECUTE PROCEDURE {schema}.refresh_updated_at();
    """

def upgrade():
    op.create_table(..., schema="foo")
    ...

    # Add updated_at triggers for all tables
    op.execute(sa.text(create_refresh_updated_at_func.format(schema="foo")))
    for table in MY_LIST_OF_TABLES:
        op.execute(sa.text(create_trigger.format(schema="foo", table=table)))

def downgrade():
    op.drop_table(..., schema="foo")
    ...

    op.execute(sa.text("DROP FUNCTION foo.refresh_updated_at() CASCADE"))