Rough attempt at implementing an audit log against Flask/Flask Login/SQLAlchemy models

audit_mixin.py

# Requires use of Flask Login for the user tracking
# Implement by using AuditableMixin in your model class declarations
# e.g. class ImportantThing(AuditableMixin, Base):
import json

from flask_login import current_user
from sqlalchemy import event, inspect
from sqlalchemy.orm import class_mapper
from sqlalchemy.orm.attributes import get_history

from app import db

ACTION_CREATE = 1
ACTION_UPDATE = 2
ACTION_DELETE = 3


def _current_user_id_or_none():
    try:
        return current_user.id
    except:
        return None
      
      
class Base(db.Model):
    """Base model class to implement db columns and features every model should have"""
    __abstract__ = True

    def __tablename__(cls):
        return cls.__name__.lower()

    id = db.Column(db.Integer, primary_key=True, nullable=False)


class TimestampableMixin:
    """Allow a model to track its creation and update times"""
    created_at = db.Column(db.DateTime, default=db.func.current_timestamp())
    updated_at = db.Column(db.DateTime, default=db.func.current_timestamp(), onupdate=db.func.current_timestamp())


class AuditLog(TimestampableMixin, Base):
    """Model an audit log of user actions"""
    user_id = db.Column(db.Integer, doc="The ID of the user who made the change")
    target_type = db.Column(db.String(100), nullable=False, doc="The table name of the altered object")
    target_id = db.Column(db.Integer, doc="The ID of the altered object")
    action = db.Column(db.Integer, doc="Create (1), update (2), or delete (3)")
    state_before = db.Column(db.Text, doc="Stores a JSON string representation of a dict containing the altered column "
                                          "names and original values")
    state_after = db.Column(db.Text, doc="Stores a JSON string representation of a dict containing the altered column "
                                         "names and new values")

    def __init__(self, target_type, target_id, action, state_before, state_after):
        self.user_id = _current_user_id_or_none()
        self.target_type = target_type
        self.target_id = target_id
        self.action = action
        self.state_before = state_before
        self.state_after = state_after

    def __repr__(self):
        return '<AuditLog %r: %r -> %r>' % (self.user_id, self.target_type, self.action)

    def save(self, connection):
        connection.execute(
            self.__table__.insert(),
            user_id=self.user_id,
            target_type=self.target_type,
            target_id=self.target_id,
            action=self.action,
            state_before=self.state_before,
            state_after=self.state_after
        )


class AuditableMixin:
    """Allow a model to be automatically audited"""

    @staticmethod
    def create_audit(connection, object_type, object_id, action, **kwargs):
        audit = AuditLog(
            object_type,
            object_id,
            action,
            kwargs.get('state_before'),
            kwargs.get('state_after')
        )
        audit.save(connection)

    @classmethod
    def __declare_last__(cls):
        event.listen(cls, 'after_insert', cls.audit_insert)
        event.listen(cls, 'after_delete', cls.audit_delete)
        event.listen(cls, 'after_update', cls.audit_update)

    @staticmethod
    def audit_insert(mapper, connection, target):
        """Listen for the `after_insert` event and create an AuditLog entry"""
        target.create_audit(connection, target.__tablename__, target.id, ACTION_CREATE)

    @staticmethod
    def audit_delete(mapper, connection, target):
        """Listen for the `after_delete` event and create an AuditLog entry"""
        target.create_audit(connection, target.__tablename__, target.id, ACTION_DELETE)

    @staticmethod
    def audit_update(mapper, connection, target):
        """Listen for the `after_update` event and create an AuditLog entry with before and after state changes"""
        state_before = {}
        state_after = {}
        inspr = inspect(target)
        attrs = class_mapper(target.__class__).column_attrs
        for attr in attrs:
            hist = getattr(inspr.attrs, attr.key).history
            if hist.has_changes():
                state_before[attr.key] = get_history(target, attr.key)[2].pop()
                state_after[attr.key] = getattr(target, attr.key)

        target.create_audit(connection, target.__tablename__, target.id, ACTION_UPDATE,
                            state_before=json.dumps(state_before),
                            state_after=json.dumps(state_after))

I had to explicitly set my tablename to get this work:

class AuditLog(Base, lib.audit_mixin.AuditLog):

    __tablename__ = 'checkrecon_auditlog'

Also, I made the code free of Flask-isms, so that anything that wanted to use the code could:
https://gist.github.com/metaperl/34e7b801532f2b9da02c91a7797e2e85#file-audit_mixin-py-L63

and Base is actually a class from the Reahl web app framework --- reahl.sqlalchemysupport.Base.

Does not work on date objects - they are not JSON seriablizeable. See https://stackoverflow.com/questions/11875770/how-to-overcome-datetime-datetime-not-json-serializable/36142844#36142844

I had to explicitly set my tablename to get this work:

class AuditLog(Base, lib.audit_mixin.AuditLog):

    __tablename__ = 'checkrecon_auditlog'

Also, I made the code free of Flask-isms, so that anything that wanted to use the code could:
https://gist.github.com/metaperl/34e7b801532f2b9da02c91a7797e2e85#file-audit_mixin-py-L63

and Base is actually a class from the Reahl web app framework --- reahl.sqlalchemysupport.Base.

This can work after add @declared_attr

from sqlalchemy.ext.declarative import declared_attr

class Base(db.Model):
    """Base model class to implement db columns and features every model should have"""
    __abstract__ = True

     @declared_attr  # <-- missing this line
    def __tablename__(cls):
        return cls.__name__.lower()

    id = db.Column(db.Integer, primary_key=True, nullable=False)

Thanks for your work.
I suggest that we should add something to state_before when we make ACTION_DELETE, because only target_id does not tell much about the record which already deleted (it's fine with soft delete).
This might become handy if your table has more than one unique keys, for example User table often has id and email as unique keys.