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