SQLAlchemy中server_onupdate未触发updated_at更新的问题咨询

你的问题核心有两点：

1. server_onupdate=text('now()') 仅在通过SQLAlchemy ORM更新Task对象时生效，直接执行原生SQL更新（包括关联表projects的更新）不会触发该逻辑——因为它是ORM层面的配置，而非数据库级别的自动更新规则。
2. 要实现无论哪种更新方式（ORM/原生SQL/关联表变更触发的级联操作）都自动更新updated_at，必须依赖数据库触发器。

1. 添加数据库触发器实现全局自动更新 #

在SQLAlchemy模型中，可通过__table_args__添加创建触发器的DDL语句，确保表创建时同步生成触发器：

from sqlalchemy import Column, Integer, String, TIMESTAMP, ForeignKey, text, DDL, UniqueConstraint
from sqlalchemy.ext.declarative import declarative_base

Base = declarative_base()

class Task(Base):
    __tablename__ = "tasks"

    id = Column(Integer, primary_key=True, index=True, nullable=False)
    name = Column(String, index=True, nullable=False)
    project_id = Column(Integer, ForeignKey("projects.id", ondelete="CASCADE"), nullable=False)
    created_at = Column(TIMESTAMP(timezone=True), nullable=False, server_default=text("now()"))
    updated_at = Column(TIMESTAMP(timezone=True), nullable=False, server_default=text("now()"))

    __table_args__ = (
        UniqueConstraint("name", "project_id", name="unique_task_name_project_id"),
        # PostgreSQL触发器：更新任意字段时自动刷新updated_at
        DDL("""
            CREATE OR REPLACE FUNCTION update_task_updated_at()
            RETURNS TRIGGER AS $$
            BEGIN
                NEW.updated_at = NOW();
                RETURN NEW;
            END;
            $$ LANGUAGE plpgsql;
            
            CREATE TRIGGER trigger_task_update
            BEFORE UPDATE ON tasks
            FOR EACH ROW
            EXECUTE FUNCTION update_task_updated_at();
        """),
    )

    def __repr__(self):
        return f"<Task {self.name}>"

不同数据库适配： #

• 若使用MySQL，替换触发器DDL为：

  CREATE TRIGGER trigger_task_update
  BEFORE UPDATE ON tasks
  FOR EACH ROW
  SET NEW.updated_at = CURRENT_TIMESTAMP();
  

2. 禁止手动设置updated_at字段 #

可从ORM和数据库两个层面实现限制：

层面1：ORM层面拦截 #

通过SQLAlchemy事件监听，强制重置或禁止手动赋值：

from sqlalchemy import event

# 强制重置为数据库当前时间，覆盖手动设置的值
@event.listens_for(Task, "before_update")
def receive_before_update(mapper, connection, target):
    target.updated_at = text("now()")

# 可选：直接禁止手动赋值，触发报错
@event.listens_for(Task.updated_at, "set", propagate=True)
def prevent_updated_at_set(target, value, oldvalue, initiator):
    raise ValueError("禁止手动设置`updated_at`字段")

层面2：数据库层面限制 #

利用触发器强制覆盖任何手动输入的值（前面的触发器已经实现此效果，NEW.updated_at = NOW()会直接覆盖手动修改的内容）；也可通过规则进一步限制：

-- PostgreSQL示例：禁止手动修改updated_at，仅允许触发器更新
CREATE RULE prevent_manual_updated_at AS
ON UPDATE TO tasks
WHERE NEW.updated_at IS NOT DISTINCT FROM OLD.updated_at
DO INSTEAD NOTHING;

• 若tasks表已存在，需手动执行触发器创建语句，或删除旧表后重新生成。
• 若需要projects表更新时同步刷新关联tasks的updated_at，可给projects表添加额外触发器：

  -- PostgreSQL示例：projects更新时，更新关联tasks的updated_at
  CREATE OR REPLACE FUNCTION update_tasks_on_project_change()
  RETURNS TRIGGER AS $$
  BEGIN
      UPDATE tasks SET updated_at = NOW() WHERE project_id = OLD.id;
      RETURN NEW;
  END;
  $$ LANGUAGE plpgsql;
  
  CREATE TRIGGER trigger_project_update
  AFTER UPDATE ON projects
  FOR EACH ROW
  EXECUTE FUNCTION update_tasks_on_project_change();
  

内容的提问来源于stack exchange，提问作者Carol Eisen