ngse · May 21, 2023 13:56 · Mar 26, 2017 · Mar 26, 2017
diff --git a/audit_mixin.py b/audit_mixin.py
@@ -29,8 +29,6 @@ class Base(db.Model):
     def __tablename__(cls):
         return cls.__name__.lower()
 
-    __table_args__ = {'mysql_engine': 'InnoDB'}
-
     id = db.Column(db.Integer, primary_key=True, nullable=False)
 
 

diff --git a/audit_mixin.py b/audit_mixin.py
@@ -0,0 +1,122 @@
+# 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()
+
+    __table_args__ = {'mysql_engine': 'InnoDB'}
+
+    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))
No results found