MPT-24785 Let stream() consumers opt out of deletion stubs

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

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:

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,13 +437,47 @@ class StreamingMixin[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: 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,14 +570,28 @@ 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()
 
     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,13 +616,47 @@ 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,
         *,
         limit: int | None = None,
         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):