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)