Advertisement
Guest User

Untitled

a guest
Mar 26th, 2017
55
0
Never
Not a member of Pastebin yet? Sign Up, it unlocks many cool features!
text 4.46 KB | None | 0 0
  1. # Requires use of Flask Login for the user tracking
  2. # Implement by using AuditableMixin in your model class declarations
  3. # e.g. class ImportantThing(AuditableMixin, Base):
  4. import json
  5.  
  6. from flask_login import current_user
  7. from sqlalchemy import event, inspect
  8. from sqlalchemy.orm import class_mapper
  9. from sqlalchemy.orm.attributes import get_history
  10.  
  11. from app import db
  12.  
  13. ACTION_CREATE = 1
  14. ACTION_UPDATE = 2
  15. ACTION_DELETE = 3
  16.  
  17.  
  18. def _current_user_id_or_none():
  19. try:
  20. return current_user.id
  21. except:
  22. return None
  23.  
  24.  
  25. class Base(db.Model):
  26. """Base model class to implement db columns and features every model should have"""
  27. __abstract__ = True
  28.  
  29. def __tablename__(cls):
  30. return cls.__name__.lower()
  31.  
  32. id = db.Column(db.Integer, primary_key=True, nullable=False)
  33.  
  34.  
  35. class TimestampableMixin:
  36. """Allow a model to track its creation and update times"""
  37. created_at = db.Column(db.DateTime, default=db.func.current_timestamp())
  38. updated_at = db.Column(db.DateTime, default=db.func.current_timestamp(), onupdate=db.func.current_timestamp())
  39.  
  40.  
  41. class AuditLog(TimestampableMixin, Base):
  42. """Model an audit log of user actions"""
  43. user_id = db.Column(db.Integer, doc="The ID of the user who made the change")
  44. target_type = db.Column(db.String(100), nullable=False, doc="The table name of the altered object")
  45. target_id = db.Column(db.Integer, doc="The ID of the altered object")
  46. action = db.Column(db.Integer, doc="Create (1), update (2), or delete (3)")
  47. state_before = db.Column(db.Text, doc="Stores a JSON string representation of a dict containing the altered column "
  48. "names and original values")
  49. state_after = db.Column(db.Text, doc="Stores a JSON string representation of a dict containing the altered column "
  50. "names and new values")
  51.  
  52. def __init__(self, target_type, target_id, action, state_before, state_after):
  53. self.user_id = _current_user_id_or_none()
  54. self.target_type = target_type
  55. self.target_id = target_id
  56. self.action = action
  57. self.state_before = state_before
  58. self.state_after = state_after
  59.  
  60. def __repr__(self):
  61. return '<AuditLog %r: %r -> %r>' % (self.user_id, self.target_type, self.action)
  62.  
  63. def save(self, connection):
  64. connection.execute(
  65. self.__table__.insert(),
  66. user_id=self.user_id,
  67. target_type=self.target_type,
  68. target_id=self.target_id,
  69. action=self.action,
  70. state_before=self.state_before,
  71. state_after=self.state_after
  72. )
  73.  
  74.  
  75. class AuditableMixin:
  76. """Allow a model to be automatically audited"""
  77.  
  78. @staticmethod
  79. def create_audit(connection, object_type, object_id, action, **kwargs):
  80. audit = AuditLog(
  81. object_type,
  82. object_id,
  83. action,
  84. kwargs.get('state_before'),
  85. kwargs.get('state_after')
  86. )
  87. audit.save(connection)
  88.  
  89. @classmethod
  90. def __declare_last__(cls):
  91. event.listen(cls, 'after_insert', cls.audit_insert)
  92. event.listen(cls, 'after_delete', cls.audit_delete)
  93. event.listen(cls, 'after_update', cls.audit_update)
  94.  
  95. @staticmethod
  96. def audit_insert(mapper, connection, target):
  97. """Listen for the `after_insert` event and create an AuditLog entry"""
  98. target.create_audit(connection, target.__tablename__, target.id, ACTION_CREATE)
  99.  
  100. @staticmethod
  101. def audit_delete(mapper, connection, target):
  102. """Listen for the `after_delete` event and create an AuditLog entry"""
  103. target.create_audit(connection, target.__tablename__, target.id, ACTION_DELETE)
  104.  
  105. @staticmethod
  106. def audit_update(mapper, connection, target):
  107. """Listen for the `after_update` event and create an AuditLog entry with before and after state changes"""
  108. state_before = {}
  109. state_after = {}
  110. inspr = inspect(target)
  111. attrs = class_mapper(target.__class__).column_attrs
  112. for attr in attrs:
  113. hist = getattr(inspr.attrs, attr.key).history
  114. if hist.has_changes():
  115. state_before[attr.key] = get_history(target, attr.key)[2].pop()
  116. state_after[attr.key] = getattr(target, attr.key)
  117.  
  118. target.create_audit(connection, target.__tablename__, target.id, ACTION_UPDATE,
  119. state_before=json.dumps(state_before),
  120. state_after=json.dumps(state_after))
Advertisement
Add Comment
Please, Sign In to add comment
Advertisement