SQLAlchemy 2.0 hybrid properties, column_property and before_flush audit events: working versions (tested 2026-10-15)

sqlalchemy==2.1.4
pytest==9.1.1

from datetime import datetime, timezone

from sqlalchemy import String, event, func
from sqlalchemy.ext.hybrid import hybrid_property
from sqlalchemy.orm import DeclarativeBase, Mapped, Session, column_property, mapped_column


class Base(DeclarativeBase):
    pass


class Account(Base):
    __tablename__ = "accounts"

    id: Mapped[int] = mapped_column(primary_key=True)
    first: Mapped[str] = mapped_column(String(50))
    last: Mapped[str] = mapped_column(String(50))
    created_at: Mapped[datetime | None] = mapped_column(default=None)
    updated_at: Mapped[datetime | None] = mapped_column(default=None)

    @hybrid_property
    def full_name(self) -> str:
        # instance level: plain Python value
        return f"{self.first} {self.last}"

    @full_name.expression
    def full_name(cls):
        # class level: SQL expression usable in WHERE/ORDER BY
        return cls.first + " " + cls.last


# column_property added after mapping, since it references the mapped columns
Account.name_len = column_property(func.length(Account.first) + func.length(Account.last))


@event.listens_for(Session, "before_flush")
def stamp_audit_fields(session, flush_context, instances):
    # before_flush changes are included in the same INSERT/UPDATE
    now = datetime.now(timezone.utc)
    for obj in session.new:
        if isinstance(obj, Account):
            obj.created_at = now
    for obj in session.dirty:
        if isinstance(obj, Account) and session.is_modified(obj):
            obj.updated_at = now

import pytest
from sqlalchemy import create_engine, select
from sqlalchemy.orm import Session

from models import Account, Base


@pytest.fixture
def session():
    engine = create_engine("sqlite://")
    Base.metadata.create_all(engine)
    with Session(engine) as s:
        yield s
    engine.dispose()


def test_hybrid_instance_and_expression(session):
    session.add(Account(first="Ada", last="Lovelace"))
    session.flush()
    acct = session.scalars(select(Account).where(Account.full_name == "Ada Lovelace")).one()
    assert acct.full_name == "Ada Lovelace"


def test_column_property_is_selectable(session):
    session.add(Account(first="Ada", last="Lovelace"))
    session.commit()
    assert session.execute(select(Account.name_len)).scalar_one() == 11


def test_before_flush_stamps_created_at(session):
    a = Account(first="A", last="B")
    session.add(a)
    session.commit()
    assert a.created_at is not None
    assert a.updated_at is None
    # SQLite drops tzinfo on storage, so the reloaded value is naive
    assert a.created_at.tzinfo is None


def test_updated_at_only_on_real_change(session):
    a = Account(first="Ada", last="Lovelace")
    session.add(a)
    session.commit()
    session.commit()  # nothing dirty, so no stamp
    assert a.updated_at is None
    a.last = "King"
    session.commit()
    assert a.updated_at is not None

[pytest]
pythonpath = .