Streaming deserialize: add source stream and task

[bjester]

Summary

• Extracts common behavior for streaming source and task classes into new base classes for use with both serialization and deserialization
• Adds new StoreQuerySet methods that apply filtering for selecting records based on whether they have deserialization errors. These can be used in Kolibri later
• Adds method to model registry for querying store records in model dependency order in parity with the method used by serialization
• Creates deserialization stream and task specific classes as foundation for streaming deserialization stage

TODO

• Have tests been written for the new code?
• Has documentation been written/updated?
• New dependencies (if any) added to requirements file

Reviewer guidance

• Do the tests cover the new and refactored code enough?
• Did I properly translate the self-ref-ordering and deserialization error filtering that were added after this PR was opened in draft? (I rebased it and applied

Issues addressed

Closes #317

AI Usage

I used Claude to do the rebase and bring it up-to-date with the upstream changes. It made a rightful mess of it, and so I rewrote some of the changes to better follow the patterns and then had it clean up the mess by updating tests.

[rtibblesbot]

🟡 Waiting for changes

Last updated: 2026-08-28 19:18 UTC

[rtibblesbot]

PR #318 — 12 of 14 prior findings resolved or acknowledged; 2 still open, plus one new suggestion on prefix_conditions (all inline).

HEAD unchanged at b0b1fd4c since the last pass. Test jobs pass on 3.6–3.14 across SQLite and Postgres; the only red check is Handle review requested / Check if author is contributor, a permissions workflow unrelated to the code. No UI files, so Phase 3 did not apply.

Prior-finding status

RESOLVED — morango/sync/stream/source.py:14 — SourceTask declares no __slots__
RESOLVED — morango/sync/stream/source.py:88 — _seen accumulates even when nothing can duplicate
RESOLVED — morango/sync/stream/source.py:75 — "passes thoughts to stream_for_filter"
ACKNOWLEDGED — morango/sync/stream/source.py:80 — Partition-major iteration drops model-dependency ordering
RESOLVED — morango/sync/stream/deserialize.py:69 — Fresh {} per store record gives the FK cache no reuse
RESOLVED — morango/sync/stream/deserialize.py:58 — skip_errored inverted the legacy default
RESOLVED — morango/registry.py:121 — _self_ref_order sort applied to every model
RESOLVED — morango/models/core.py:453 — NULLIF annotation declared BooleanField
ACKNOWLEDGED — morango/models/core.py:444 — Legacy call site in operations.py still hand-rolls the annotation
RESOLVED — tests/testapp/tests/sync/stream/test_deserialize.py:70 — StoreModelSourceTestCase mocked the ORM throughout
RESOLVED — tests/testapp/tests/sync/stream/test_deserialize.py:116 — Both NULL and empty-string deserialization_error against real rows
RESOLVED — tests/testapp/tests/test_registry.py:130 — Asserting returned rows rather than generated SQL
UNADDRESSED — morango/sync/stream/deserialize.py:69 — fk_cache or {} drops a caller-supplied empty dict
UNADDRESSED — morango/sync/stream/source.py:47 — _seen and fk_cache are not reset in begin()

---

@rtibblesbot's comments are generated by an LLM, and should be evaluated accordingly

How was this generated?

Compared the current PR state against findings from a prior review:

• Retrieved prior bot reviews via the GitHub API
• Classified each prior finding as RESOLVED, UNADDRESSED, ACKNOWLEDGED, or CONTESTED
• Only raised NEW findings for newly introduced code
• Ran the same phased review passes as a first review (core, frontend/backend lenses, manual QA when required)
• Synthesized one review from the passes and chose the verdict from the findings, CI status, and QA evidence

[bjester]

Code Review: Streaming deserialize — add source stream and task

1. Executive Summary

• Overall Assessment: COMMENT — no blockers. Two design notes worth resolving before this base class carries the rest of the series.
• Summary: All four of #317's deliverables land (DeserializeTask, StoreModelSource, the shared MorangoSource/SourceTask base, tests), plus the StoreQueryset error filters and get_store_querysets. The extraction is genuinely a shared base rather than a copy — AppModelSource shrinks from ~45 lines to 12 and keeps behaviour. The test suite for the ordering guarantee runs against real rows on both backends, which is the right level for something deserialization correctness depends on.
• Impact Surface: morango/sync/stream/{source,serialize,deserialize}.py, morango/registry.py, morango/models/core.py (StoreQueryset/StoreManager). No migration needed — no field changes. No existing code path is rewired, so blast radius on this PR alone is limited to the new source; the risk arrives when it is wired into the deserialize operation.
• Verification: pytest tests/testapp/tests/sync/stream/ tests/testapp/tests/test_registry.py tests/testapp/tests/models/test_core.py → 119 passed. CI green across 3.6–3.14 / SQLite + Postgres; the only red check is the contributor-check app-token step (infrastructure).

---

2. Principle-Grounded Findings

⚠️ Architecture & Design (Should-Fix)

• stream() is a partial function — it raises an opaque TypeError unless begin() was called first (morango/sync/stream/source.py:47-51, :89)

  • Principle: Defensive Contracts — a public entry point should either work or fail with a message that names the violated precondition.
  • Issue: _seen is None until begin() runs, and stream() does obj.id not in self._seen. Confirmed on this tree:

    TypeError: argument of type 'NoneType' is not iterable
    

    Nothing in that message points at the missing begin(). This is also a narrow regression against the pre-refactor AppModelSource, which initialised _seen in __init__ and was safe to call standalone — the evidence is in this PR's own diff, where three previously-passing tests had to gain a source.begin() line (tests/.../test_serialize.py:74,118).
  • Remediation: initialise in __init__ and reset in begin(). Keeps the reuse contract that test_begin_resets_seen_between_runs pins, and makes a stray stream() correct rather than fatal:

    self._seen: set = set()
    
    def begin(self) -> None:
        """Reset seen set at the beginning of the stream"""
        self._seen = set()

    If the None sentinel is deliberate — "you must go through Pipeline.end()" — then assert it, so the failure names itself: _assert(self._seen is not None, "begin() must be called before stream()").

• _seen grows unbounded on the filtered path — the memory cost this pipeline exists to eliminate (morango/sync/stream/source.py:84-93)

  • Principle: the module docstring's own goal — "streamed one-by-one … reducing memory overhead" (stream/core.py:1-6).
  • Issue: The no-filter case is already exempted, so what remains is exactly the case that matters: a filtered facility sync retains one 32-char id per record streamed, for the lifetime of the source. At ~130 bytes per set entry that is ~130 MB at 1M Store rows — on the low-resource deployments this design targets.
  • Remediation: the dedup can be stateless. Passes differ only by their partition condition, so a record yielded in pass k is precisely a record whose partition matches prefix k — excluding earlier prefixes in later passes is exactly equivalent to the id set, and pushes the work into the query:

    def stream(self) -> Generator[T, None, None]:
        processed = []
        for prefix in self.prefix_conditions():
            for obj in self.stream_for_filter(prefix, exclude_prefixes=processed):
                yield obj
            processed.append(prefix)

    with stream_for_filter applying .exclude(partition__startswith=p) per entry. Prefix counts are small, processed is bounded by the filter size rather than the row count, and _seen disappears. Holds for partition_order="desc" too. This is distinct from the prefix-reduction idea you declined earlier — that one traded query count for coverage reasoning; this one removes state without changing either.

💡 Tactical Suggestions & Polish (Optional)

• begin() wipes a caller-supplied fk_cache (morango/sync/stream/deserialize.py:69-75): self.fk_cache.clear() mutates an object the caller owns, which makes the constructor parameter unable to do the one thing injection is usually for — handing in a warm cache. The remaining value is post-run inspection, which is legitimate but non-obvious. Either say so on the parameter (:param fk_cache: … cleared at the start of every run) or only clear when the source created the dict itself.

• Boolean-flag method sits in the public queryset API (morango/models/core.py:444): filter_deserialization_error(has_error) is a flag argument whose two call sites are the intention-revealing wrappers right below it. Renaming it _filter_deserialization_error leaves callers with only the two readable spellings. (The wrappers themselves are the right shape — this is just about which of the three is the advertised one.)

• The "no cross-partition FK references" invariant lives only in a PR thread (morango/sync/stream/deserialize.py:80-81): the comment claims model-major streaming gets FK targets deserialized first, which holds globally only because partitions never contain FKs across each other. That premise is the load-bearing part and it is nowhere in the code. One clause — "partitions do not contain cross-partition FK references, so per-partition model ordering is sufficient" — saves the next reader the rediscovery.

• AppModelSource no longer dedups when sync_filter is None (morango/sync/stream/source.py:90-91): a deliberate and near-certainly safe narrowing (a single pass over non-joined querysets cannot repeat a row), but the test that used to cover it was retargeted to the filtered path (test_serialize.py:92) rather than replaced, so nothing asserts the unfiltered behaviour any more. Worth a line in the PR body if not a test.

• Nothing pins __slots__ effectiveness (morango/sync/stream/source.py:11-18): the __slots__ = () on SourceTask is load-bearing and verified working here —

  DeserializeTask has __dict__: False
  AttributeError: 'DeserializeTask' object has no attribute 'zzz'
  

  — but it is one easily-dropped line away from silently reintroducing a __dict__ on every task in the stream. with self.assertRaises(AttributeError): task.nope = 1 in DeserializeTaskTestCase costs two lines and holds the line for the whole series.

• abc.ABC in MorangoSource(Source[T], abc.ABC) is redundant (morango/sync/stream/source.py:26): Source → PipelineModule → StreamModule(abc.ABC) already supplies ABCMeta.

• Docs and CHANGELOG (docs/architecture/index.rst:135-176): the "Streaming architecture" section enumerates the stream module vocabulary and the serialization pipeline; MorangoSource, SourceTask, and the now-load-bearing begin() lifecycle aren't in it. Deferring the prose to the end of the series is reasonable — worth saying so explicitly against the unchecked box rather than leaving it ambiguous.

✅ Positive Highlights

• GetStoreQuerysetsTestCase asserts returned rows, not generated SQL — with the reasoning written into the docstrings, and staying valid across SQLite and Postgres. test_orders_unresolved_parents_last in particular pins nulls-last for _self_ref_order, which is the subtle half of the ordering contract.
• StoreModelSourceStreamTestCase runs against real Store rows and the real registry. test_stream__skip_errored covering both NULL and "" is the historical case NullIf exists for, and test_stream__models_in_dependency_order pins the registry-order guarantee at the level a consumer actually depends on.
• StoreManager(models.Manager.from_queryset(StoreQueryset)) — deletes a hand-rolled get_queryset and makes every queryset method reachable from the manager, which is what let exclude_has_deserialization_error() read as one call at the use site.
• The NullIf-over-OR comment (core.py:449-450) records why the odd-looking annotation is there, on both backends. That is the comment that stops someone "simplifying" it back into an OR.
• Gating the _self_ref_order sort on get_self_referential_fk (registry.py:121) keeps the majority of models off an unindexed sort while preserving the tree ordering where it is needed.

---

3. Reviewer Guidance, Answered

Do the tests cover the new and refactored code enough?

Coverage of the new code is good, and the real-row tests are the right choice. Two gaps, both listed above: __slots__ effectiveness, and unfiltered AppModelSource dedup (whose old test moved to the filtered path).

Did I properly translate the self-ref-ordering and deserialization error filtering?

Yes on both, as far as this PR's surface goes.

• Error filtering is a faithful extraction of operations.py:326-336, and the TextField() output field is more accurate than the BooleanField() it came from. Both null and "" are handled, and tested against real rows.
• Self-ref ordering matches the semantics _update_legacy_self_ref_order_for_model establishes (operations.py:728-745): roots at 0, children at parent+1, NULL when unresolvable — so ascending-nulls-last is parents-before-children with the unresolvable tail last. Correct.

One difference to carry forward rather than fix here: get_store_querysets scopes each queryset to a single model_name, where _deserialize_from_store ORs in morango_model_dependencies (operations.py:307-311). That is the better shape — the registry already orders dependencies first via _insert_model_in_dependency_order, so each record is now covered exactly once instead of revisited — and it matches get_model_querysets. Worth knowing when the consumer lands, since the legacy self-ref branch computed clean_parents/dirty_children over that wider set.

The other thing the consumer will have to supply: the legacy self-ref branch didn't just order records, it classified the failures — MorangoDirtyParent and MorangoMissingParent (operations.py:380-396). Ordering by _self_ref_order replaces the iteration but not the diagnostics.

---

4. Next Steps

• Make stream() safe or self-diagnosing without begin() (source.py:47)
• Decide on stateless prefix-exclusion vs. keeping _seen, and record which and why
• Document or narrow begin()'s clearing of an injected fk_cache
• Optional: _filter_deserialization_error rename; __slots__ regression test; cross-partition-FK invariant in the comment; drop redundant abc.ABC
• Confirm docs/CHANGELOG are intentionally deferred to a later part of the series