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?

There is no equivalent syntax in Postgres. You can only achieve that using a trigger — user330315
– user330315, Commented Feb 1, 2022 at 19:09
It is spelled out in the docs SQL expressions. See onupdate. — Adrian Klaver
– Adrian Klaver, Commented Feb 1, 2022 at 19:13
@AdrianKlaver According to my tests simply passing onupdate=func.utc_timestamp() has no effect on a Postgres backend (it might in others) — swimmer
– swimmer, Commented Feb 2, 2022 at 15:41
@a_horse_with_no_name that is consistent with some of the reading I've done. I hope someone will answer with a sample implementation of a trigger for the example users table above — swimmer
– swimmer, Commented Feb 2, 2022 at 15:44

swimmer · Accepted Answer · 2022-02-10 21:15:20Z

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

So we concluded that onupdate= doesn't work with Posgresql and the only solution is to create a trigger?
server_onupdate does not work like you expect it. But onupdate does.

Tom · Accepted Answer · 2024-11-26 20:40:23Z

I believe that the accepted answer is for SQLAlchemy 1.4, and 2.0 has quite a different syntax, so I'm adding this answer for 2.x:

from typing import Optional
from datetime import datetime

from sqlalchemy import BigInteger, ForeignKey, Identity, text, Text, TIMESTAMP
from sqlalchemy.orm import Mapped, mapped_column
from sqlalchemy.dialects.postgresql import JSONB


class Post(Base):
    __tablename__ = "posts"
    
    id: Mapped[int] = mapped_column(
        BigInteger, Identity(always=True), primary_key=True
    )   
    user_id: Mapped[Optional[UUID]] = mapped_column(ForeignKey("users.id"))
    title: Mapped[str] = mapped_column(Text, nullable=False)
    content: Mapped[str] = mapped_column(Text, nullable=False)
    cache: Mapped[Optional[dict]] = mapped_column(JSONB)
    created_at: Mapped[datetime] = mapped_column(
        TIMESTAMP(timezone=True),
        nullable=False,
        server_default=text("CURRENT_TIMESTAMP")
    )

Which produces ...

CREATE TABLE posts (
  id BIGINT GENERATED ALWAYS AS IDENTITY, 
  user_id UUID, 
  title TEXT NOT NULL, 
  content TEXT NOT NULL, 
  cache JSONB, 
  created_at TIMESTAMP WITH TIME ZONE DEFAULT CURRENT_TIMESTAMP NOT NULL, 
  PRIMARY KEY (id),

And I believe this generally conforms to Postgres recommendations.

Collectives™ on Stack Overflow

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

2 Answers 2

SQLAlchemy implementation

Integration with Alembic

2 Comments

Comments

Your Answer

Linked

Hot Network Questions

Collectives™ on Stack Overflow

2 Answers 2

SQLAlchemy implementation

Integration with Alembic

2 Comments

Comments

Your Answer

Sign up or log in

Post as a guest

Linked

Related