Allow alternative user model for tracking history_user (#371)

* Allow alternative user model for tracking history_user

* Add support for custom get_user option

* Re-order arguments to keep original order for existing arguments

* Rename test classes to be more descriptive

* Remove history argument for get_user and fix typo

Author: rwlogel

CHANGES.rst

@@ -1,6 +1,10 @@
 Changes
 =======
 
+Unreleased
+----------
+- Add ability to specify alternative user_model for tracking
+
 2.1.1 (2018-06-15)
 ------------------
 - Fixed out-of-memory exception when running populate_history management command (gh-408)

docs/advanced.rst

@@ -140,7 +140,65 @@ referencing the ``changed_by`` field:
         def _history_user(self, value):
             self.changed_by = value
 
-Admin integration requires that you use a ``_history_user.setter`` attribute with your custom ``_history_user`` property (see :ref:`admin_integration`).
+Admin integration requires that you use a ``_history_user.setter`` attribute with
+your custom ``_history_user`` property (see :ref:`admin_integration`).
+
+Another option for identifying the change user is by providing a function via ``get_user``.
+If provided it will be called everytime that the ``history_user`` needs to be
+identified with the following key word arguments:
+
+* ``instance``:  The current instance being modified
+* ``request``:  If using the middleware the current request object will be provided if they are authenticated.
+
+This is very helpful when using ``register``:
+
+.. code-block:: python
+
+    from django.db import models
+    from simple_history.models import HistoricalRecords
+
+    class Poll(models.Model):
+        question = models.CharField(max_length=200)
+        pub_date = models.DateTimeField('date published')
+        changed_by = models.ForeignKey('auth.User')
+
+
+    def get_poll_user(instance, **kwargs):
+        return instance.changed_by
+
+    register(Poll, get_user=get_poll_user)
+
+
+Change User Model
+------------------------------------
+
+If you need to use a different user model then ``settings.AUTH_USER_MODEL``,
+pass in the required model to ``user_model``.  Doing this requires ``_history_user``
+or ``get_user`` is provided as detailed above.
+
+.. code-block:: python
+
+    from django.db import models
+    from simple_history.models import HistoricalRecords
+
+    class PollUser(models.Model):
+        user_id = models.ForeignKey('auth.User')
+
+
+    # Only PollUsers should be modifying a Poll
+    class Poll(models.Model):
+        question = models.CharField(max_length=200)
+        pub_date = models.DateTimeField('date published')
+        changed_by = models.ForeignKey(PollUser)
+        history = HistoricalRecords(user_model=PollUser)
+
+        @property
+        def _history_user(self):
+            return self.changed_by
+
+        @_history_user.setter
+        def _history_user(self, value):
+            self.changed_by = value
 
 Custom ``history_id``
 ---------------------

simple_history/__init__.py

@@ -29,6 +29,6 @@ def register(
     records.manager_name = manager_name
     records.table_name = table_name
     records.module = app and ("%s.models" % app) or model.__module__
+    records.cls = model
     records.add_extra_methods(model)
     records.finalize(model)
-    models.registered_models[model._meta.db_table] = model

simple_history/models.py

@@ -24,20 +24,29 @@
 registered_models = {}
 
 
+def default_get_user(request, **kwargs):
+    try:
+        return request.user
+    except AttributeError:
+        return None
+
+
 class HistoricalRecords(object):
     thread = threading.local()
 
     def __init__(self, verbose_name=None, bases=(models.Model,),
                  user_related_name='+', table_name=None, inherit=False,
-                 excluded_fields=None,
-                 history_id_field=None,
-                 history_change_reason_field=None):
+                 excluded_fields=None, history_id_field=None,
+                 history_change_reason_field=None,
+                 user_model=None, get_user=default_get_user):
         self.user_set_verbose_name = verbose_name
         self.user_related_name = user_related_name
         self.table_name = table_name
         self.inherit = inherit
         self.history_id_field = history_id_field
         self.history_change_reason_field = history_change_reason_field
+        self.user_model = user_model
+        self.get_user = get_user
         if excluded_fields is None:
             excluded_fields = []
         self.excluded_fields = excluded_fields
@@ -74,15 +83,11 @@ def save_without_historical_record(self, *args, **kwargs):
 
     def finalize(self, sender, **kwargs):
         inherited = False
-        try:
-            hint_class = self.cls
-        except AttributeError:  # called via `register`
-            pass
-        else:
-            if hint_class is not sender:  # set in concrete
-                inherited = (self.inherit and issubclass(sender, hint_class))
-                if not inherited:
-                    return  # set in abstract
+        if self.cls is not sender:  # set in concrete
+            inherited = (self.inherit and issubclass(sender, self.cls))
+            if not inherited:
+                return  # set in abstract
+
         if hasattr(sender._meta, 'simple_history_manager_attribute'):
             raise exceptions.MultipleRegistrationsError(
                 '{}.{} registered multiple times for history tracking.'.format(
@@ -207,7 +212,9 @@ def copy_fields(self, model):
     def get_extra_fields(self, model, fields):
         """Return dict of extra fields added to the historical record model"""
 
-        user_model = getattr(settings, 'AUTH_USER_MODEL', 'auth.User')
+        user_model = self.user_model or getattr(
+            settings, 'AUTH_USER_MODEL', 'auth.User'
+        )
 
         def revert_url(self):
             """URL for this change in the default admin site."""
@@ -349,12 +356,14 @@ def get_history_user(self, instance):
         try:
             return instance._history_user
         except AttributeError:
+            request = None
             try:
                 if self.thread.request.user.is_authenticated:
-                    return self.thread.request.user
-                return None
+                    request = self.thread.request
             except AttributeError:
-                return None
+                pass
+
+        return self.get_user(instance=instance, request=request)
 
 
 def transform_field(field):

simple_history/tests/models.py

@@ -415,6 +415,71 @@ class InheritTracking4(TrackedAbstractBaseA):
     pass
 
 
+class BucketMember(models.Model):
+    name = models.CharField(max_length=30)
+    user = models.OneToOneField(
+        User,
+        related_name="bucket_member",
+        on_delete=models.CASCADE
+    )
+
+
+class BucketData(models.Model):
+    changed_by = models.ForeignKey(
+        BucketMember,
+        on_delete=models.SET_NULL,
+        null=True, blank=True,
+    )
+    history = HistoricalRecords(user_model=BucketMember)
+
+    @property
+    def _history_user(self):
+        return self.changed_by
+
+
+def get_bucket_member_changed_by(instance, **kwargs):
+    try:
+        return instance.changed_by
+    except AttributeError:
+        return None
+
+
+class BucketDataRegisterChangedBy(models.Model):
+    changed_by = models.ForeignKey(
+        BucketMember,
+        on_delete=models.SET_NULL,
+        null=True, blank=True,
+    )
+
+
+register(
+    BucketDataRegisterChangedBy,
+    user_model=BucketMember,
+    get_user=get_bucket_member_changed_by
+)
+
+
+def get_bucket_member_request_user(request, **kwargs):
+    try:
+        return request.user.bucket_member
+    except AttributeError:
+        return None
+
+
+class BucketDataRegisterRequestUser(models.Model):
+    data = models.CharField(max_length=30)
+
+    def get_absolute_url(self):
+        return reverse('bucket_data-detail', kwargs={'pk': self.pk})
+
+
+register(
+    BucketDataRegisterRequestUser,
+    user_model=BucketMember,
+    get_user=get_bucket_member_request_user
+)
+
+
 class UUIDModel(models.Model):
     id = models.UUIDField(primary_key=True, default=uuid.uuid4, editable=False)
     history = HistoricalRecords(

simple_history/tests/tests/test_admin.py

@@ -16,6 +16,8 @@
 from simple_history.tests.tests.utils import middleware_override_settings
 from ..models import (
     Book,
+    BucketData,
+    BucketMember,
     Choice,
     ConcreteExternal,
     Employee,
@@ -286,7 +288,7 @@ def test_other_admin(self):
         change_url = get_history_url(state, 0, site="other_admin")
         self.app.get(change_url)
 
-    def test_deleteting_user(self):
+    def test_deleting_user(self):
         """Test deletes of a user does not cascade delete the history"""
         self.login()
         poll = Poll(question="why?", pub_date=today)
@@ -301,6 +303,21 @@ def test_deleteting_user(self):
         historical_poll = poll.history.all()[0]
         self.assertEqual(historical_poll.history_user, None)
 
+    def test_deleteting_member(self):
+        """Test deletes of a BucketMember doesn't cascade delete the history"""
+        self.login()
+        member = BucketMember.objects.create(name="member1", user=self.user)
+        bucket_data = BucketData(changed_by=member)
+        bucket_data.save()
+
+        historical_poll = bucket_data.history.all()[0]
+        self.assertEqual(historical_poll.history_user, member)
+
+        member.delete()
+
+        historical_poll = bucket_data.history.all()[0]
+        self.assertEqual(historical_poll.history_user, None)
+
     def test_missing_one_to_one(self):
         """A relation to a missing one-to-one model should still show
         history"""

simple_history/tests/tests/test_middleware.py

@@ -4,7 +4,11 @@
 from django.urls import reverse
 
 from simple_history.tests.custom_user.models import CustomUser
-from simple_history.tests.models import Poll
+from simple_history.tests.models import (
+    BucketDataRegisterRequestUser,
+    BucketMember,
+    Poll
+)
 from simple_history.tests.tests.utils import middleware_override_settings
 
 
@@ -160,3 +164,18 @@ def test_user_is_not_set_on_delete_view_when_not_logged_in(self):
 
         self.assertListEqual([ph.history_user_id for ph in poll_history],
                              [None, None])
+
+    def test_bucket_member_is_set_on_create_view_when_logged_in(self):
+        self.client.force_login(self.user)
+        member1 = BucketMember.objects.create(name="member1", user=self.user)
+        data = {
+            'data': 'Test Data',
+        }
+        self.client.post(reverse('bucket_data-add'), data=data)
+        bucket_datas = BucketDataRegisterRequestUser.objects.all()
+        self.assertEqual(bucket_datas.count(), 1)
+
+        history = bucket_datas.first().history.all()
+
+        self.assertListEqual([h.history_user_id for h in history],
+                             [member1.id])

simple_history/tests/tests/test_models.py

@@ -20,6 +20,9 @@
     AdminProfile,
     Book,
     Bookcase,
+    BucketData,
+    BucketDataRegisterChangedBy,
+    BucketMember,
     Choice,
     City,
     ConcreteAttr,
@@ -375,6 +378,34 @@ def test_model_with_excluded_fields(self):
         self.assertIn('question', all_fields_names)
         self.assertNotIn('pub_date', all_fields_names)
 
+    def test_user_model_override(self):
+        user1 = User.objects.create_user('user1', '[email protected]')
+        user2 = User.objects.create_user('user2', '[email protected]')
+        member1 = BucketMember.objects.create(name="member1", user=user1)
+        member2 = BucketMember.objects.create(name="member2", user=user2)
+        bucket_data = BucketData.objects.create(changed_by=member1)
+        bucket_data.changed_by = member2
+        bucket_data.save()
+        bucket_data.changed_by = None
+        bucket_data.save()
+        self.assertEqual([d.history_user for d in bucket_data.history.all()],
+                         [None, member2, member1])
+
+    def test_user_model_override_registered(self):
+        user1 = User.objects.create_user('user1', '[email protected]')
+        user2 = User.objects.create_user('user2', '[email protected]')
+        member1 = BucketMember.objects.create(name="member1", user=user1)
+        member2 = BucketMember.objects.create(name="member2", user=user2)
+        bucket_data = BucketDataRegisterChangedBy.objects.create(
+            changed_by=member1
+        )
+        bucket_data.changed_by = member2
+        bucket_data.save()
+        bucket_data.changed_by = None
+        bucket_data.save()
+        self.assertEqual([d.history_user for d in bucket_data.history.all()],
+                         [None, member2, member1])
+
     def test_uuid_history_id(self):
         entry = UUIDModel.objects.create()
 

simple_history/tests/urls.py

@@ -4,6 +4,8 @@
 from django.contrib import admin
 
 from simple_history.tests.view import (
+    BucketDataRegisterRequestUserCreate,
+    BucketDataRegisterRequestUserDetail,
     PollCreate,
     PollDelete,
     PollDetail,
@@ -17,6 +19,12 @@
 urlpatterns = [
     url(r'^admin/', admin.site.urls),
     url(r'^other-admin/', other_admin.site.urls),
+    url(r'^bucket_data/add/$',
+        BucketDataRegisterRequestUserCreate.as_view(),
+        name='bucket_data-add'),
+    url(r'^bucket_data/(?P<pk>[0-9]+)/$',
+        BucketDataRegisterRequestUserDetail.as_view(),
+        name='bucket_data-detail'),
     url(r'^poll/add/$', PollCreate.as_view(), name='poll-add'),
     url(r'^poll/(?P<pk>[0-9]+)/$', PollUpdate.as_view(), name='poll-update'),
     url(r'^poll/(?P<pk>[0-9]+)/delete/$', PollDelete.as_view(),