TL;DR - version controlled database triggers to automatically update updated_at field, and make the ORM aware and issue a RETURNING statement.
WHY? - because we want to avoid relying purely on ORM's onupdate to make suer update_at field is up-to-date.
NOTE: this code was designed for PostgreSQL, but would be similar for other databases.
To extend on this answer, for a default value you should use server_default as suggested. However, for having a value that will be updated upon an UPDATE statement, you have 2 options. First is described in the linked answer. The second option is using database triggers. I found alembic-utils useful to make sure triggered is migrated and updated when the version-controlled is modified. Combining it with a Base model and table listing allowed adding those created/updated fields to all ORM models.
from sqlalchemy.orm import Mapped, mapped_column
from sqlalchemy import func, FetchedValue
class Base(DeclarativeBase):
__abstract__ = True
metadata = MetaData(naming_convention=convention)
id: Mapped[int] = mapped_column(Identity(), primary_key=True)
created_at: Mapped[datetime] = mapped_column(server_default=func.now())
updated_at: Mapped[datetime] = mapped_column(
server_default=FetchedValue(), server_onupdate=FetchedValue()
)
# IMPORTANT! see details below
__mapper_args__ = {"eager_defaults": True}
class Customer(Base):
__tablename__ = "customers"
name: Mapped[str] = mapped_column(nullable=False)
triggers.py:
from alembic_utils.pg_function import PGFunction
from alembic_utils.pg_trigger import PGTrigger
from alembic_utils.replaceable_entity import register_entities
# import your `Base` module - ONLY AFTER all other modules were imported as well
# that's anyway required for handling the metadata in `alembic`.
from ... import Base
update_update_time_func = PGFunction(
schema="public",
signature="update_updated_at()",
definition="""
RETURNS TRIGGER AS $$
BEGIN
NEW.updated_at = (CURRENT_TIMESTAMP AT TIME ZONE 'UTC');
RETURN NEW;
END;
$$ LANGUAGE plpgsql
""",
)
register_entities([update_update_time_func])
update_time_triggers = [
PGTrigger(
schema="public",
signature=f"update_{tb_name}_updated_at",
on_entity=tb_name,
definition=f"""
BEFORE INSERT OR UPDATE ON {tb_name}
FOR EACH ROW EXECUTE FUNCTION update_updated_at()
""",
)
for tb_name, tb in Base.metadata.tables.items()
if any(c.name == "updated_at" for c in tb.columns)
]
register_entities(update_time_triggers)
In migrations/env.py import your triggers module.
It will register the trigger entities and the stored procedure.
running alembic revision --autogenerate -m "add triggers" should identify the function and triggers, and alembic upgrade head should add the function and triggers.
Eager defaults and performance: implementing this solution, it was important for me both:
- After an INSERT or UPDATE operation (done via
ORM), the instance at hand will be updated by the database-generated values (created and updated timestamps, in this case).
- The operation will not trigger a SELECT, but instead use RETURNING where appropriate.
To demonstrate the above consider the following code:
with Session() as session:
c = Customer(name='foo')
session.add(c)
print(f'Instance before INSERT: {c}')
print('INSERT logs:')
session.commit()
print(f'Instance after INSERT: {c}')
c.name = 'bar'
print('UPDATE logs:')
session.commit()
print(f'Instance after UPDATE: {c}')
Which produces these logs (slightly edited for readability):
Instance before INSERT:
<Customer(name='foo')>
INSERT logs:
BEGIN (implicit)
INSERT INTO customers (name) VALUES (%(name)s) RETURNING customers.id, customers.created_at, customers.updated_at
[cached since 1.849e+04s ago] {'name': 'foo'}
COMMIT
Instance after INSERT:
<Customer(name='foo', id=8, created_at=datetime.datetime(2023, 12, 5, 14, 23, 3, 964627), updated_at=datetime.datetime(2023, 12, 5, 14, 23, 3, 964627))>
UPDATE logs:
BEGIN (implicit)
UPDATE customers SET name=%(name)s WHERE customers.id = %(customers_id)s RETURNING customers.updated_at
[cached since 220s ago] {'name': 'bar', 'customers_id': 8}
COMMIT
Instance after UPDATE:
<Customer(name='bar', id=8, created_at=datetime.datetime(2023, 12, 5, 14, 23, 3, 964627), updated_at=datetime.datetime(2023, 12, 5, 14, 23, 3, 970578))>
