Streaming deserialize: add source stream and task - #318
Conversation
02b79c4 to
1349f21
Compare
51f7cfc to
acc181f
Compare
bjester
changed the title
bjester
changed the title
🟡 Waiting for changesLast updated: 2026-08-26 14:48 UTC |
rtibblesbot
left a comment
rtibblesbot
left a comment
There was a problem hiding this comment.
PR #318 delivers all four of issue #317's items — DeserializeTask, StoreModelSource, the shared MorangoSource base, and tests for both — plus the StoreQueryset error filters and get_store_querysets. No blocking defects.
The substantive comments are about semantics the shared base imports into deserialization: partition-major iteration vs. the model-major ordering _deserialize_from_store guarantees today, a per-task fk_cache that can never hit, a skip_errored default that inverts the legacy one, and a lost __slots__ optimization.
CI: Linting and Migrations pass on acc181f; Python tests were still running. The one completed failure (Handle pull request events at Generate App Token) is infrastructure, unrelated to the diff. Locally tests/sync/stream/, test_registry.py and models/test_core.py pass (106) and makemigrations --check is clean.
No UI files, so Phase 3 / manual QA did not apply.
@rtibblesbot's comments are generated by an LLM, and should be evaluated accordingly
How was this generated?
Ran a phased review pipeline over the pull request diff:
- Classified the diff to select review passes (core, frontend, backend) and whether manual QA was required
- Core review pass checked correctness, design, architecture, testing, completeness, and DRY/SRP/Rule-of-Three principles
- Specialized frontend/backend review passes applied framework-specific lenses where those files changed
- For UI changes: manual QA and an accessibility audit against a live dev server, when available
- Checked CI status and linked issue acceptance criteria
- Synthesized one review from those passes and chose the verdict from the findings, CI status, and QA evidence
acc181f to
b0b1fd4
Compare
rtibblesbot
left a comment
rtibblesbot
left a comment
There was a problem hiding this comment.
PR #318 — all 11 prior findings resolved or acknowledged (verified in the tree, not on thread state); 2 new findings, both on code added last round.
CI: test matrices green; only the contributor-check app-token step fails — infrastructure. No UI files.
skip_errored now defaults to False, matching _deserialize_from_store(skip_erroring=False) rather than #317's text — my request last round, still right.
Prior-finding status
RESOLVED — morango/sync/stream/source.py:14 — SourceTask lacks __slots__
RESOLVED — morango/sync/stream/source.py:87 — _seen unbounded (never-reset residual refiled below)
RESOLVED — morango/sync/stream/source.py:75 — "passes thoughts to"
RESOLVED — morango/sync/stream/deserialize.py:69 — fresh {} per record defeats FK cache
RESOLVED — morango/sync/stream/deserialize.py:58 — skip_errored inverted the legacy default
RESOLVED — morango/registry.py:120 — _self_ref_order sort applied to every model
RESOLVED — morango/models/core.py:444 — NULLIF yields text, not a boolean
RESOLVED — tests/testapp/tests/sync/stream/test_deserialize.py:70 — stream tests mocked the ORM
ACKNOWLEDGED — morango/sync/stream/source.py:80 — partition-major ordering; no cross-partition FKs
ACKNOWLEDGED — morango/models/core.py:444 — operations.py duplication; that code is slated for removal
n/a (praise) — tests/testapp/tests/test_registry.py:130 — asserts rows, not SQL
@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
| :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 or {} |
There was a problem hiding this comment.
suggestion: {} is falsy, so fk_cache or {} drops a caller-supplied empty dict — and construction is the only time it is passed, so the shared cache I asked for last round never shares. Use fk_cache if fk_cache is not None else {}.
No test references fk_cache, so reverting to a per-record {} stays green. self.assertIs(tasks[0].fk_cache, cache) catches both.
| self.sync_filter = sync_filter | ||
| self.dirty_only = dirty_only | ||
| self.partition_order = partition_order | ||
| self._seen = set() |
There was a problem hiding this comment.
suggestion: _seen and the source-owned fk_cache are instance-scoped, but Pipeline.end() calls Source.begin() once per run (stream/core.py:112) and MorangoSource does not override it. A reused source with a sync_filter yields nothing on a second run; cache entries meaning "FK target validated" never expire, where legacy scopes them to one _deserialize_from_store call (operations.py:297). Unreachable today, but the new constructor invites callers to hold a source — reset both in begin().
|
|
||
| self.assertEqual(sorted(self._stream_ids(dirty_only=False)), sorted([dirty.id, clean.id])) | ||
|
|
||
| def test_stream__skip_errored(self): |
There was a problem hiding this comment.
✅ Resolved — addressed in the current code.
praise: Both NULL and empty-string deserialization_error against real rows — the historical case NullIf exists for, and one a mocked queryset could not catch.
rtibblesbot
left a comment
rtibblesbot
left a comment
There was a problem hiding this comment.
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
| # 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] |
There was a problem hiding this comment.
suggestion: Every prefix in the filter yields a pass, including prefixes already covered by a shorter one — and StoreModelSource.stream_for_filter issues one query per registered model per pass.
Filter("a:test:z\na\na:initial") (the shape in test_prefix_conditions__with_filter_asc__realistic) sorts to ["a", "a:initial", "a:test:z"]. Pass 1 matches every row the other two can match; passes 2 and 3 are 2 × N-models queries whose rows are all discarded against _seen.
Dropping prefixes that extend another prefix in the same filter preserves the covered set exactly — Filter.contains_partition (morango/models/certificates.py:383-385) already does the test. In the common single-covering-prefix case it also collapses to one pass, which drops the _seen bookkeeping and keeps the whole stream in model-dependency order.
Separate from the cross-partition ordering point answered last round; that answer stands.
| :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 or {} |
There was a problem hiding this comment.
suggestion: (still open) {} is falsy, so fk_cache or {} silently substitutes a new dict when a caller passes an empty one — the caller then never sees the entries the source populates. fk_cache if fk_cache is not None else {} keeps the caller's object.
No test asserts the tasks share the source's dict object; assertIs(task.fk_cache, source.fk_cache) would pin both this and the sharing behaviour.
| self.sync_filter = sync_filter | ||
| self.dirty_only = dirty_only | ||
| self.partition_order = partition_order | ||
| self._seen = set() |
There was a problem hiding this comment.
suggestion: (still open) _seen — and fk_cache in the subclass — are instance state that lives for the source's lifetime, but Source.begin() (morango/sync/stream/core.py:60) exists so a source can be re-run. MorangoSource doesn't override it, so a second stream() on the same instance yields nothing: every id is already in _seen.
Overriding begin() to clear both makes the reuse contract explicit even if the pipeline currently only runs each source once.
2 participants
Summary
StoreQuerySetmethods that apply filtering for selecting records based on whether they have deserialization errors. These can be used in Kolibri laterTODO
Reviewer guidance
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.