如何使用 SQLAlchemy 在 Postgres 中设置 DEFAULT ON UPDATE CURRENT_TIMESTAMP？

Question

这是How to set DEFAULT ON UPDATE CURRENT_TIMESTAMP in mysql with sqlalchemy? 的姐妹问题？，但专注于 Postgres 而不是 MySQL。

假设我们要创建一个表users，datemodified其中的一列在更新行时默认更新为当前时间戳。MySQL的姐妹PR中给出的解决方案是：

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

如何使用 Postgres 后端获得相同的功能？

Answer 1

最终我按照评论中 a_horse_with_no_name 的建议使用触发器实现了这一点。接下来是完整的 SQLAlchemy 实现以及与 Alembic 的集成。

SQLAlchemy 实现

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

与 Alembic 集成

就我而言，将触发器创建与 Alembic 集成，并将触发器添加到n 个维度表（所有维度表都有一updated_at列）非常重要。

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