Advertisement
Not a member of Pastebin yet?
Sign Up,
it unlocks many cool features!
- # 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))
Advertisement
Add Comment
Please, Sign In to add comment
Advertisement