SQLAlchemy: Set audit columns prior to insert or update

user_12106502 0
Viewed 46

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
auditing sqlalchemy
2 Answers

Managed to figure it out, though somewhat concerned about using a dunder method __subclasses__() on declarative_base. If there is a better alternative do suggest.

def on_insert(mapper, connection, target):
    target.created_by = context["user"].id
    target.updated_at = datetime.utcnow()


def on_update(mapper, connection, target):
    target.updated_by = context["user"].id
    target.updated_at = datetime.utcnow()


Base.metadata.create_all()

mapped_classes = Base.__subclasses__()
for mapped_class in mapped_classes:
    event.listen(mapped_class, 'before_insert', on_insert)
    event.listen(mapped_class, 'before_update', on_update)

The context being referred to here is actually starlette-context

user_121065020

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)
user_58335900
Related