From 8eae4ad120e818b4f847e60c08c82d3785489c6a Mon Sep 17 00:00:00 2001 From: Lukasz Lancucki Date: Tue, 1 Sep 2026 11:37:45 +0100 Subject: [PATCH] feat(http): let stream() consumers opt out of deletion stubs Add a keyword-only skip_deleted flag to StreamingMixin.stream() and AsyncStreamingMixin.stream() that withholds DeletionStub objects at yield time, for consumers that do not ingest deletions and would otherwise write the isinstance branch only to drop the stubs. The flag is typed with overloads, so skip_deleted=True narrows the yield type to models only while the default call keeps the union. Filtering runs after the completeness bookkeeping and the progress tick: the MPT-Item-Count verification still sees every raw record, and a progress report still reaches the declared total, which counts stubs. The number of yielded objects therefore intentionally falls short of MPT-Item-Count when the snapshot contains stubs, which the usage guide now calls out. Co-Authored-By: Claude Fable 5 --- docs/architecture.md | 5 +- docs/usage.md | 39 +++++- mpt_api_client/http/mixins/streaming_mixin.py | 132 +++++++++++++++++- .../unit/http/mixins/test_streaming_mixin.py | 116 +++++++++++++++ 4 files changed, 285 insertions(+), 7 deletions(-) diff --git a/docs/architecture.md b/docs/architecture.md index 23d93db5..cd8730c6 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -155,7 +155,10 @@ yielded record count once the body is fully consumed, raising `MPTStreamingIncom on mismatch. An iterator closed early skips the comparison. A record marked with `$meta.deleted` is a deletion stub rather than data, and is yielded as a `DeletionStub` instead of a model, so it still counts towards the declared item count but cannot be ingested -as a record. The JSONL mixins serve endpoints that assign `application/jsonl` their own +as a record. A consumer that ingests no deletions can opt out with the keyword-only +`skip_deleted=True`, which withholds stubs at yield time — after the completeness bookkeeping +and the progress tick — and is typed with overloads, so the call narrows to an iterator of +models. The JSONL mixins serve endpoints that assign `application/jsonl` their own meaning outside streaming mode. `stream()` picks its wire format per request with the `stream_format` argument, which sets diff --git a/docs/usage.md b/docs/usage.md index aa12c68a..6229950c 100644 --- a/docs/usage.md +++ b/docs/usage.md @@ -380,13 +380,50 @@ no longer exists at all and carries no state. Every member selected for the stream yields exactly one object, stubs included, so a stub counts towards `MPT-Item-Count`. Do not filter stubs out before the completeness check -described below, or a complete export reads as short. +described below, or a complete export reads as short; the safe way to drop them is the +`skip_deleted` flag described next, which filters only after that check has been fed. If you validate incoming payloads against a strict schema, relax the required-member validation for stubs: a stub satisfies only `id`, so checking it against a schema that requires the full record fails on a conformant payload. Branch on `DeletionStub` first and skip the record validation for stubs, rather than loosening the schema for records too. +### Opting Out Of Deletion Stubs + +A consumer that does not ingest deletions — read-only analytics, an ad-hoc export — gains +nothing from the `isinstance` branch: it would drop the stubs and move on. Declare that +instead with the keyword-only `skip_deleted` flag, and only models are yielded: + +```python +for order in service.stream(skip_deleted=True): + upsert_local_record(order) +``` + +The flag is typed with overloads, so a type checker resolves `stream(skip_deleted=True)` to +`Iterator[Model]` — `AsyncIterator[Model]` on the async service — and an opted-out consumer +carries no union type in downstream signatures. The default call keeps +`Iterator[Model | DeletionStub]` and the branch it forces. + +Filtering happens at yield time, after the client's own bookkeeping, so the stream keeps its +guarantees: + +- The completeness accounting counts the raw records ahead of the filter, and the + comparison against `MPT-Item-Count` still runs once the body is fully consumed: a short + stream raises `MPTStreamingIncompleteError` regardless of the flag, though records + yielded before the truncated tail have been processed by then, as in the default mode. +- A `progress` receiver still gets `item_processed` for every record, withheld stubs + included; the declared total counts stubs, so a report fed only visible records would + never reach it. + +The intentional exception: with `skip_deleted=True` the number of objects your loop sees no +longer matches `MPT-Item-Count` when the snapshot contains stubs. Do not compare your own +count against the header in this mode — the client has already verified that the full +snapshot arrived. + +Opting out declares that deletions are irrelevant to this consumer. A sync job that mirrors +the collection must keep the default and branch on `DeletionStub`, or members deleted +upstream survive locally forever. + ### Streaming Errors All streaming-specific failures derive from `MPTStreamingError`, so one handler covers them: diff --git a/mpt_api_client/http/mixins/streaming_mixin.py b/mpt_api_client/http/mixins/streaming_mixin.py index a17dbff9..a71682e9 100644 --- a/mpt_api_client/http/mixins/streaming_mixin.py +++ b/mpt_api_client/http/mixins/streaming_mixin.py @@ -2,6 +2,7 @@ from collections.abc import AsyncIterator, Iterator, Mapping from contextlib import AsyncExitStack, ExitStack from enum import StrEnum +from typing import Literal, overload from httpx import Response as HTTPXResponse @@ -436,6 +437,7 @@ class StreamingMixin[Model: BaseModel](QueryableMixin): ``application/jsonl`` their own meaning outside streaming mode. """ + @overload def stream( self, *, @@ -443,6 +445,39 @@ def stream( offset: int | None = None, stream_format: StreamFormat = StreamFormat.JSONL, progress: Progress | None = None, + skip_deleted: Literal[True], + ) -> Iterator[Model]: ... + + @overload + def stream( + self, + *, + limit: int | None = None, + offset: int | None = None, + stream_format: StreamFormat = StreamFormat.JSONL, + progress: Progress | None = None, + skip_deleted: Literal[False] = False, + ) -> Iterator[Model | DeletionStub]: ... + + @overload + def stream( + self, + *, + limit: int | None = None, + offset: int | None = None, + stream_format: StreamFormat = StreamFormat.JSONL, + progress: Progress | None = None, + skip_deleted: bool, + ) -> Iterator[Model | DeletionStub]: ... + + def stream( + self, + *, + limit: int | None = None, + offset: int | None = None, + stream_format: StreamFormat = StreamFormat.JSONL, + progress: Progress | None = None, + skip_deleted: bool = False, ) -> Iterator[Model | DeletionStub]: """Stream a result set in streaming mode, yielding one object per record. @@ -470,15 +505,27 @@ def stream( A member's ``Accept`` string is coerced to the member; any other value raises `ValueError` before the request is sent. progress: Optional progress receiver. `item_processed` is called once per - yielded object, stubs included, before it is yielded, and `completed` once + record, stubs included — even a stub withheld by ``skip_deleted``, so a + progress report still reaches the declared total — and `completed` once when the response body is fully consumed and verified complete. `set_total_items` is called with ``$meta.pagination.total`` when the envelope reports it, and never in the line-delimited format, which carries no envelope. + skip_deleted: When set, deletion stubs are filtered out at yield time, for a + consumer that does not ingest deletions and would otherwise write the + ``isinstance`` branch only to drop the stubs. The completeness accounting + counts raw records ahead of the filter, and the count is compared with + ``MPT-Item-Count`` only once the body is fully consumed, so a short + stream raises exactly as it does without the flag — records yielded + before a truncated tail have been processed by then. The number of + yielded objects intentionally falls short of ``MPT-Item-Count`` when the + snapshot contains stubs. The default keeps the contract-faithful shape: + one object per snapshot member, stubs visible. Yields: Resources, one per record of the response, each either a model or a - `DeletionStub` for a member deleted after the membership snapshot. + `DeletionStub` for a member deleted after the membership snapshot; only the + models when ``skip_deleted`` is set. Raises: MPTStreamingNotEnabledError: If the API does not confirm streaming mode. @@ -523,7 +570,7 @@ def stream( stream_format, self._collection_key, # type: ignore[attr-defined] ) - yield from self._stream_results(events, progress) + yield from self._stream_results(events, progress, skip_deleted=skip_deleted) if progress: progress.completed() @@ -531,6 +578,20 @@ def _stream_results( self, events: Iterator[StreamEvent], progress: Progress | None, + *, + skip_deleted: bool, + ) -> Iterator[Model | DeletionStub]: + # A withheld stub was still ticked upstream: the declared total includes stubs, + # so a progress report fed only visible records would never reach it. + for result in self._deserialized_results(events, progress): + if skip_deleted and isinstance(result, DeletionStub): + continue + yield result + + def _deserialized_results( + self, + events: Iterator[StreamEvent], + progress: Progress | None, ) -> Iterator[Model | DeletionStub]: for event in events: if isinstance(event, StreamedTotal): @@ -555,6 +616,39 @@ class AsyncStreamingMixin[Model: BaseModel](QueryableMixin): ``application/jsonl`` their own meaning outside streaming mode. """ + @overload + def stream( + self, + *, + limit: int | None = None, + offset: int | None = None, + stream_format: StreamFormat = StreamFormat.JSONL, + progress: AsyncProgress | None = None, + skip_deleted: Literal[True], + ) -> AsyncIterator[Model]: ... + + @overload + def stream( + self, + *, + limit: int | None = None, + offset: int | None = None, + stream_format: StreamFormat = StreamFormat.JSONL, + progress: AsyncProgress | None = None, + skip_deleted: Literal[False] = False, + ) -> AsyncIterator[Model | DeletionStub]: ... + + @overload + def stream( + self, + *, + limit: int | None = None, + offset: int | None = None, + stream_format: StreamFormat = StreamFormat.JSONL, + progress: AsyncProgress | None = None, + skip_deleted: bool, + ) -> AsyncIterator[Model | DeletionStub]: ... + async def stream( self, *, @@ -562,6 +656,7 @@ async def stream( offset: int | None = None, stream_format: StreamFormat = StreamFormat.JSONL, progress: AsyncProgress | None = None, + skip_deleted: bool = False, ) -> AsyncIterator[Model | DeletionStub]: """Stream a result set in streaming mode, yielding one object per record. @@ -589,15 +684,27 @@ async def stream( A member's ``Accept`` string is coerced to the member; any other value raises `ValueError` before the request is sent. progress: Optional progress receiver. `item_processed` is awaited once per - yielded object, stubs included, before it is yielded, and `completed` once + record, stubs included — even a stub withheld by ``skip_deleted``, so a + progress report still reaches the declared total — and `completed` once when the response body is fully consumed and verified complete. `set_total_items` is called with ``$meta.pagination.total`` when the envelope reports it, and never in the line-delimited format, which carries no envelope. + skip_deleted: When set, deletion stubs are filtered out at yield time, for a + consumer that does not ingest deletions and would otherwise write the + ``isinstance`` branch only to drop the stubs. The completeness accounting + counts raw records ahead of the filter, and the count is compared with + ``MPT-Item-Count`` only once the body is fully consumed, so a short + stream raises exactly as it does without the flag — records yielded + before a truncated tail have been processed by then. The number of + yielded objects intentionally falls short of ``MPT-Item-Count`` when the + snapshot contains stubs. The default keeps the contract-faithful shape: + one object per snapshot member, stubs visible. Yields: Resources, one per record of the response, each either a model or a - `DeletionStub` for a member deleted after the membership snapshot. + `DeletionStub` for a member deleted after the membership snapshot; only the + models when ``skip_deleted`` is set. Raises: MPTStreamingNotEnabledError: If the API does not confirm streaming mode. @@ -645,6 +752,7 @@ async def stream( self._collection_key, # type: ignore[attr-defined] ), progress, + skip_deleted=skip_deleted, ): yield result if progress: @@ -654,6 +762,20 @@ async def _stream_results( self, events: AsyncIterator[StreamEvent], progress: AsyncProgress | None, + *, + skip_deleted: bool, + ) -> AsyncIterator[Model | DeletionStub]: + # A withheld stub was still ticked upstream: the declared total includes stubs, + # so a progress report fed only visible records would never reach it. + async for result in self._deserialized_results(events, progress): + if skip_deleted and isinstance(result, DeletionStub): + continue + yield result + + async def _deserialized_results( + self, + events: AsyncIterator[StreamEvent], + progress: AsyncProgress | None, ) -> AsyncIterator[Model | DeletionStub]: async for event in events: if isinstance(event, StreamedTotal): diff --git a/tests/unit/http/mixins/test_streaming_mixin.py b/tests/unit/http/mixins/test_streaming_mixin.py index ad6042e2..4934bb3d 100644 --- a/tests/unit/http/mixins/test_streaming_mixin.py +++ b/tests/unit/http/mixins/test_streaming_mixin.py @@ -634,6 +634,57 @@ def test_stream_keeps_deleted_status_as_a_record(nullable_fields_service, delete assert result[0].status == "DELETED" +@respx.mock +def test_stream_skip_deleted_yields_only_models( + nullable_fields_service, deletion_stub_record, data_record +): + records = [data_record, deletion_stub_record] + respx.get(STREAM_URL).mock(return_value=records_response(records)) + + result = list(nullable_fields_service.stream(skip_deleted=True)) + + assert [entry.id for entry in result] == ["ID-1"] + assert all(isinstance(entry, NullableFieldsModel) for entry in result) + + +@respx.mock +def test_stream_skip_deleted_all_stubs_empty(nullable_fields_service, deletion_stub_record): + respx.get(STREAM_URL).mock(return_value=records_response([deletion_stub_record])) + + result = list(nullable_fields_service.stream(skip_deleted=True)) + + assert result == [] + + +@respx.mock +def test_stream_skip_deleted_reports_progress( + nullable_fields_service, deletion_stub_record, recording_progress, data_record +): + records = [data_record, deletion_stub_record] + respx.get(STREAM_URL).mock(return_value=records_response(records)) + stream = nullable_fields_service.stream(progress=recording_progress, skip_deleted=True) + + list(stream) # act + + assert recording_progress.events == [ + ("item_processed",), + ("item_processed",), + ("completed",), + ] + + +@respx.mock +def test_stream_skip_deleted_verifies_count( + nullable_fields_service, deletion_stub_record, data_record +): + records = [data_record, deletion_stub_record] + respx.get(STREAM_URL).mock(return_value=records_response(records, item_count="3")) + iterator = nullable_fields_service.stream(skip_deleted=True) + + with pytest.raises(MPTStreamingIncompleteError, match=COUNT_MISMATCH_MATCH): + list(iterator) + + @respx.mock async def test_async_stream_yields_deletion_stub( async_nullable_fields_service, deletion_stub_record, data_record @@ -678,6 +729,51 @@ async def test_async_stream_progress_counts_stub( ] +@respx.mock +async def test_async_stream_skip_deleted_only_models( + async_nullable_fields_service, deletion_stub_record, data_record +): + records = [data_record, deletion_stub_record] + respx.get(STREAM_URL).mock(return_value=records_response(records)) + stream = async_nullable_fields_service.stream(skip_deleted=True) + + result = [entry async for entry in stream] + + assert [entry.id for entry in result] == ["ID-1"] + assert all(isinstance(entry, NullableFieldsModel) for entry in result) + + +@respx.mock +async def test_async_skip_deleted_reports_progress( + async_nullable_fields_service, deletion_stub_record, async_recording_progress, data_record +): + records = [data_record, deletion_stub_record] + respx.get(STREAM_URL).mock(return_value=records_response(records)) + stream = async_nullable_fields_service.stream( + progress=async_recording_progress, skip_deleted=True + ) + + [entry async for entry in stream] # act + + assert async_recording_progress.events == [ + ("item_processed",), + ("item_processed",), + ("completed",), + ] + + +@respx.mock +async def test_async_skip_deleted_verifies_count( + async_nullable_fields_service, deletion_stub_record, data_record +): + records = [data_record, deletion_stub_record] + respx.get(STREAM_URL).mock(return_value=records_response(records, item_count="3")) + iterator = async_nullable_fields_service.stream(skip_deleted=True) + + with pytest.raises(MPTStreamingIncompleteError, match=COUNT_MISMATCH_MATCH): + [entry async for entry in iterator] + + def test_deserialize_stream_record_builds_a_model(data_record): result = deserialize_stream_record(data_record, NullableFieldsModel) @@ -915,6 +1011,26 @@ def test_stream_envelope_keeps_deleted_status(nullable_fields_service, deleted_s assert isinstance(result[0], NullableFieldsModel) +@respx.mock +def test_stream_envelope_skip_deleted_total( + nullable_fields_service, data_record, deletion_stub_record, recording_progress +): + body = envelope_body([data_record, deletion_stub_record], total=2) + respx.get(STREAM_URL).mock(return_value=envelope_response(body)) + stream = nullable_fields_service.stream( + stream_format=StreamFormat.JSON, progress=recording_progress, skip_deleted=True + ) + + list(stream) # act + + assert recording_progress.events == [ + ("set_total_items", 2), + ("item_processed",), + ("item_processed",), + ("completed",), + ] + + @respx.mock def test_stream_envelope_raises_on_count_mismatch(streaming_service, data_record): body = envelope_body([data_record], total=1)