[SOLVED] SQLAlchemy: Set audit columns prior to insert or update

SQLAlchemy: Set audit columns prior to insert or update

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

Solution

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)