Changes between Version 1 and Version 2 of AuditTrail

Timestamp:

Aug 27, 2007, 8:42:32 AM (17 years ago)

Author:

George Vilches <gav@…>

Comment:

--

AuditTrail

@@ -25 +25 @@
 }}}
 
-This simple addition will do the rest, allowing you to run `syncdb` and install the audit model. Once it's installed, the following code will work as shown below. As you will see, `Person.history` becomes a manager that's used to access the audit trail for a particular object.
+This simple addition will do the rest, allowing you to run `syncdb` and install the audit model. Once it's installed, the following code will work as shown below.  As you will see, `Person.history` becomes a manager that's used to access the audit trail for a particular object.  The type of manager available depends on how you access the audit trail.  From an instance, the audit trail will automatically be filtered to only return results related to that instance.  From the class itself, the results will not be filtered in any way, and is the likely approach for doing reporting across several audited items.
 
 {{{
@@ -43 +43 @@
 John Public as of 2007-08-14 20:31:21.852000: 65000
 John Public as of 2007-08-14 20:30:58.959000: 50000
+>>> person2 = Person.objects.create(first_name='Tom', last_name='Smith', salary=25000)
+>>> person2
+<Person: Tom Smith>
+>>> person.history.count()
+2
+>>> person.history.count()
+2L
+>>> person2.history.count()
+1L
+>>> Person.history.count()
+3L
 }}}
 
@@ -63 +74 @@
 {{{
 #!python
-import re
-
 from django.dispatch import dispatcher
 from django.db import models
 from django.core.exceptions import ImproperlyConfigured
+import re
+import types
+try:
+    import settings_audit
+except ImportError:
+    settings_audit = None
 
 value_error_re = re.compile("^.+'(.+)'$")
 
 class AuditTrail(object):
-    def __init__(self, show_in_admin=False):
-        self.show_in_admin = show_in_admin
+    def __init__(self, show_in_admin=False, save_change_type=True, audit_deletes=True,
+                 track_fields=None):
+        self.opts = {}
+        self.opts['show_in_admin'] = show_in_admin
+        self.opts['save_change_type'] = save_change_type
+        self.opts['audit_deletes'] = audit_deletes
+        if track_fields:
+            self.opts['track_fields'] = track_fields
+        else:
+            self.opts['track_fields'] = []
 
     def contribute_to_class(self, cls, name):
         # This should only get added once the class is otherwise complete
-
         def _contribute(sender):
-            model = create_audit_model(sender, self.show_in_admin)
+            model = create_audit_model(sender, **self.opts)
             descriptor = AuditTrailDescriptor(model._default_manager, sender._meta.pk.attname)
             setattr(sender, name, descriptor)
 
+            def _audit_track(instance, field_arr):
+                field_name = field_arr[0]
+                try:
+                    return getattr(instance, field_name)
+                except:
+                    if len(field_arr) > 2:
+                        if callable(field_arr[2]):
+                            fn = field_arr[2]
+                            return fn(instance)
+                        else:
+                            return field_arr[2]
+
             def _audit(sender, instance):
-                # Write model changes to the audit model
+                # Write model changes to the audit model.
+                # instance is the current (non-audit) model.
                 kwargs = {}
                 for field in sender._meta.fields:
-                    kwargs[field.attname] = getattr(instance, field.attname)
+                    #kwargs[field.attname] = getattr(instance, field.attname)
+                    kwargs[field.name] = getattr(instance, field.name)
+                if self.opts['save_change_type']:
+                    kwargs['_audit_change_type'] = 'U'
+                for field_arr in model._audit_track:
+                    kwargs[field_arr[0]] = _audit_track(instance, field_arr)
                 model._default_manager.create(**kwargs)
             dispatcher.connect(_audit, signal=models.signals.post_save, sender=cls, weak=False)
+
+            if self.opts['audit_deletes']:
+                def _audit_delete(sender, instance):
+                    # Write model changes to the audit model
+                    kwargs = {}
+                    for field in sender._meta.fields:
+                        kwargs[field.name] = getattr(instance, field.name)
+                    if self.opts['save_change_type']:
+                        kwargs['_audit_change_type'] = 'D'
+                    for field_arr in model._audit_track:
+                        kwargs[field_arr[0]] = _audit_track(instance, field_arr)
+                    model._default_manager.create(**kwargs)
+                dispatcher.connect(_audit_delete, signal=models.signals.pre_delete, sender=cls, weak=False)
 
         dispatcher.connect(_contribute, signal=models.signals.class_prepared, sender=cls, weak=False)
@@ -100 +153 @@
     def __get__(self, instance=None, owner=None):
         if instance == None:
-            raise AttributeError, "Audit trail is only accessible via %s instances." % type.__name__
-        return create_audit_manager(self.manager, self.pk_attribute, instance._get_pk_val())
+            #raise AttributeError, "Audit trail is only accessible via %s instances." % type.__name__
+            return create_audit_manager_class(self.manager)
+        else:
+            return create_audit_manager_with_pk(self.manager, self.pk_attribute, instance._get_pk_val())
 
     def __set__(self, instance, value):
         raise AttributeError, "Audit trail may not be edited in this manner."
 
-def create_audit_manager(manager, pk_attribute, pk):
+def create_audit_manager_with_pk(manager, pk_attribute, pk):
+    """Create an audit trail manager based on the current object"""
+    class AuditTrailWithPkManager(manager.__class__):
+        def __init__(self):
+            self.model = manager.model
+
+        def get_query_set(self):
+            return super(AuditTrailWithPkManager, self).get_query_set().filter(**{pk_attribute: pk})
+    return AuditTrailWithPkManager()
+
+def create_audit_manager_class(manager):
     """Create an audit trail manager based on the current object"""
     class AuditTrailManager(manager.__class__):
         def __init__(self):
             self.model = manager.model
-
-        def get_query_set(self):
-            return super(AuditTrailManager, self).get_query_set().filter(**{pk_attribute: pk})
     return AuditTrailManager()
 
-def create_audit_model(cls, show_in_admin):
+def create_audit_model(cls, **kwargs):
     """Create an audit model for the specific class"""
     name = cls.__name__ + 'Audit'
@@ -133 +195 @@
         '_audit__str__': cls.__str__.im_func,
         '__str__': lambda self: '%s as of %s' % (self._audit__str__(), self._audit_timestamp),
+        '_audit_track': _track_fields(track_fields=kwargs['track_fields'], unprocessed=True)
     }
 
-    if show_in_admin:
+    if 'save_change_type' in kwargs and kwargs['save_change_type']:
+        attrs['_audit_change_type'] = models.CharField(maxlength=1)
+
+    if 'show_in_admin' in kwargs and kwargs['show_in_admin']:
         # Enable admin integration
         class Admin:
@@ -143 +209 @@
     # Copy the fields from the existing model to the audit model
     for field in cls._meta.fields:
-        if field.attname in attrs:
+        #if field.attname in attrs:
+        if field.name in attrs:
             raise ImproperlyConfigured, "%s cannot use %s as it is needed by AuditTrail." % (cls.__name__, field.attname)
-        attrs[field.attname] = copy_field(field)
-
+        #attrs[field.attname] = copy_field(field)
+        attrs[field.name] = copy_field(field)
+
+    for track_field in _track_fields(kwargs['track_fields']):
+        if track_field['name'] in attrs:
+            raise NameError('Field named "%s" already exists in audit version of %s' % (track_field['name'], cls.__name__))
+        attrs[track_field['name']] = copy_field(track_field['field'])
+    
     return type(name, (models.Model,), attrs)
 
@@ -169 +242 @@
     while copied_field is None:
         try:
-            copied_field = cls(**kwargs)
+            if isinstance(field, models.ForeignKey):
+                copied_field = cls(field.rel.to, **kwargs)
+            elif isinstance(field, models.OneToOneField):
+                copied_field = models.ForeignKey(field.rel.to, **kwargs)
+            else:
+                copied_field = cls(**kwargs)
         except (TypeError, ValueError), e:
             # Some attributes, like creation_counter, aren't valid arguments
@@ -180 +258 @@
 
     return copied_field
+
+def _build_track_field(track_item):
+    track = {}
+    track['name'] = track_item[0]
+    if isinstance(track_item[1], models.Field):
+        track['field'] = track_item[1]
+    elif issubclass(track_item[1], models.Model):
+        track['field'] = models.ForeignKey(track_item[1])
+    else:
+        raise TypeError('Track fields only support items that are Fields or Models.')
+    return track
+
+def _track_fields(track_fields=None, unprocessed=False):
+    # Add in the fields from the Audit class "track" attribute.
+    tracks_found = []
+    
+    if settings_audit:
+        global_track_fields = getattr(settings_audit, 'GLOBAL_TRACK_FIELDS', [])
+        for track_item in global_track_fields:
+            if unprocessed:
+                tracks_found.append(track_item)
+            else:
+                tracks_found.append(_build_track_field(track_item))
+    
+    if track_fields:
+        for track_item in track_fields:
+            if unprocessed:
+                tracks_found.append(track_item)
+            else:
+                tracks_found.append(_build_track_field(track_item))
+    return tracks_found
 }}}