Support customizing the history manager and historical queryset classes.

docs/historical_model.rst

@@ -179,6 +179,68 @@ IMPORTANT: Setting `custom_model_name` to `lambda x:f'{x}'` is not permitted.
            An error will be generated and no history model created if they are the same.
 
 
+Custom History Manager and Historical QuerySets
+-----------------------------------------------
+
+To manipulate the History Manager or the Historical QuerySet, you can specify the
+``history_manager`` and ``historical_queryset`` options. Tht values must be subclasses
+of ``simple_history.manager.HistoryManager`` and
+``simple_history.manager.HistoricalQuerySet`` respectively.
+
+Keep in mind, you can use either or both of these options. To understand the
+difference between a `Manager` and QuerySet, see the Django `Manager documentation`_.
+
+
+.. code-block:: python
+    from datetime import timedelta
+    from django.db import models
+    from django.utils import timezone
+    from simple_history.models import HistoricalRecords
+    from simple_history.manager import HistoryManager, HistoricalQuerySet
+
+    class HistoryQuestionManager(HistoryManager):
+        def published(self):
+            return self.filter(pub_date__lte=timezone.now())
+
+    class HistoryQuestionQuerySet(HistoricalQuerySet):
+        def question_prefixed(self):
+            return self.filter(question__startswith='Question: ')
+
+    class Question(models.Model):
+        pub_date = models.DateTimeField('date published')
+        history = HistoricalRecords(
+            history_manager=HistoryQuestionManager,
+            historical_queryset=HistoryQuestionQuerySet,
+        )
+
+    queryset = Question.history.published().question_prefixed()
+
+
+To reuse a `QuerySet` from the model, see the following code example:
+
+.. code-block:: python
+    from datetime import timedelta
+    from django.db import models
+    from django.utils import timezone
+    from simple_history.models import HistoricalRecords
+    from simple_history.manager import HistoryManager, HistoricalQuerySet
+
+    class QuestionQuerySet(HistoricalQuerySet):
+        def question_prefixed(self):
+            return self.filter(question__startswith='Question: ')
+
+
+    class HistoryQuestionQuerySet(QuestionQuerySet, HistoricalQuerySet):
+        """Redefine QuerySet with base class HistoricalQuerySet"""
+
+    class Question(models.Model):
+        pub_date = models.DateTimeField('date published')
+        history = HistoricalRecords(historical_queryset=HistoryQuestionQuerySet)
+        manager = models.Manager.from_queryset(HistoryQuestionQuerySet)()
+
+.. _Manager documentation: https://docs.djangoproject.com/en/stable/topics/db/managers/
+
+
 TextField as `history_change_reason`
 ------------------------------------
 

simple_history/manager.py

@@ -119,14 +119,6 @@ def _instanceize(self):
                 setattr(historic, "_as_of", self._as_of)
 
 
-class HistoryDescriptor:
-    def __init__(self, model):
-        self.model = model
-
-    def __get__(self, instance, owner):
-        return HistoryManager.from_queryset(HistoricalQuerySet)(self.model, instance)
-
-
 class HistoryManager(models.Manager):
     def __init__(self, model, instance=None):
         super().__init__()
@@ -272,3 +264,15 @@ def bulk_history_create(
         return self.model.objects.bulk_create(
             historical_instances, batch_size=batch_size
         )
+
+
+class HistoryDescriptor:
+    def __init__(self, model, manager=HistoryManager, queryset=HistoricalQuerySet):
+        self.model = model
+        self.queryset_class = queryset
+        self.manager_class = manager
+
+    def __get__(self, instance, owner):
+        return self.manager_class.from_queryset(self.queryset_class)(
+            self.model, instance
+        )

simple_history/models.py

@@ -31,7 +31,12 @@
 from simple_history import utils
 
 from . import exceptions
-from .manager import SIMPLE_HISTORY_REVERSE_ATTR_NAME, HistoryDescriptor
+from .manager import (
+    SIMPLE_HISTORY_REVERSE_ATTR_NAME,
+    HistoricalQuerySet,
+    HistoryDescriptor,
+    HistoryManager,
+)
 from .signals import (
     post_create_historical_m2m_records,
     post_create_historical_record,
@@ -100,6 +105,8 @@ def __init__(
         user_db_constraint=True,
         no_db_index=list(),
         excluded_field_kwargs=None,
+        history_manager=HistoryManager,
+        historical_queryset=HistoricalQuerySet,
         m2m_fields=(),
         m2m_fields_model_field_name="_history_m2m_fields",
         m2m_bases=(models.Model,),
@@ -122,6 +129,8 @@ def __init__(
         self.user_setter = history_user_setter
         self.related_name = related_name
         self.use_base_model_db = use_base_model_db
+        self.history_manager = history_manager
+        self.historical_queryset = historical_queryset
         self.m2m_fields = m2m_fields
         self.m2m_fields_model_field_name = m2m_fields_model_field_name
 
@@ -215,7 +224,11 @@ def finalize(self, sender, **kwargs):
                 weak=False,
             )
 
-        descriptor = HistoryDescriptor(history_model)
+        descriptor = HistoryDescriptor(
+            history_model,
+            manager=self.history_manager,
+            queryset=self.historical_queryset,
+        )
         setattr(sender, self.manager_name, descriptor)
         sender._meta.simple_history_manager_attribute = self.manager_name
 

simple_history/tests/models.py

@@ -9,6 +9,7 @@
 from django.urls import reverse
 
 from simple_history import register
+from simple_history.manager import HistoricalQuerySet, HistoryManager
 from simple_history.models import HistoricalRecords, HistoricForeignKey
 
 from .custom_user.models import CustomUser as User
@@ -155,6 +156,25 @@ class PollWithManyToManyCustomHistoryID(models.Model):
     )
 
 
+class PollQuerySet(HistoricalQuerySet):
+    def questions(self):
+        return self.filter(question__startswith="Question ")
+
+
+class PollManager(HistoryManager):
+    def low_ids(self):
+        return self.filter(id__lte=3)
+
+
+class PollWithQuerySetCustomizations(models.Model):
+    question = models.CharField(max_length=200)
+    pub_date = models.DateTimeField("date published")
+
+    history = HistoricalRecords(
+        history_manager=PollManager, historical_queryset=PollQuerySet
+    )
+
+
 class HistoricalRecordsWithExtraFieldM2M(HistoricalRecords):
     def get_extra_fields_m2m(self, model, through_model, fields):
         extra_fields = super().get_extra_fields_m2m(model, through_model, fields)

simple_history/tests/tests/test_models.py

@@ -103,6 +103,7 @@
     PollWithManyToManyCustomHistoryID,
     PollWithManyToManyWithIPAddress,
     PollWithNonEditableField,
+    PollWithQuerySetCustomizations,
     PollWithSelfManyToMany,
     PollWithSeveralManyToMany,
     Province,
@@ -800,6 +801,42 @@ def test_history_with_unknown_field(self):
         with self.assertNumQueries(0):
             new_record.diff_against(old_record, excluded_fields=["unknown_field"])
 
+    def test_history_with_custom_queryset(self):
+        PollWithQuerySetCustomizations.objects.create(
+            id=1, pub_date=today, question="Question 1"
+        )
+        PollWithQuerySetCustomizations.objects.create(
+            id=2, pub_date=today, question="Low Id"
+        )
+        PollWithQuerySetCustomizations.objects.create(
+            id=10, pub_date=today, question="Random"
+        )
+
+        self.assertEqual(
+            set(
+                PollWithQuerySetCustomizations.history.low_ids().values_list(
+                    "question", flat=True
+                )
+            ),
+            {"Question 1", "Low Id"},
+        )
+        self.assertEqual(
+            set(
+                PollWithQuerySetCustomizations.history.questions().values_list(
+                    "question", flat=True
+                )
+            ),
+            {"Question 1"},
+        )
+        self.assertEqual(
+            set(
+                PollWithQuerySetCustomizations.history.low_ids()
+                .questions()
+                .values_list("question", flat=True)
+            ),
+            {"Question 1"},
+        )
+
 
 class GetPrevRecordAndNextRecordTestCase(TestCase):
     def assertRecordsMatch(self, record_a, record_b):