SQLAlchemy结合PostgreSQL时onupdate=func.now()时间戳异常疑问

执行以下代码时，5秒休眠后预期date_updated字段的秒数部分会发生变化，但实际仅毫秒部分改变；若改用database_url = 'sqlite:///:memory:'则符合预期，请问这是什么原因？

from sqlalchemy import create_engine, func, TIMESTAMP
from sqlalchemy.orm import Mapped, mapped_column, Session, DeclarativeBase
from sqlalchemy.dataclasses import MappedAsDataclass
from sqlalchemy.engine import URL
import time
from datetime import datetime

class Base(MappedAsDataclass, DeclarativeBase):
    pass


class Test(Base):
    __tablename__ = 'test'

    test_id: Mapped[int] = mapped_column(primary_key=True, init=False)
    name: Mapped[str]
    date_created: Mapped[datetime] = mapped_column(
        TIMESTAMP(timezone=True),
        insert_default=func.now(),
        init=False
    )
    date_updated: Mapped[datetime] = mapped_column(
        TIMESTAMP(timezone=True),
        nullable=True,
        insert_default=None,
        onupdate=func.now(),
        init=False
    )


database_url: URL = URL.create(
    drivername='postgresql+psycopg',
    username='my_username',
    password='my_password',
    host='localhost',
    port=5432,
    database='my_db'
)
engine = create_engine(database_url, echo=True)
Base.metadata.drop_all(engine)
Base.metadata.create_all(engine)

with Session(engine) as session:
    test = Test(name='foo')
    session.add(test)
    session.commit()

    print(test)
    
    time.sleep(5)

    test.name = 'bar'    
    session.commit()

    print(test.date_created.time()) # prints: 08:07:45.413737
    print(test.date_updated.time()) # prints: 08:07:45.426483

原因分析 #

核心问题出在PostgreSQL的func.now()行为与Session（事务）的绑定关系：

• PostgreSQL中，now()（等价于transaction_timestamp()）会返回当前事务的启动时间，整个事务周期内多次调用都会返回同一个值。你在同一个Session里完成了插入、休眠、更新操作，这些都属于同一个事务上下文，所以第二次commit时onupdate=func.now()拿到的还是事务启动时的时间，仅毫秒差异是事务内操作的微小延迟导致的。
• 而SQLite的func.now()没有事务绑定限制，每次调用都会取当前系统的实际时间，所以休眠后秒数会正常更新。

解决方法 #

有两种常用方案：

• 方案1：拆分事务：休眠后关闭当前Session，重新创建Session再执行更新操作，新事务的now()会取最新时间。
• 方案2：改用clock_timestamp()：PostgreSQL的clock_timestamp()函数会返回当前实际的系统时间，不受事务启动时间限制。修改date_updated的onupdate参数为func.clock_timestamp()即可：

  date_updated: Mapped[datetime] = mapped_column(
      TIMESTAMP(timezone=True),
      nullable=True,
      insert_default=None,
      onupdate=func.clock_timestamp(),
      init=False
  )
  

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