From 31af519d03171b361cacce0b687494b805396d83 Mon Sep 17 00:00:00 2001 From: Blaine Jester Date: Wed, 15 Apr 2026 11:44:29 -0700 Subject: [PATCH 1/5] Create base source class to share between serializatin and deserialization --- morango/sync/stream/source.py | 104 ++++++++++++++++++ tests/testapp/tests/sync/stream/test_core.py | 24 ++++ .../tests/sync/stream/test_serialize.py | 10 +- .../testapp/tests/sync/stream/test_source.py | 50 +++++++++ 4 files changed, 186 insertions(+), 2 deletions(-) create mode 100644 morango/sync/stream/source.py create mode 100644 tests/testapp/tests/sync/stream/test_source.py diff --git a/morango/sync/stream/source.py b/morango/sync/stream/source.py new file mode 100644 index 00000000..29f6d220 --- /dev/null +++ b/morango/sync/stream/source.py @@ -0,0 +1,104 @@ +import abc +from typing import Generator +from typing import Iterator +from typing import Optional +from typing import TypeVar + +from morango.models.certificates import Filter +from morango.sync.stream.core import Source + + +class SourceTask(abc.ABC): + """Typing for source object passed through streaming pipeline""" + + __slots__ = () + + @property + @abc.abstractmethod + def id(self) -> str: + pass + + +T = TypeVar("T", bound=SourceTask) + + +class MorangoSource(Source[T], abc.ABC): + """ + Common source functionality for Morango sources, such as SyncableModels and Store records. + """ + + def __init__( + self, + profile: str, + sync_filter: Optional[Filter] = None, + dirty_only: bool = True, + partition_order: str = "asc", + ): + """ + :param profile: The Morango model profile + :param sync_filter: The Filter object for this sync + :param dirty_only: Whether to filter on dirty records only + :param partition_order: Controls how the filter specificity is applied, "asc" or "desc" + """ + self.profile = profile + self.sync_filter = sync_filter + self.dirty_only = dirty_only + self.partition_order = partition_order + self._seen: Optional[set] = None + + def begin(self) -> None: + """Initialize seen set at the beginning of the stream""" + self._seen = set() + + def prefix_conditions(self) -> Generator[Optional[str], None, None]: + """ + Generates partition prefixes for queries based on the sync filter and partition order. + + This method outputs prefixes in sorted order according to the specified partition + order. If no sync filter is provided, it yields `None` to indicate a query + without filtering by partition. + + :return: A generator yielding partition prefixes or `None` if no filtering is applied. + """ + if self.sync_filter is None: + # yield None once, so we do one query without a partition filter (everything) + yield None + else: + partitions_prefixes = [str(prefix) for prefix in self.sync_filter] + partition_iterator = sorted( + partitions_prefixes, + reverse=self.partition_order == "desc", + ) + + for prefix in partition_iterator: + yield prefix + + def stream(self) -> Generator[T, None, None]: + """ + Streams unique objects based on prefix conditions. This generator method iterates over + partition conditions defined in the sync_filter and passes through to `stream_for_filter` + to stream back objects, ensuring that only objects with unique `id` values are yielded. + + :return: A generator yielding unique objects. + """ + for partition_condition in self.prefix_conditions(): + for obj in self.stream_for_filter(partition_condition): + # partition filtering could result in overlaps, and since we're walking + # through the partitions one by one, we should avoid duplicates. Morango + # syncable models and store records have unique IDs across the entire profile + if obj.id not in self._seen: + # without sync filters, we do not need to worry about repeating objects + if self.sync_filter is not None: + self._seen.add(obj.id) + yield obj + + @abc.abstractmethod + def stream_for_filter(self, partition_condition: Optional[str]) -> Iterator[T]: + """ + This method is intended to generate an iterator that yields data based on the given + filtering condition. + + :param partition_condition: A string representing a partition filter prefix condition + :return: An iterator yielding items + """ + pass diff --git a/tests/testapp/tests/sync/stream/test_core.py b/tests/testapp/tests/sync/stream/test_core.py index 91394475..31e0b817 100644 --- a/tests/testapp/tests/sync/stream/test_core.py +++ b/tests/testapp/tests/sync/stream/test_core.py @@ -10,6 +10,12 @@ class FakeSource(Source): + def __init__(self): + self.begin_count = 0 + + def begin(self): + self.begin_count += 1 + def stream(self): yield 1 yield 2 @@ -39,6 +45,15 @@ def test_stream(self): source = FakeSource() self.assertEqual([1, 2, 3], list(source.stream())) + def test_begin_defaults_to_a_noop(self): + """Subclasses only override `begin` when they have stream state to initialize""" + + class NoBeginSource(Source): + def stream(self): + yield 1 + + self.assertIsNone(NoBeginSource().begin()) + def test_pipe(self): source = FakeSource() transform = FakeTransform() @@ -122,6 +137,15 @@ def test_pipeline_execution(self): self.assertEqual(count, 3) self.assertEqual([2, 4, 6], sink.consumed) + def test_pipeline_begins_the_source(self): + """`end` is what initializes source stream state, before anything is pulled from it""" + source = FakeSource() + pipeline = source.pipe(FakeTransform()) + + self.assertEqual(source.begin_count, 0) + pipeline.end(FakeSink()) + self.assertEqual(source.begin_count, 1) + def test_pipeline_chaining(self): source = FakeSource() pipeline = source.pipe(FakeTransform()).pipe(FakeTransform()) diff --git a/tests/testapp/tests/sync/stream/test_serialize.py b/tests/testapp/tests/sync/stream/test_serialize.py index daea5e33..66e0be7d 100644 --- a/tests/testapp/tests/sync/stream/test_serialize.py +++ b/tests/testapp/tests/sync/stream/test_serialize.py @@ -89,14 +89,20 @@ def test_stream__seen_once(self, mock_get_model_querysets): qs.filter.return_value = qs qs.iterator.return_value = [obj, obj] - source = AppModelSource(profile="test") + source = AppModelSource(profile="test", sync_filter=Filter("a")) tasks = list(source.stream()) self.assertEqual(len(tasks), 1) self.assertEqual(tasks[0].model, model) self.assertEqual(tasks[0].obj, obj) mock_get_model_querysets.assert_called_once_with("test") - qs.filter.assert_called_once_with(_morango_dirty_bit=True) + self.assertEqual( + qs.filter.mock_calls, + [ + mock.call(_morango_partition__startswith="a"), + mock.call(_morango_dirty_bit=True), + ], + ) @mock.patch("morango.sync.stream.serialize.syncable_models.get_model_querysets") def test_stream__partition(self, mock_get_model_querysets): diff --git a/tests/testapp/tests/sync/stream/test_source.py b/tests/testapp/tests/sync/stream/test_source.py new file mode 100644 index 00000000..606a2162 --- /dev/null +++ b/tests/testapp/tests/sync/stream/test_source.py @@ -0,0 +1,50 @@ +from django.test import SimpleTestCase + +from morango.models.certificates import Filter +from morango.sync.stream.source import MorangoSource +from morango.sync.stream.source import SourceTask + + +class FakeTask(SourceTask): + __slots__ = ("_id",) + + def __init__(self, task_id): + self._id = task_id + + @property + def id(self): + return self._id + + +class FakeSource(MorangoSource[FakeTask]): + def __init__(self, *args, task_ids=(), **kwargs): + super().__init__(*args, **kwargs) + self.task_ids = task_ids + + def stream_for_filter(self, partition_condition): + for task_id in self.task_ids: + yield FakeTask(task_id) + + +class MorangoSourceBeginTestCase(SimpleTestCase): + def test_seen_is_uninitialized_before_begin(self): + source = FakeSource("test") + self.assertIsNone(source._seen) + + def test_begin_initializes_seen(self): + source = FakeSource("test") + source.begin() + self.assertEqual(source._seen, set()) + + def test_begin_resets_seen_between_runs(self): + """A source may be streamed more than once, and must not carry state across runs""" + source = FakeSource("test", sync_filter=Filter("a"), task_ids=("1", "2")) + + source.begin() + first = [task.id for task in source.stream()] + self.assertEqual(first, ["1", "2"]) + + # without `begin`, every id is already in `_seen` and nothing would be yielded + source.begin() + second = [task.id for task in source.stream()] + self.assertEqual(second, ["1", "2"]) From f4ca9bf953e9fb080bf0b97cbd1ef8b287661e67 Mon Sep 17 00:00:00 2001 From: Blaine Jester Date: Wed, 15 Apr 2026 11:45:26 -0700 Subject: [PATCH 2/5] Refactor serialize source to use new base class --- morango/sync/stream/serialize.py | 70 +++++-------------- .../tests/sync/stream/test_serialize.py | 8 ++- 2 files changed, 23 insertions(+), 55 deletions(-) diff --git a/morango/sync/stream/serialize.py b/morango/sync/stream/serialize.py index 65b8ebd2..e8a0c6d2 100644 --- a/morango/sync/stream/serialize.py +++ b/morango/sync/stream/serialize.py @@ -6,7 +6,6 @@ from typing import Type from django.core.serializers.json import DjangoJSONEncoder -from django.db.models import Q from morango.models.certificates import Filter from morango.models.core import DatabaseMaxCounter @@ -19,14 +18,15 @@ from morango.registry import syncable_models from morango.sync.stream.core import Buffer from morango.sync.stream.core import Sink -from morango.sync.stream.core import Source from morango.sync.stream.core import Transform from morango.sync.stream.core import Unbuffer +from morango.sync.stream.source import MorangoSource +from morango.sync.stream.source import SourceTask logger = logging.getLogger(__name__) -class SerializeTask(object): +class SerializeTask(SourceTask): """Carrier class for providing context through the pipeline""" __slots__ = ( @@ -35,7 +35,6 @@ class SerializeTask(object): "store", "counter", "_self_ref_fk_value", - "_self_ref_fk_value", "_self_ref_order", ) @@ -47,6 +46,10 @@ def __init__(self, model: Type[SyncableModel], obj: SyncableModel): self._self_ref_fk_value: Optional[str] = None self._self_ref_order: Optional[int] = None + @property + def id(self) -> str: + return self.obj.id + @property def is_store_update(self): return self.store is not None and not self.store._state.adding @@ -80,59 +83,22 @@ def set_self_ref_order(self, value: Optional[int]): self._self_ref_order = value -class AppModelSource(Source[SerializeTask]): +class AppModelSource(MorangoSource[SerializeTask]): """ Yields ``SerializeTask`` objects for every syncable-model record that matches the optional *sync_filter*. """ - def __init__( - self, - profile: str, - sync_filter: Optional[Filter] = None, - dirty_only: bool = True, - partition_order: str = "asc", - ): - """ - :param profile: The Morango model profile - :param sync_filter: The Filter object for this sync - :param dirty_only: Whether to filter on dirty records only - :param partition_order: Controls how the filter specificity is applied, "asc" or "desc" - """ - self.profile = profile - self.sync_filter = sync_filter - self.dirty_only = dirty_only - self.partition_order = partition_order - self._seen = set() - - def prefix_conditions(self) -> Generator[Optional[Q], None, None]: - if self.sync_filter is None: - # yield None once, so we do one query without a partition filter (everything) - yield None - else: - partitions_prefixes = [str(prefix) for prefix in self.sync_filter] - partition_iterator = sorted( - partitions_prefixes, - reverse=self.partition_order == "desc", - ) - - for prefix in partition_iterator: - yield Q(_morango_partition__startswith=prefix) - - def stream(self) -> Generator[SerializeTask, None, None]: - for partition_condition in self.prefix_conditions(): - for qs in syncable_models.get_model_querysets(self.profile): - if partition_condition is not None: - qs = qs.filter(partition_condition) - if self.dirty_only: - qs = qs.filter(_morango_dirty_bit=True) - for obj in qs.iterator(): - # partition filtering could result in overlaps, and since we're walking - # through the partitions one by one, we should avoid duplicates. Morango - # syncable models have unique IDs across the entire profile - if obj.id not in self._seen: - self._seen.add(obj.id) - yield SerializeTask(qs.model, obj) + def stream_for_filter( + self, partition_condition: Optional[str] + ) -> Generator[SerializeTask, None, None]: + for qs in syncable_models.get_model_querysets(self.profile): + if partition_condition is not None: + qs = qs.filter(_morango_partition__startswith=partition_condition) + if self.dirty_only: + qs = qs.filter(_morango_dirty_bit=True) + for obj in qs.iterator(): + yield SerializeTask(qs.model, obj) class StoreLookup(Transform[List[SerializeTask]]): diff --git a/tests/testapp/tests/sync/stream/test_serialize.py b/tests/testapp/tests/sync/stream/test_serialize.py index 66e0be7d..e1826e74 100644 --- a/tests/testapp/tests/sync/stream/test_serialize.py +++ b/tests/testapp/tests/sync/stream/test_serialize.py @@ -2,7 +2,6 @@ import uuid import mock -from django.db.models import Q from django.test import SimpleTestCase from django.test import TestCase @@ -60,7 +59,7 @@ def test_prefix_conditions__with_filter(self): source = AppModelSource(profile="test", sync_filter=sync_filter) conditions = list(source.prefix_conditions()) self.assertEqual(len(conditions), 2) - self.assertEqual(str(conditions[0]), "(AND: ('_morango_partition__startswith', 'a'))") + self.assertEqual(str(conditions[0]), "a") @mock.patch("morango.sync.stream.serialize.syncable_models.get_model_querysets") def test_stream__no_partition(self, mock_get_model_querysets): @@ -72,6 +71,7 @@ def test_stream__no_partition(self, mock_get_model_querysets): qs.iterator.return_value = [obj] source = AppModelSource(profile="test") + source.begin() tasks = list(source.stream()) self.assertEqual(len(tasks), 1) @@ -90,6 +90,7 @@ def test_stream__seen_once(self, mock_get_model_querysets): qs.iterator.return_value = [obj, obj] source = AppModelSource(profile="test", sync_filter=Filter("a")) + source.begin() tasks = list(source.stream()) self.assertEqual(len(tasks), 1) @@ -114,13 +115,14 @@ def test_stream__partition(self, mock_get_model_querysets): qs.iterator.return_value = [obj, obj] source = AppModelSource(profile="test", sync_filter=Filter("a"), dirty_only=False) + source.begin() tasks = list(source.stream()) self.assertEqual(len(tasks), 1) self.assertEqual(tasks[0].model, model) self.assertEqual(tasks[0].obj, obj) mock_get_model_querysets.assert_called_once_with("test") - qs.filter.assert_called_once_with(Q(_morango_partition__startswith="a")) + qs.filter.assert_called_once_with(_morango_partition__startswith="a") class StoreLookupTestCase(SimpleTestCase): From 1323ee739e64f9695fd1cb4efd799fe846ae5cb9 Mon Sep 17 00:00:00 2001 From: Blaine Jester Date: Tue, 25 Aug 2026 11:39:43 -0700 Subject: [PATCH 3/5] Add Store queryset method for filtering on deserialization errors --- morango/models/core.py | 27 +++++++++-- tests/testapp/tests/models/test_core.py | 63 +++++++++++++++++++++++++ 2 files changed, 87 insertions(+), 3 deletions(-) diff --git a/morango/models/core.py b/morango/models/core.py index b32f35b3..76a46c77 100644 --- a/morango/models/core.py +++ b/morango/models/core.py @@ -23,6 +23,7 @@ from django.db.models.expressions import CombinedExpression from django.db.models.fields.related import ForeignKey from django.db.models.functions import Cast +from django.db.models.functions import NullIf from django.utils import timezone from django.utils.functional import cached_property @@ -440,10 +441,30 @@ def char_ids_list(self): .values_list("fixed_id", flat=True) ) + def filter_deserialization_error(self, has_error: bool) -> "StoreQueryset": + """ + Filters the queryset to return Store records that have or have not any + deserialization errors + """ + # this nullIf assertion is generally more performant than an OR'd statement on + # unindexed columns, in both SQLite and PostgreSQL + return self.annotate( + _deserialization_error=NullIf( + F("deserialization_error"), Value(""), output_field=models.TextField() + ) + ).filter(_deserialization_error__isnull=not has_error) + + def filter_has_deserialization_error(self) -> "StoreQueryset": + """Filters the queryset to return Store records that have deserialization errors""" + return self.filter_deserialization_error(True) + + def exclude_has_deserialization_error(self) -> "StoreQueryset": + """Filters the queryset to return Store records that have no deserialization error""" + return self.filter_deserialization_error(False) + -class StoreManager(models.Manager): - def get_queryset(self): - return StoreQueryset(self.model, using=self._db) +class StoreManager(models.Manager.from_queryset(StoreQueryset)): + pass class Store(AbstractStore): diff --git a/tests/testapp/tests/models/test_core.py b/tests/testapp/tests/models/test_core.py index 66f1e1bb..cdc68d2f 100644 --- a/tests/testapp/tests/models/test_core.py +++ b/tests/testapp/tests/models/test_core.py @@ -544,3 +544,66 @@ def test_buffer_self_ref_order_rejects_negative(self): buffer = self._buffer(-1) with self.assertRaises(ValidationError): buffer.full_clean() + + +class StoreQuerysetDeserializationErrorTestCase(TestCase): + """ + Tests for the `StoreQueryset` deserialization error filters. `deserialization_error` was + historically non-nullable and set to an empty string when there was no error, so both an + empty string and null have to be treated as "this record has not errored". + """ + + def _store(self, deserialization_error): + return StoreFactory( + id=uuid.uuid4().hex, + partition="test", + serialized="{}", + last_saved_instance=uuid.uuid4().hex, + last_saved_counter=1, + deserialization_error=deserialization_error, + ) + + def setUp(self): + self.empty = self._store("") + self.null = self._store(None) + self.errored = self._store("it broke") + + def test_filter_has_deserialization_error(self): + self.assertEqual( + [store.id for store in Store.objects.all().filter_has_deserialization_error()], + [self.errored.id], + ) + + def test_exclude_has_deserialization_error(self): + self.assertEqual( + sorted(store.id for store in Store.objects.all().exclude_has_deserialization_error()), + sorted([self.empty.id, self.null.id]), + ) + + def test_filter_deserialization_error__true(self): + self.assertEqual( + [store.id for store in Store.objects.all().filter_deserialization_error(True)], + [self.errored.id], + ) + + def test_filter_deserialization_error__false(self): + self.assertEqual( + sorted(store.id for store in Store.objects.all().filter_deserialization_error(False)), + sorted([self.empty.id, self.null.id]), + ) + + def test_filters_are_complementary(self): + """Every record falls into exactly one of the two filters""" + self.assertEqual( + Store.objects.all().filter_has_deserialization_error().count() + + Store.objects.all().exclude_has_deserialization_error().count(), + Store.objects.count(), + ) + + def test_chains_onto_an_existing_queryset(self): + """The filter is a queryset method, so it must compose with prior filtering""" + queryset = Store.objects.filter(id=self.errored.id).filter_has_deserialization_error() + self.assertEqual([store.id for store in queryset], [self.errored.id]) + + queryset = Store.objects.filter(id=self.empty.id).filter_has_deserialization_error() + self.assertEqual(list(queryset), []) From 5ee2ac396a88c5b441b538398f26c340b8c0e0ef Mon Sep 17 00:00:00 2001 From: Blaine Jester Date: Tue, 25 Aug 2026 11:40:56 -0700 Subject: [PATCH 4/5] Add complementary method for querying store records during deserialization --- morango/registry.py | 12 +++ tests/testapp/tests/test_registry.py | 117 +++++++++++++++++++++++++++ 2 files changed, 129 insertions(+) diff --git a/morango/registry.py b/morango/registry.py index 745b1f4c..15b93e90 100644 --- a/morango/registry.py +++ b/morango/registry.py @@ -110,6 +110,18 @@ def get_model_querysets(self, profile) -> Generator[QuerySet, None, None]: queryset = queryset.order_by(*self._get_nulls_last_ordering(ordering)) yield queryset + def get_store_querysets(self, profile) -> Generator[QuerySet, None, None]: + """ + Complementary method to `get_model_querysets` but for Store querysets + """ + from morango.models.core import Store + + for model in self.get_models(profile): + store_qs = Store.objects.filter(profile=profile, model_name=model.morango_model_name) + if self.get_self_referential_fk(model) is not None: + store_qs = store_qs.order_by(*self._get_nulls_last_ordering(("_self_ref_order",))) + yield store_qs + @staticmethod def _get_nulls_last_ordering(ordering): normalized = [] diff --git a/tests/testapp/tests/test_registry.py b/tests/testapp/tests/test_registry.py index 80ccc6f9..740c6739 100644 --- a/tests/testapp/tests/test_registry.py +++ b/tests/testapp/tests/test_registry.py @@ -1,10 +1,15 @@ +import uuid from collections import defaultdict +from contextlib import contextmanager import mock from django.test import SimpleTestCase +from django.test import TestCase from morango.registry import syncable_models +from .helpers import StoreFactory + class SyncableModelRegistryTestCase(SimpleTestCase): def setUp(self): @@ -116,3 +121,115 @@ def test_get_model_querysets_applies_ordering_per_model(self): mock_get_ordering.assert_called_once_with(("field_a",)) queryset_a.order_by.assert_called_once_with("normalized_a") queryset_b.order_by.assert_not_called() + + +class GetStoreQuerysetsTestCase(TestCase): + """ + `get_store_querysets` is the deserialization counterpart to `get_model_querysets`, and is + responsible for the ordering guarantees the deserialization stage depends upon. + + These assert the order of records actually returned, rather than the query used to fetch + them, since the latter varies by backend. + """ + + def _model(self, model_name, self_ref_fk=None): + """ + :param self_ref_fk: the attname of the model's self-referential FK, or None if it has none + """ + return mock.Mock(morango_model_name=model_name, self_ref_fk=self_ref_fk) + + @contextmanager + def _registry(self, models): + with mock.patch.object(syncable_models, "get_models", return_value=models) as get_models: + with mock.patch.object( + syncable_models, + "get_self_referential_fk", + side_effect=lambda model: model.self_ref_fk, + ): + yield get_models + + def _store(self, model_name, self_ref_order=None, profile="facilitydata"): + return StoreFactory( + id=uuid.uuid4().hex, + profile=profile, + model_name=model_name, + partition="partition", + serialized="{}", + last_saved_instance=uuid.uuid4().hex, + last_saved_counter=1, + _self_ref_order=self_ref_order, + ) + + def _queryset_for(self, model_name, self_ref_fk="parent_id"): + with self._registry([self._model(model_name, self_ref_fk=self_ref_fk)]): + return next(iter(syncable_models.get_store_querysets("facilitydata"))) + + def test_orders_parents_before_children(self): + # inserted out of order, so row order alone would not produce the expected result + for self_ref_order in (2, 0, 1): + self._store("abc", self_ref_order=self_ref_order) + + self.assertEqual( + [store._self_ref_order for store in self._queryset_for("abc")], + [0, 1, 2], + ) + + def test_orders_unresolved_parents_last(self): + # nulls are inserted first, so row order alone would place them at the front. A null + # `_self_ref_order` means the record's parent could not be resolved + self._store("abc", self_ref_order=None) + self._store("abc", self_ref_order=None) + for self_ref_order in (1, 0, 2): + self._store("abc", self_ref_order=self_ref_order) + + self.assertEqual( + [store._self_ref_order for store in self._queryset_for("abc")], + [0, 1, 2, None, None], + ) + + def test_no_ordering_for_models_without_a_self_referential_fk(self): + """ + `_self_ref_order` is null for every record of a model with no self-referential FK, so + there is no tree to walk and sorting by it would only cost the database work + """ + for _ in range(3): + self._store("abc") + + queryset = self._queryset_for("abc", self_ref_fk=None) + + self.assertFalse(queryset.ordered) + self.assertEqual(queryset.count(), 3) + + def test_orders_only_the_self_referential_models(self): + models = [self._model("plain"), self._model("tree", self_ref_fk="parent_id")] + + with self._registry(models): + plain_qs, tree_qs = list(syncable_models.get_store_querysets("facilitydata")) + + self.assertFalse(plain_qs.ordered) + self.assertTrue(tree_qs.ordered) + + def test_filters_by_profile_and_model_name(self): + expected = self._store("abc") + self._store("other") + self._store("abc", profile="otherprofile") + + self.assertEqual([store.id for store in self._queryset_for("abc")], [expected.id]) + + def test_yields_one_queryset_per_model_in_dependency_order(self): + self._store("first") + self._store("second") + self._store("second") + self._store("third") + + models = [self._model("first"), self._model("second"), self._model("third")] + with self._registry(models) as mock_get_models: + querysets = list(syncable_models.get_store_querysets("facilitydata")) + + mock_get_models.assert_called_once_with("facilitydata") + # the registry orders models such that a model's foreign key targets precede it, and each + # queryset must stay paired with its model to preserve that order downstream + self.assertEqual( + [[store.model_name for store in queryset] for queryset in querysets], + [["first"], ["second", "second"], ["third"]], + ) From 6af15df8423cff252b5e30cc7978654099c4ed11 Mon Sep 17 00:00:00 2001 From: Blaine Jester Date: Wed, 15 Apr 2026 11:46:43 -0700 Subject: [PATCH 5/5] Add new source for deserialization of store records --- morango/sync/stream/deserialize.py | 92 ++++++++ .../tests/sync/stream/test_deserialize.py | 217 ++++++++++++++++++ 2 files changed, 309 insertions(+) create mode 100644 morango/sync/stream/deserialize.py create mode 100644 tests/testapp/tests/sync/stream/test_deserialize.py diff --git a/morango/sync/stream/deserialize.py b/morango/sync/stream/deserialize.py new file mode 100644 index 00000000..a39be8f2 --- /dev/null +++ b/morango/sync/stream/deserialize.py @@ -0,0 +1,92 @@ +from typing import Dict +from typing import Generator +from typing import List +from typing import Optional +from typing import Type + +from morango.models.certificates import Filter +from morango.models.core import Store +from morango.models.core import SyncableModel +from morango.registry import syncable_models +from morango.sync.stream.source import MorangoSource +from morango.sync.stream.source import SourceTask + + +class DeserializeTask(SourceTask): + """Carrier class for providing context through the deserialization pipeline.""" + + __slots__ = ("store", "app_model", "fk_cache", "errors") + + def __init__(self, store: Store, fk_cache: Dict): + self.store = store + self.fk_cache: Dict = fk_cache + self.app_model: Optional[SyncableModel] = None + self.errors: List[Exception] = [] + + @property + def id(self) -> str: + return self.store.id + + @property + def model(self) -> Type[SyncableModel]: + return syncable_models.get_model(self.store.profile, self.store.model_name) + + @property + def has_errors(self) -> bool: + return len(self.errors) > 0 + + def set_app_model(self, app_model: Optional[SyncableModel]) -> None: + self.app_model = app_model + + def add_error(self, error: Exception) -> None: + self.errors.append(error) + + +class StoreModelSource(MorangoSource[DeserializeTask]): + """ + Yields ``DeserializeTask`` objects for dirty store models that match the optional + *sync_filter*. + """ + + def __init__( + self, + profile: str, + sync_filter: Optional[Filter] = None, + dirty_only: bool = True, + partition_order: str = "asc", + fk_cache: Optional[Dict] = None, + skip_errored: bool = False, + ): + """ + :param profile: The Morango model profile + :param sync_filter: The Filter object for this sync + :param dirty_only: Whether to filter on dirty records only + :param partition_order: Controls how the filter specificity is applied, "asc" or "desc" + :param fk_cache: Dictionary cache for FK references + :param skip_errored: Whether to skip Store records with deserialization errors + """ + super().__init__(profile, sync_filter, dirty_only, partition_order) + self.fk_cache = fk_cache if fk_cache is not None else {} + self.skip_errored = skip_errored + + def begin(self) -> None: + """Reset fk_cache at the beginning of stream""" + super().begin() + self.fk_cache.clear() + + def stream_for_filter( + self, partition_condition: Optional[str] + ) -> Generator[DeserializeTask, None, None]: + # the registry yields models in foreign key dependency order, so streaming model by model + # ensures a record's foreign key targets are deserialized before it is + for store_qs in syncable_models.get_store_querysets(self.profile): + qs = store_qs + if partition_condition is not None: + qs = qs.filter(partition__startswith=partition_condition) + if self.dirty_only: + qs = qs.filter(dirty_bit=True) + if self.skip_errored: + qs = qs.exclude_has_deserialization_error() + + for store_model in qs.iterator(): + yield DeserializeTask(store_model, self.fk_cache) diff --git a/tests/testapp/tests/sync/stream/test_deserialize.py b/tests/testapp/tests/sync/stream/test_deserialize.py new file mode 100644 index 00000000..41f7c870 --- /dev/null +++ b/tests/testapp/tests/sync/stream/test_deserialize.py @@ -0,0 +1,217 @@ +import uuid + +import mock +from django.test import SimpleTestCase +from django.test import TestCase + +from morango.models.certificates import Filter +from morango.models.core import Store +from morango.models.core import SyncableModel +from morango.sync.stream.deserialize import DeserializeTask +from morango.sync.stream.deserialize import StoreModelSource + +from ...helpers import StoreFactory + + +class DeserializeTaskTestCase(SimpleTestCase): + def setUp(self): + self.store = mock.Mock(spec_set=Store) + self.store.profile = "test" + self.store.model_name = "testmodel" + self.task = DeserializeTask(self.store, {}) + + @mock.patch("morango.sync.stream.deserialize.syncable_models.get_model") + def test_model(self, mock_get_model): + model = mock.Mock(spec_set=SyncableModel) + mock_get_model.return_value = model + + self.assertEqual(self.task.model, model) + mock_get_model.assert_called_once_with("test", "testmodel") + + def test_has_errors(self): + self.assertFalse(self.task.has_errors) + self.task.add_error(ValueError("bad data")) + self.assertTrue(self.task.has_errors) + + def test_set_app_model(self): + app_model = mock.Mock(spec_set=SyncableModel) + self.task.set_app_model(app_model) + self.assertEqual(self.task.app_model, app_model) + + +class StoreModelSourcePrefixConditionsTestCase(SimpleTestCase): + """`prefix_conditions` is pure ordering logic over the sync filter and touches no ORM""" + + def test_prefix_conditions__none(self): + source = StoreModelSource(profile="test") + self.assertEqual(list(source.prefix_conditions()), [None]) + + def test_prefix_conditions__with_filter_asc(self): + source = StoreModelSource(profile="test", sync_filter=Filter("b\na"), partition_order="asc") + self.assertEqual(list(source.prefix_conditions()), ["a", "b"]) + + def test_prefix_conditions__with_filter_asc__realistic(self): + """ + Partitions often take the form of `{id}:{additional specificity}`, so the important aspect + of partition ordering is that the least specific filter is generally first + """ + source = StoreModelSource( + profile="test", sync_filter=Filter("a:test:z\na\na:initial"), partition_order="asc" + ) + self.assertEqual(list(source.prefix_conditions()), ["a", "a:initial", "a:test:z"]) + + def test_prefix_conditions__with_filter_desc(self): + source = StoreModelSource( + profile="test", sync_filter=Filter("a\nb"), partition_order="desc" + ) + self.assertEqual(list(source.prefix_conditions()), ["b", "a"]) + + +class StoreModelSourceBeginTestCase(SimpleTestCase): + def test_begin_clears_fk_cache(self): + """The FK cache is only valid within a single run, since the app models can change""" + fk_cache = {"facility": "stale"} + source = StoreModelSource(profile="test", fk_cache=fk_cache) + + source.begin() + + self.assertEqual(fk_cache, {}) + self.assertIs(source.fk_cache, fk_cache) + + def test_begin_initializes_seen(self): + """Delegates to the base source, which owns the seen set""" + source = StoreModelSource(profile="test") + self.assertIsNone(source._seen) + + source.begin() + + self.assertEqual(source._seen, set()) + + +class StoreModelSourceStreamTestCase(TestCase): + """ + Streams against real `Store` rows and the real registry, so that the filters are validated as + queries the database actually accepts, and not merely as the keyword arguments the source + happened to pass along + """ + + profile = "facilitydata" + + def _store( + self, + model_name="user", + partition="a", + dirty_bit=True, + deserialization_error=None, + self_ref_order=None, + ): + return StoreFactory( + id=uuid.uuid4().hex, + profile=self.profile, + model_name=model_name, + partition=partition, + serialized="{}", + last_saved_instance=uuid.uuid4().hex, + last_saved_counter=1, + dirty_bit=dirty_bit, + deserialization_error=deserialization_error, + _self_ref_order=self_ref_order, + ) + + def _stream_ids(self, **kwargs): + source = StoreModelSource(profile=self.profile, **kwargs) + source.begin() + return [task.store.id for task in source.stream()] + + def test_stream__dirty_only(self): + dirty = self._store(dirty_bit=True) + self._store(dirty_bit=False) + + self.assertEqual(self._stream_ids(), [dirty.id]) + + def test_stream__dirty_only_false(self): + dirty = self._store(dirty_bit=True) + clean = self._store(dirty_bit=False) + + self.assertEqual(sorted(self._stream_ids(dirty_only=False)), sorted([dirty.id, clean.id])) + + def test_stream__skip_errored(self): + """ + `deserialization_error` was historically non-nullable and set to an empty string, so both + an empty string and null have to be treated as "this record has not errored" + """ + null_error = self._store(deserialization_error=None) + empty_error = self._store(deserialization_error="") + self._store(deserialization_error="it broke") + + self.assertEqual( + sorted(self._stream_ids(skip_errored=True)), + sorted([null_error.id, empty_error.id]), + ) + + def test_stream__errored_included_by_default(self): + """Errored records are retried unless the caller opts into skipping them""" + errored = self._store(deserialization_error="it broke") + clean = self._store(deserialization_error=None) + + self.assertEqual(sorted(self._stream_ids()), sorted([errored.id, clean.id])) + + def test_stream__no_partition_filter(self): + first = self._store(partition="a") + second = self._store(partition="zzz") + + self.assertEqual(sorted(self._stream_ids()), sorted([first.id, second.id])) + + def test_stream__partition_filter(self): + included = self._store(partition="a:one") + self._store(partition="b:two") + + self.assertEqual(self._stream_ids(sync_filter=Filter("a")), [included.id]) + + def test_stream__deduplicates_across_partition_passes(self): + """ + A less specific partition prefix already matches everything beneath it, so the same record + turns up in more than one pass and must only be yielded once + """ + shared = self._store(partition="a:b") + + self.assertEqual(self._stream_ids(sync_filter=Filter("a\na:b")), [shared.id]) + + def test_stream__parents_before_children(self): + """`facility` is the profile's self-referential model, so its records carry a tree depth""" + child = self._store(model_name="facility", self_ref_order=2) + root = self._store(model_name="facility", self_ref_order=0) + middle = self._store(model_name="facility", self_ref_order=1) + + self.assertEqual(self._stream_ids(), [root.id, middle.id, child.id]) + + def test_stream__unresolved_parents_last(self): + unresolved = self._store(model_name="facility", self_ref_order=None) + root = self._store(model_name="facility", self_ref_order=0) + + self.assertEqual(self._stream_ids(), [root.id, unresolved.id]) + + def test_stream__models_in_dependency_order(self): + """ + The registry orders models so that a record's foreign key targets precede it, and the + source must preserve that. `user` is registered ahead of `facility` for this profile. + """ + facility = self._store(model_name="facility") + user = self._store(model_name="user") + + self.assertEqual(self._stream_ids(), [user.id, facility.id]) + + def test_stream__ignores_other_profiles(self): + included = self._store() + StoreFactory( + id=uuid.uuid4().hex, + profile="otherprofile", + model_name="user", + partition="a", + serialized="{}", + last_saved_instance=uuid.uuid4().hex, + last_saved_counter=1, + dirty_bit=True, + ) + + self.assertEqual(self._stream_ids(), [included.id])