Add HistoricForeignKey for chasing relationships through a timepoint.

Add is_historic, to_historic helpers.

Author: jeking3

CHANGES.rst

@@ -25,11 +25,12 @@ Full list of changes:
 - Added pre-commit for better commit quality (gh-896)
 - Added ability to break into debugger on unit test failure (gh-890)
 - Russian translations update (gh-897)
+- Fix bug with ``history.diff_against`` with non-editable fields (gh-923)
+- Added HistoricForeignKey (gh-940)
 - Add Python 3.10 to test matrix (gh-899)
 - Added support for Django 4.0 (gh-898)
-- Dropped support for Python 3.6, which reached end-of-life on 2021-12-23 (gh-946).
-- Fix bug with ``history.diff_against`` with non-editable fields (gh-923)
 - Dropped support for Django 3.1 (gh-952)
+- Dropped support for Python 3.6, which reached end-of-life on 2021-12-23 (gh-946)
 - RecordModels now support a ``no_db_index`` setting, to drop indices in historical models, default stays the same (gh-720)
 
 3.0.0 (2021-04-16)

docs/querying_history.rst

@@ -137,6 +137,31 @@ When the queryset is returning historical records, `pk` refers to the
 `history_id` primary key.
 
 
+is_historic and to_historic
+---------------------------
+
+If you use `as_of` to query history, the resulting instance will have an
+attribute named `_history` added to it.  This property will contain the
+historical model record that the instance was derived from.  Calling
+is_historic is an easy way to check if an instance was derived from a
+historic timepoint (even if it is the most recent version).
+
+You can use `to_historic` to return the historical model that was used
+to furnish the instance at hand, if it is actually historic.
+
+
+HistoricForeignKey
+------------------
+
+If you have two historic tables linked by foreign key, you can change it
+to use a HistoricForeignKey so that chasing relations from an `as_of`
+acquired instance (at a specific timepoint) will honor that timepoint
+when accessing the related object(s).  This works for both forward and
+reverse relationships.
+
+See the `HistoricForeignKeyTest` code and models for an example.
+
+
 most_recent
 -----------
 

simple_history/manager.py

@@ -8,6 +8,10 @@
     get_change_reason_from_object,
 )
 
+# when converting a historical record to an instance, this attribute is added
+# to the instance so that code can reverse the instance to its historical record
+SIMPLE_HISTORY_REVERSE_ATTR_NAME = "_history"
+
 
 class HistoricalQuerySet(QuerySet):
     """
@@ -22,6 +26,7 @@ class HistoricalQuerySet(QuerySet):
     def __init__(self, *args, **kwargs):
         super().__init__(*args, **kwargs)
         self._as_instances = False
+        self._as_of = None
         self._pk_attr = self.model.instance_type._meta.pk.attname
 
     def as_instances(self):
@@ -89,6 +94,7 @@ def latest_of_each(self):
     def _clone(self):
         c = super()._clone()
         c._as_instances = self._as_instances
+        c._as_of = self._as_of
         c._pk_attr = self._pk_attr
         return c
 
@@ -108,6 +114,9 @@ def _instanceize(self):
             and isinstance(self._result_cache[0], self.model)
         ):
             self._result_cache = [item.instance for item in self._result_cache]
+            for item in self._result_cache:
+                historic = getattr(item, SIMPLE_HISTORY_REVERSE_ATTR_NAME)
+                setattr(historic, "_as_of", self._as_of)
 
 
 class HistoryDescriptor:
@@ -195,6 +204,8 @@ def as_of(self, date):
         """
         queryset = self.get_queryset().filter(history_date__lte=date)
         if not self.instance:
+            if isinstance(queryset, HistoricalQuerySet):
+                queryset._as_of = date
             queryset = queryset.latest_of_each().as_instances()
             return queryset
 
@@ -209,7 +220,10 @@ def as_of(self, date):
             raise self.instance.DoesNotExist(
                 "%s had already been deleted." % self.instance._meta.object_name
             )
-        return history_obj.instance
+        result = history_obj.instance
+        historic = getattr(result, SIMPLE_HISTORY_REVERSE_ATTR_NAME)
+        setattr(historic, "_as_of", date)
+        return result
 
     def bulk_history_create(
         self,

simple_history/models.py

@@ -11,17 +11,25 @@
 from django.db import models
 from django.db.models import ManyToManyField
 from django.db.models.fields.proxy import OrderWrt
+from django.db.models.fields.related import ForeignKey
+from django.db.models.fields.related_descriptors import (
+    ForwardManyToOneDescriptor,
+    ReverseManyToOneDescriptor,
+    create_reverse_many_to_one_manager,
+)
+from django.db.models.query import QuerySet
 from django.forms.models import model_to_dict
 from django.urls import reverse
 from django.utils import timezone
 from django.utils.encoding import smart_str
+from django.utils.functional import cached_property
 from django.utils.text import format_lazy
 from django.utils.translation import gettext_lazy as _
 
 from simple_history import utils
 
 from . import exceptions
-from .manager import HistoryDescriptor
+from .manager import SIMPLE_HISTORY_REVERSE_ATTR_NAME, HistoryDescriptor
 from .signals import post_create_historical_record, pre_create_historical_record
 from .utils import get_change_reason_from_object
 
@@ -422,7 +430,10 @@ def get_instance(self):
                     pass
                 else:
                     attrs.update(values)
-            return model(**attrs)
+            result = model(**attrs)
+            # this is the only way external code could know an instance is historical
+            setattr(result, SIMPLE_HISTORY_REVERSE_ATTR_NAME, self)
+            return result
 
         def get_next_record(self):
             """
@@ -632,6 +643,108 @@ def transform_field(field):
         field.serialize = True
 
 
+class HistoricForwardManyToOneDescriptor(ForwardManyToOneDescriptor):
+    """
+    Overrides get_queryset to provide historic query support, should the
+    instance be historic (and therefore was generated by a timepoint query)
+    and the other side of the relation also uses a history manager.
+    """
+
+    def get_queryset(self, **hints) -> QuerySet:
+        instance = hints.get("instance")
+        if instance:
+            history = getattr(instance, SIMPLE_HISTORY_REVERSE_ATTR_NAME, None)
+            histmgr = getattr(
+                self.field.remote_field.model,
+                getattr(
+                    self.field.remote_field.model._meta,
+                    "simple_history_manager_attribute",
+                    "_notthere",
+                ),
+                None,
+            )
+            if history and histmgr:
+                return histmgr.as_of(history._as_of)
+        return super().get_queryset(**hints)
+
+
+class HistoricReverseManyToOneDescriptor(ReverseManyToOneDescriptor):
+    """
+    Overrides get_queryset to provide historic query support, should the
+    instance be historic (and therefore was generated by a timepoint query)
+    and the other side of the relation also uses a history manager.
+    """
+
+    @cached_property
+    def related_manager_cls(self):
+        related_model = self.rel.related_model
+
+        class HistoricRelationModelManager(related_model._default_manager.__class__):
+            def get_queryset(self):
+                try:
+                    return self.instance._prefetched_objects_cache[
+                        self.field.remote_field.get_cache_name()
+                    ]
+                except (AttributeError, KeyError):
+                    history = getattr(
+                        self.instance, SIMPLE_HISTORY_REVERSE_ATTR_NAME, None
+                    )
+                    histmgr = getattr(
+                        self.model,
+                        getattr(
+                            self.model._meta,
+                            "simple_history_manager_attribute",
+                            "_notthere",
+                        ),
+                        None,
+                    )
+                    if history and histmgr:
+                        queryset = histmgr.as_of(history._as_of)
+                    else:
+                        queryset = super().get_queryset()
+                    return self._apply_rel_filters(queryset)
+
+        return create_reverse_many_to_one_manager(
+            HistoricRelationModelManager, self.rel
+        )
+
+
+class HistoricForeignKey(ForeignKey):
+    """
+    Allows foreign keys to work properly from a historic instance.
+
+    If you use as_of queries to extract historical instances from
+    a model, and you have other models that are related by foreign
+    key and also historic, changing them to a HistoricForeignKey
+    field type will allow you to naturally cross the relationship
+    boundary at the same point in time as the origin instance.
+
+    A historic instance maintains an attribute ("_historic") when
+    it is historic, holding the historic record instance and the
+    timepoint used to query it ("_as_of").  HistoricForeignKey
+    looks for this and uses an as_of query against the related
+    object so the relationship is assessed at the same timepoint.
+    """
+
+    forward_related_accessor_class = HistoricForwardManyToOneDescriptor
+    related_accessor_class = HistoricReverseManyToOneDescriptor
+
+
+def is_historic(instance):
+    """
+    Returns True if the instance was acquired with an as_of timepoint.
+    """
+    return to_historic(instance) is not None
+
+
+def to_historic(instance):
+    """
+    Returns a historic model instance if the instance was acquired with
+    an as_of timepoint, or None.
+    """
+    return getattr(instance, SIMPLE_HISTORY_REVERSE_ATTR_NAME, None)
+
+
 class HistoricalObjectDescriptor:
     def __init__(self, model, fields_included):
         self.model = model

simple_history/tests/models.py

@@ -4,10 +4,12 @@
 from django.apps import apps
 from django.conf import settings
 from django.db import models
+from django.db.models.deletion import CASCADE
+from django.db.models.fields.related import ForeignKey
 from django.urls import reverse
 
 from simple_history import register
-from simple_history.models import HistoricalRecords
+from simple_history.models import HistoricalRecords, HistoricForeignKey
 
 from .custom_user.models import CustomUser as User
 from .external.models import AbstractExternal, AbstractExternal2, AbstractExternal3
@@ -788,3 +790,67 @@ class ModelWithMultipleNoDBIndex(models.Model):
         "Library", on_delete=models.CASCADE, null=True, related_name="+"
     )
     history = HistoricalRecords(no_db_index=["name", "fk", "other"])
+
+
+class TestOrganization(models.Model):
+    name = models.CharField(max_length=15, unique=True)
+
+
+class TestOrganizationWithHistory(models.Model):
+    name = models.CharField(max_length=15, unique=True)
+    history = HistoricalRecords()
+
+
+class TestParticipantToHistoricOrganization(models.Model):
+    """
+    Non-historic table foreign key to historic table.
+
+    In this case it should simply behave like ForeignKey because
+    the origin model (this one) cannot be historic, so foreign key
+    lookups are always "current".
+    """
+
+    name = models.CharField(max_length=15, unique=True)
+    organization = HistoricForeignKey(
+        TestOrganizationWithHistory, on_delete=CASCADE, related_name="participants"
+    )
+
+
+class TestHistoricParticipantToOrganization(models.Model):
+    """
+    Historic table foreign key to non-historic table.
+
+    In this case it should simply behave like ForeignKey because
+    the origin model (this one) can be historic but the target model
+    is not, so foreign key lookups are always "current".
+    """
+
+    name = models.CharField(max_length=15, unique=True)
+    organization = HistoricForeignKey(
+        TestOrganization, on_delete=CASCADE, related_name="participants"
+    )
+    history = HistoricalRecords()
+
+
+class TestHistoricParticipanToHistoricOrganization(models.Model):
+    """
+    Historic table foreign key to historic table.
+
+    In this case as_of queries on the origin model (this one)
+    or on the target model (the other one) will traverse the
+    foreign key relationship honoring the timepoint of the
+    original query.  This only happens when both tables involved
+    are historic.
+
+    NOTE: related_name has to be different than the one used in
+          TestParticipantToHistoricOrganization as they are
+          sharing the same target table.
+    """
+
+    name = models.CharField(max_length=15, unique=True)
+    organization = HistoricForeignKey(
+        TestOrganizationWithHistory,
+        on_delete=CASCADE,
+        related_name="historic_participants",
+    )
+    history = HistoricalRecords()

simple_history/tests/tests/test_manager.py

@@ -5,6 +5,8 @@
 from django.db import IntegrityError
 from django.test import TestCase, override_settings, skipUnlessDBFeature
 
+from simple_history.manager import SIMPLE_HISTORY_REVERSE_ATTR_NAME
+
 from ..models import Document, Poll, RankedDocument
 
 User = get_user_model()
@@ -69,12 +71,23 @@ def test_modified(self):
 
 class AsOfAdditionalTestCase(TestCase):
     def test_create_and_delete(self):
-        now = datetime.now()
         document = Document.objects.create()
+        now = datetime.now()
         document.delete()
-        for doc_change in Document.history.all():
-            doc_change.history_date = now
-            doc_change.save()
+
+        docs_as_of_now = Document.history.as_of(now)
+        doc = docs_as_of_now[0]
+        # as_of queries inject a property allowing callers
+        # to go from instance to historical instance
+        historic = getattr(doc, SIMPLE_HISTORY_REVERSE_ATTR_NAME)
+        self.assertIsNotNone(historic)
+        # as_of queries inject the time point of the original
+        # query into the historic record so callers can do magical
+        # things like chase historic foreign key relationships
+        # by patching forward and reverse one-to-one relationship
+        # processing (see issue 880)
+        self.assertEqual(historic._as_of, now)
+
         docs_as_of_tmw = Document.history.as_of(now + timedelta(days=1))
         with self.assertNumQueries(1):
             self.assertFalse(list(docs_as_of_tmw))
@@ -122,6 +135,12 @@ def test_as_of(self):
         ids = {item["id"] for item in queryset.values("id")}
         self.assertEqual(ids, {document1.id, document2.id})
 
+        # these records are historic
+        record = queryset[0]
+        historic = getattr(record, SIMPLE_HISTORY_REVERSE_ATTR_NAME)
+        self.assertIsInstance(historic, RankedDocument.history.model)
+        self.assertEqual(historic._as_of, t1)
+
         # at t2 we have one record left
         queryset = RankedDocument.history.as_of(t2)
         self.assertEqual(queryset.count(), 1)