SQLAlchemy: Set audit columns prior to insert or update

Question

Each of my mapped class contains created_by and updated_by audit properties that I would like to set automatically upon INSERT and UPDATE of respective objects.

class User(Base):
    __tablename__ = 'user'

    id = Column(BigInteger, primary_key=True)
    name = Column(Text, nullable=False)
    ...

class Address(Base):
    __tablename__ = 'address'

    id = Column(BigInteger, primary_key=True)
    street = Column(Text, nullable=False)
    ...
    created_by = Column(BigInteger)  # references user.id
    updated_by = Column(BigInteger)  # references user.id
    ...

Is there a way to handle this centrally in SQLAlchemy? I looked at the events but it appears it needs to be setup for every single mapped class individually (note the SomeClass in the decorator).

@event.listens_for(SomeClass, 'before_insert')
def on_insert(mapper, connection, target):
    target.created_by = context["current_user"] # I want to be able to do this not just for 'SomeClass' but for all mapped classes

@event.listens_for(SomeClass, 'before_update')
def on_update(mapper, connection, target):
    target.updated_by = context["current_user"] # I want to be able to do this not just for 'SomeClass' but for all mapped classes

Answer 1

One solution here is to use the default parameters in the Column class provided by sqlalchemy. You can actually pass a callable to both default (to execute when first created) and onupdate to execute whenever updated.

def get_current_user():
    return context["user"].id


class Address(Base):
    __tablename__ = 'address'
    ...
    created_by = Column(default = get_current_user)
    updated_by = Column(default = get_current_user, onupdate=get_current_user)