From 47f23501a8e19375379a0265c6569a854c224b3c Mon Sep 17 00:00:00 2001 From: Lukasz Lancucki Date: Wed, 2 Sep 2026 09:26:30 +0100 Subject: [PATCH] docs: add the streaming guide Add docs/streaming.md as the single place a stream consumer is documented. It covers stream() versus iterate() and when each is the right read, the memory characteristics of both, what the platform guarantees about a stream, both wire formats, limit semantics, and the three obligations a consumer cannot skip: verifying completeness against MPT-Item-Count, branching on DeletionStub before ingesting a record, and restarting rather than resuming a failed export. The client-timeout trap gets its own section, because a deferred first byte is bounded by the read timeout rather than the connect timeout and a default-timeout client fails on a large export that is working correctly. Choosing The Wire Format covers StreamFormat.JSONL and StreamFormat.JSON: what differs between them, where only the envelope carries $meta.pagination.total and therefore reports a progress total, and what is format-independent. $meta.pagination.total equals MPT-Item-Count but precedes the data, so it is a total to display rather than a completeness check. The deletion-stub examples establish a fresh attempt key, stage under it, and promote after the loop rather than writing each record as it arrives. stream() yields records before it can verify the count, so applying them directly leaves partial local state behind on a truncated export. A retry opens a new snapshot, so it takes a new key rather than reusing one. Cleanup catches broadly and re-raises, because a malformed body raises json.JSONDecodeError and a consumer's own staging code can fail too; either way promotion is skipped and durable staging would otherwise survive indefinitely. The guidance names the trap the pattern invites: stage somewhere durable rather than accumulating the export in a list, which would give up the memory bound. Opting Out Of Deletion Stubs documents skip_deleted: the overload-typed return, that filtering happens after the client's bookkeeping so completeness and progress are unaffected, that a caller's own object count no longer matches MPT-Item-Count in that mode, and that a mirroring consumer must keep the default or lose upstream deletions. Reduce the Streaming Large Result Sets section of docs/usage.md to an orientation and a link, so the detail lives in one document rather than two, relocating its wire-format and opt-out material into the guide rather than dropping it. Record that ownership split in docs/documentation.md, which named docs/usage.md as the source of truth for all examples and did not list the guide at all, so a later edit would have landed in the wrong file and recreated the drift. Index the guide from README.md and AGENTS.md. Add MPTStreamingFormatMismatchError to the exception hierarchy in docs/architecture.md, which had not recorded it, and correct two statements there and in docs/usage.md that this change made inconsistent: completeness counts raw records consumed rather than records yielded, and a streaming read is bounded by the larger of stream_read_timeout and read_timeout rather than substituting one for the other. MPT-24259 Co-Authored-By: Claude Opus 5 --- AGENTS.md | 1 + README.md | 1 + docs/architecture.md | 13 +- docs/documentation.md | 9 +- docs/streaming.md | 631 ++++++++++++++++++++++++++++++++++++++++++ docs/usage.md | 318 ++++----------------- 6 files changed, 701 insertions(+), 272 deletions(-) create mode 100644 docs/streaming.md diff --git a/AGENTS.md b/AGENTS.md index c8a07e22..d8f024ff 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -25,6 +25,7 @@ When applicable, read the repository documentation in this order: 7. `docs/documentation.md` — repository-specific documentation rules 8. `docs/unit_tests.md` — unit test structure and guidance 9. `docs/e2e_tests.md` — end-to-end test setup and execution +10. `docs/streaming.md` — streaming read mode, consumer obligations, and timeouts Then inspect the code paths relevant to the task: diff --git a/README.md b/README.md index 2067ab27..f5708b36 100644 --- a/README.md +++ b/README.md @@ -19,6 +19,7 @@ Start with these documents: - [docs/e2e_tests.md](docs/e2e_tests.md): end-to-end test setup and execution - [docs/contributing.md](docs/contributing.md): repository-specific workflow and links to shared standards - [docs/documentation.md](docs/documentation.md): repository-specific documentation rules +- [docs/streaming.md](docs/streaming.md): streaming guide — `stream()` versus `iterate()`, consumer obligations, and timeouts - [docs/rql.md](docs/rql.md): fluent RQL query builder guide - [MPT OpenAPI Spec](https://api.s1.show/public/v1/openapi.json): upstream API contract diff --git a/docs/architecture.md b/docs/architecture.md index 61a3de28..cc2dd792 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -154,8 +154,9 @@ streaming read mode on a regular collection route and require the API to echo th `MPT-Streaming` response header, raising `MPTStreamingNotEnabledError` when it does not. They also verify completeness: the declared `MPT-Item-Count` is read before the first record — raising `MPTStreamingItemCountMissingError` when absent or unusable — and compared with the -yielded record count once the body is fully consumed, raising `MPTStreamingIncompleteError` -on mismatch. An iterator closed early skips the comparison. A record marked with +count of raw records consumed once the body is fully consumed, raising +`MPTStreamingIncompleteError` on mismatch. The count is taken before `skip_deleted` withholds +any stub, so a filtered stub still counts. 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. A consumer that ingests no deletions can opt out with the keyword-only @@ -185,6 +186,9 @@ unchanged, omitting whichever is unset. Validating them locally is deliberately the server owns pagination-input validation, and the inputs it accepts in streaming mode are still changing. +See [the streaming guide](streaming.md) for the consumer-facing contract these mixins +implement. + Example service definition: ```python @@ -216,8 +220,8 @@ Transport-level settings (`base_url`, `timeout`, `retries`) are grouped in the constructors as `transport=TransportSettings(...)`. Timeouts resolve per connection phase: `connect_timeout`, `read_timeout`, `write_timeout` and `pool_timeout` each fall back to `timeout`, and the dataclass exposes two profiles — `request_timeout` for regular requests and -`stream_timeout`, which substitutes the longer `stream_read_timeout` for the read phase because -a streamed response defers its first byte until the server has built the result set. To resolve the base URL from the +`stream_timeout`, whose read phase is the larger of `stream_read_timeout` and `read_timeout`, +because a streamed response defers its first byte until the server has built the result set. To resolve the base URL from the `MPT_API_BASE_URL` environment variable instead, pass `EnvTransportSettings()` (the default when no transport is given); the clients themselves never read the environment. The resolved settings are handed to the authentication provider through @@ -254,6 +258,7 @@ Client, transport, and API errors use the following hierarchy: MPTError ├── MPTStreamingError # base for streaming-mode failures │ ├── MPTStreamingNotEnabledError # response did not confirm streaming mode +│ ├── MPTStreamingFormatMismatchError # Content-Type differed from the requested format │ ├── MPTStreamingItemCountMissingError # no usable MPT-Item-Count declared │ ├── MPTStreamingIncompleteError # record count differed from MPT-Item-Count │ └── MPTStreamingTruncatedError # body ended before the HTTP message completed diff --git a/docs/documentation.md b/docs/documentation.md index 7b90d4ac..63a512ce 100644 --- a/docs/documentation.md +++ b/docs/documentation.md @@ -13,13 +13,20 @@ This file documents repository-specific documentation rules only. - Topic-specific documentation must live in the matching file under [`docs/`](.). - Shared engineering rules must be linked from `mpt-extension-skills` instead of copied into this repository. - When changing setup, usage, testing, or architecture behavior, update the corresponding document in the same change. -- `docs/usage.md` is the source of truth for installation, configuration, examples, and supported command entry points. +- `docs/usage.md` is the source of truth for installation, configuration, general usage + examples, and supported command entry points. +- `docs/streaming.md` is the source of truth for the streaming read mode: when to stream, + the wire formats, consumer obligations, timeouts, and streaming examples. `docs/usage.md` + carries only a short orientation and links here, so streaming guidance must not be + duplicated there. ## Current Documentation Map - [`README.md`](../README.md): overview, quick start, and documentation map - [`AGENTS.md`](../AGENTS.md): AI-agent entry point and reading order - [`usage.md`](usage.md): install, configure, and use the client +- [`streaming.md`](streaming.md): streaming read mode — access pattern, wire formats, + consumer obligations, and timeouts - [`architecture.md`](architecture.md): repository structure and major abstractions - [`local-development.md`](local-development.md): Docker-only local setup and execution - [`testing.md`](testing.md): repository-specific testing strategy diff --git a/docs/streaming.md b/docs/streaming.md new file mode 100644 index 00000000..3efafa49 --- /dev/null +++ b/docs/streaming.md @@ -0,0 +1,631 @@ +# Streaming + +This guide is for developers who need to read a large result set out of the MPT API in one +pass. It covers when to stream instead of paging, what the platform guarantees about a +stream, and the obligations a consumer must meet to read one correctly. + +For installation, client construction, and the general sync and async patterns, see +[usage.md](usage.md). For where the streaming mixins and exceptions sit in the codebase, see +[architecture.md](architecture.md). + +## `stream()` Versus `iterate()` + +Both read a whole collection. They differ in how the platform produces the result and in +what the client can guarantee about it. + +| | `iterate()` | `stream()` | +|---|---|---| +| Requests | One request per page | One request for the whole export | +| Read mode | Regular paged read | Streaming mode, opted into with `MPT-Streaming: true` | +| Membership | Re-evaluated on every page | Fixed once, when the stream opens | +| Response format | `application/json` page envelope | either wire format, chosen per request | +| Completeness | Not verifiable | Verified against `MPT-Item-Count` | +| Deleted members | Absent from later pages | Emitted as a `DeletionStub` | +| Recovery from failure | Re-fetch the failed page | Restart the whole export | +| Peak memory | One page | One record | + +Use `iterate()` when you want the collection as it is right now and you will consume all of +it promptly: it is the plain paged read, and a failure costs one page. Use it also for +endpoints that do not stream, and as the fallback when streaming is refused. + +Use `stream()` when you need a consistent export rather than a live read — a nightly sync, a +reconciliation job, a bulk load into another system. Streaming asks the platform for a +point-in-time export, so the result set does not shift underneath you while you read it, and +the client can tell you whether you received all of it. + +`stream()` is not a faster `iterate()` for small reads. It costs the platform a key scan +before the first byte and it costs you the three obligations below. For a few hundred +records, page. + +### Every Collection Service Streams + +`CollectionMixin` and `AsyncCollectionMixin` inherit `StreamingMixin` and +`AsyncStreamingMixin`, so `stream()` is available on every collection service without +per-service wiring: + +```python +from mpt_api_client import BearerTokenAuthentication, MPTClient, RQLQuery + +client = MPTClient.from_config( + authentication=BearerTokenAuthentication(""), + base_url="https://api.s1.show/public", +) + +for order in client.commerce.orders.filter(RQLQuery(status="Processing")).stream(): + print(order.id) +``` + +Streaming mode is a property of the request, not of the service. The route is the ordinary +collection route, so `filter()`, `order_by()` and `select()` chain before `stream()` exactly +as they do before `iterate()`; the client turns the read into a stream by sending the +`MPT-Streaming` header. + +You can assume a standard collection endpoint streams: the platform's shared framework +provides streaming mode on its standard read controller, and every list endpoint is expected +to support it. Two cases are the exception rather than the rule. A hand-written list action — +one with a mandatory scoping filter, or one that already assigns a media type its own meaning +— opts in explicitly. And an endpoint that has not picked up the rollout answers `501`. + +That `501` is an ordinary, expected answer rather than a fault, surfaced as +`MPTStreamingNotSupportedError`. Handle it by falling back to `iterate()`, not by checking +endpoints up front: + +```python +from mpt_api_client.exceptions import MPTStreamingNotSupportedError + +try: + records = list(client.commerce.orders.stream()) +except MPTStreamingNotSupportedError: + records = list(client.commerce.orders.iterate()) +``` + +### Do Not Confuse `stream()` With `stream_jsonl()` + +The two methods look alike and mean different things. + +| | `stream()` | `stream_jsonl()` | +|---|---|---| +| Contract | The platform streaming read mode | An endpoint's own JSONL download | +| `MPT-Streaming` header | Sent, and the response must confirm it | Not sent | +| Completeness check | Yes | No | +| Yields | Models or `DeletionStub` objects | Models only | +| Availability | Every collection service | Composed explicitly, today only by billing statement charges | + +A service can carry both, and billing statement charges does: `stream_jsonl()` is its JSONL +download, while `stream()` is the streaming-mode read it inherits with every other +collection service. Reach for `stream_jsonl()` only when you specifically want that +endpoint's JSONL contract; for everything else `stream()` is the streaming read. + +## Memory Characteristics + +`stream()` holds one record at a time. The response body is parsed as it arrives and each +record is deserialized, yielded, and dropped, so peak memory is set by the largest single +record rather than by the size of the export. A ten-million-record stream costs the same as a +ten-record one. This holds in both wire formats — the envelope is tokenized incrementally +rather than buffered. + +The buffering paths, for contrast: + +- `iterate()` holds one page. It buffers each page response in full, deserializes it into a + `ModelCollection`, and yields from that before fetching the next — so peak memory is + `batch_size` records, 100 by default. +- `fetch_page()` and `fetch_one()` buffer a single response and return it whole. +- Any call that materializes the iterator — wrapping a stream in `list()`, a comprehension, + a `sorted()` — buffers the entire result set and gives up the bound. That is a fine choice + when you know the result is small, and it is the reason the short examples in this guide + use it, but it is a decision to make on purpose rather than by accident. + +The flat profile is a property of the loop, not of the method. Keep the per-record work +inside the `for` body — write, upsert, aggregate — and the whole export stays bounded. + +## What A Stream Is + +The platform builds an export in two phases. Phase one scans keys and fixes the membership +of the export; phase two reads the records for those keys in batches and writes them to the +response body. + +Two consequences follow, and both are contract-conformant behaviour that will otherwise read +as a bug: + +- **Membership is fixed when the stream opens.** Records created after that point are not in + the export, and rows whose filter columns change after that point are not ejected from it. + A stream is a point-in-time export, not a live paged read. +- **Record content is the committed state at batch read time**, so a record's contents can + postdate the moment membership was fixed. There is no cross-record point-in-time + consistency: record A can be newer than record B in the same export. + +A record whose access is revoked mid-export still streams, because security-context filters +are applied when membership is fixed and are not re-applied per batch. + +The longer an export runs over frequently written data, the wider the drift between +membership and content: expect more deletion stubs and more post-snapshot content on a long +export than on a short one. + +## Choosing The Wire Format + +Streaming mode and the wire format are two independent per-request choices. The +`MPT-Streaming` header selects streaming; `Accept` selects the encoding, and `stream()` +exposes it as `stream_format`: + +| `stream_format` | `Accept` | Body | +|---|---|---| +| `StreamFormat.JSONL` (default) | `application/jsonl` | one record object per line, no envelope | +| `StreamFormat.JSON` | `application/json` | the standard `{$meta, data}` envelope, the same shape `iterate()` reads | + +```python +from mpt_api_client.http.mixins import StreamFormat + +for order in client.commerce.orders.stream(stream_format=StreamFormat.JSON): + print(order.id) +``` + +A `StreamFormat` member or its `Accept` string is accepted; any other value raises +`ValueError` before the request is sent, rather than failing deep in header construction. + +**Both formats are parsed as the body arrives**, so the memory bound described above holds +either way. In envelope format the JSON is tokenized incrementally: a record is deserialized +when its own closing brace arrives, not when the envelope completes. + +Keep-alives differ in shape and are invisible either way. The line-delimited format emits +blank lines; the envelope format emits insignificant whitespace between tokens, consumed +while tokenizing. Neither reaches your loop, and neither counts as a record. + +Pick the envelope when you want the total. Only the envelope carries +`$meta.pagination.total`, which equals `MPT-Item-Count` — and is likewise the capped +`min(matches, N)` under a bounded `limit=N`. It reaches a `progress` receiver through +`set_total_items` as soon as `$meta` arrives, so a progress report can render a percentage of +a streamed export. The line-delimited format carries no envelope and never reports a total. + +Pick the line-delimited format when you want the simplest thing to store or pipe: one record +per line survives `split`, `tail` and append-only files, where a single enclosing envelope +does not. + +Everything else is format-independent: query state, `limit` and `offset`, deletion stubs, the +completeness check against `MPT-Item-Count`, and every streaming error. + +## Bounding An Export + +`limit` selects between the whole snapshot and a bounded prefix of it: + +| `limit` | Meaning | +|---|---| +| absent (default) | The full snapshot | +| `-1` | The same thing, stated explicitly | +| `N` | The first `N` records of the stream order | + +Under a bounded `limit=N`, the count the response declares is the capped count, +`K = min(matches, N)` — not the uncapped number of matches. Both carriers agree: +`MPT-Item-Count` and, in the envelope format, `$meta.pagination.total`. The completeness check +compares against that capped value, so a bounded export verifies exactly like a full one. + +```python +for order in client.commerce.orders.order_by("-audit.created.at").stream(limit=100_000): + print(order.id) +``` + +`stream()` also accepts `offset`. Pagination inputs are sent exactly as given and are never +checked locally, because the server owns their validation: it currently rejects `offset` in +streaming mode with `400`, and support for it is scheduled. Passing through is correct either +way, so no client release is coupled to that change. + +## Three Obligations You Cannot Skip + +Streaming trades the safety of paging for a single-pass export. These three checks are what +you take on in exchange. Each prevents a failure that is silent without it. + +### 1. Verify Completeness Against `MPT-Item-Count` + +**Prevents:** processing a truncated export as if it were the whole result set. + +Streaming mode commits the `MPT-Item-Count` response header together with the status: the +number of records the stream will carry. It is the contract's only completeness signal. The +envelope format's `$meta.pagination.total` carries the same number, but it precedes the data +and so cannot attest that the data arrived — it is a total to display, not a check to make. + +`stream()` performs this check for you. It reads the declared count before yielding the first +record and compares it with the number of raw records consumed when the body ends — +consumed, not yielded, because `skip_deleted` filtering happens after this accounting: + +- No usable count declared → `MPTStreamingItemCountMissingError`, raised before the body is + read, so no partial data is consumed. +- Count mismatch on a fully consumed body → `MPTStreamingIncompleteError`. + +A mismatch means the stream terminated gracefully but did not carry what it promised — an +intermediary swallowed part of the body, or the export was cut short. The records you +received are not a valid subset to keep, because you cannot tell which ones are missing. +Discard them and re-run the export. + +Closing the iterator early does not raise. The check applies only to a stream consumed to +the end, so `break`-ing out of the loop on purpose is not reported as an incomplete export: + +```python +for order in client.commerce.orders.stream(): + if order.id == "ORD-0000-0001": + break # deliberate early exit, no completeness check +``` + +The count does not survive persisting the payload. If you write the raw records to storage +and verify them later, store the expected count alongside them — once the response is gone, +the export's own completeness signal is gone with it. + +Getting that number takes a deliberate step, because `stream()` verifies the count for you and +does not hand it over: it yields records, not headers. Either stream `StreamFormat.JSON`, where +the envelope's total reaches a `progress` receiver through `set_total_items`, or drop to +`client.http_client.stream(...)` and read the `MPT-Item-Count` header yourself — which means +parsing and verifying the body yourself too, so prefer the first. + +### 2. Check For A Deletion Stub Before Ingesting A Record + +**Prevents:** overwriting a live stored record with nulls. + +A member of the snapshot whose row is hard-deleted before phase two reaches it is still a +member of the export, so the platform emits a deletion stub in its place, marked with +`$meta.deleted` — the platform's metadata channel for the signal, and the name the API +contract uses for it: + +```json +{"id": "ORD-1234-5678", "$meta": {"deleted": true}} +``` + +Only `id` is guaranteed on a stub. No other property of the deleted row is carried. A +**truthy** `deleted` marker is what identifies a stub; a record with no `$meta`, no `deleted` +key, or a falsy one is data. In practice the platform omits `$meta` entirely on a normal +record, so those cases are defensive rather than expected. + +`stream()` yields these as `DeletionStub`, never as a model, so the object cannot be mistaken +for a record by code that expects one. Deserializing a stub as a model would produce an +instance whose every declared field is `None` — indistinguishable from a record whose values +really are unset — and a sync job writing that back would overwrite the stored record with +nulls. Branch on the type before ingesting the object; there is nothing else on a stub to +inspect: + +```python +import uuid + +from mpt_api_client.models import DeletionStub + +# A fresh key per attempt. A retry opens a new snapshot, so it must never reuse one. +attempt_id = uuid.uuid4().hex + +try: + for result in client.commerce.orders.stream(): + if isinstance(result, DeletionStub): + stage_delete(attempt_id, result.id) + else: + stage_upsert(attempt_id, result) + + # Reached only once stream() has verified the record count against MPT-Item-Count. + promote(attempt_id) +except Exception: + discard(attempt_id) + raise +``` + +Note what the loop does *not* do: it stages rather than writes. `stream()` yields records +before it can verify the count, so applying each record as it arrives leaves partial local +state behind on a truncated export — the one thing +[Restart, Do Not Resume](#3-restart-do-not-resume) says must not survive a failed attempt. +Stage under an attempt key and promote after the loop, which is reached only on a verified +export. + +Catch broadly rather than on `MPTStreamingError`, because not every failure that can strand a +staged attempt is one. A malformed body raises `json.JSONDecodeError`, and your own staging +code can fail too; either way promotion is skipped, and without the cleanup the durable +staging survives indefinitely. The `raise` matters as much as the `discard`: the caller still +has to learn the export failed. + +Stage somewhere durable — a staging table, a temp file, a keyed batch — not a Python list. +Accumulating the export in memory to promote it later gives up the bound that made streaming +worth using, which is the trap this pattern invites. + +Three properties of stubs matter for correctness: + +- **Stubs are counted, not filtered.** Every member of the export accounts for exactly one + record, stubs included, so a stub counts towards `MPT-Item-Count`. Do not drop stubs before + the completeness check — a complete export would read as short. The safe way to drop them is + [`skip_deleted`](#opting-out-of-deletion-stubs), which filters after that accounting has + been fed, so a withheld stub still counts. +- **A stub is not a `DELETED` status.** A `DELETED` status is a domain state on a full, + existing record, and stays a model. A stub marks a row that no longer exists at all and + carries no state. Conflating them either loses a deletion or discards a live record. +- **A stub does not satisfy a record schema.** If you validate incoming payloads against a + strict schema, branch on `DeletionStub` first and skip the record validation for stubs, + rather than loosening the schema for records too. + +Anything that reads a field other than `id` needs this branch. A loop that only reads +`order.id` is safe as written, because every streamed object carries an `id` — but it is one +edit away from not being safe. + +#### Opting Out Of Deletion Stubs + +A consumer that never ingests deletions — read-only analytics, an ad-hoc export — gains +nothing from the branch: it would drop the stubs and move on. Declare that with the +keyword-only `skip_deleted` flag instead of writing a branch that discards: + +```python +for order in client.commerce.orders.stream(skip_deleted=True): + print(order.id) +``` + +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 through its own 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 guarantees above +survive it: + +- Completeness still counts the raw records ahead of the filter and still compares them with + `MPT-Item-Count` once the body is consumed. A short stream raises + `MPTStreamingIncompleteError` either way. +- A `progress` receiver still gets `item_processed` for every record, withheld stubs + included, because the declared total counts stubs — a report fed only visible records + would never reach 100%. + +One consequence to plan for: under `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 the full snapshot +arrived, which is the check that matters. + +Opting out is a statement that deletions are irrelevant to this consumer, not a shortcut. A +job that mirrors the collection must keep the default and branch, or members deleted upstream +survive locally forever — the same silent divergence this obligation exists to prevent. + +### 3. Restart, Do Not Resume + +**Prevents:** splicing two different snapshots into one result set. + +Resume is a contract non-goal. A retry is a new request, which opens a *new* membership +snapshot, so records from a failed attempt cannot be continued or appended to a later one. +Discard everything the failed attempt produced and restart from scratch: + +```python +from mpt_api_client.exceptions import ( + MPTStreamingIncompleteError, + MPTStreamingTruncatedError, +) + +try: + records = list(client.commerce.orders.stream()) +except (MPTStreamingIncompleteError, MPTStreamingTruncatedError): + records = list(client.commerce.orders.stream()) # a new snapshot, not a continuation +``` + +`MPTStreamingTruncatedError` is how a mid-stream failure arrives: the API signals an internal +failure by aborting the connection without completing the HTTP message, so the transport +failure is the failure signal. It is raised once the response has opened — usually after +records have reached your loop, but equally when the body dies before the first one. It is +not retry exhaustion: transparent retry runs while the response is being opened and cannot +re-request once the body has started, so `MPTMaxRetryError` stays reserved for a request that +never delivered a body at all. + +The same rule covers `MPTStreamingIncompleteError`: whatever the cause, a failed export is +discarded whole and re-requested, never patched. + +If restarting a large export is expensive, write records to a staging area keyed by the +attempt and promote it only after the stream completes without raising. That keeps the +discard cheap and keeps a failed attempt from reaching the records your consumers read. + +## The Client-Timeout Trap + +This is the failure most likely to be misdiagnosed, because a correctly working export looks +exactly like a broken one. + +Phase one scans keys before a single byte of the body is written, and the platform's SLO +allows that to take up to **60 seconds** on a large export. During that time the connection +is established and the client is waiting for the response to start. + +**It is the read timeout, not the connect timeout, that bounds a deferred first byte.** A +client tuned with a generous connect timeout and a short read timeout will abandon a working +export mid-scan and report a timeout that looks like a server failure. + +`TransportSettings` exposes `stream_read_timeout` for exactly this. A streaming request's +read phase is bounded by the **larger** of `stream_read_timeout` and `read_timeout`, and +`stream_read_timeout` defaults to `120.0` — enough for the 60s SLO with headroom, where the +regular `read_timeout` default is not: + +```python +from mpt_api_client import BearerTokenAuthentication, MPTClient, TransportSettings +from mpt_api_client.http import HTTPClient + +client = MPTClient( + http_client=HTTPClient( + transport=TransportSettings( + base_url="https://api.s1.show/public", + timeout=20.0, + connect_timeout=5.0, + read_timeout=30.0, + stream_read_timeout=180.0, + ), + authentication=BearerTokenAuthentication(""), + ) +) +``` + +Three things to know about it: + +- Because the larger value wins, raising `read_timeout` raises the streaming budget too, and + streaming can never end up with the shorter of the two. +- It bounds a single read, not the whole export. No total-duration timeout is applied: an + export runs for as long as the server keeps sending. Server-side keep-alives count as data + and reset the read clock — blank lines in the line-delimited format, insignificant + whitespace between tokens in the envelope. +- Lowering it below the SLO is the trap. If you tune timeouts down for a low-latency service + and set `stream_read_timeout` from the same budget as your regular calls, large exports + start failing while small ones keep working — which reads as a size-dependent server bug + rather than a client setting. + +Anything between your client and the platform needs the same treatment, and where the cut +falls decides which error you get: + +- **Before the response headers arrive** — a proxy or load balancer whose first-byte timeout + is shorter than the phase-one budget cuts the connection while the key scan is still + running. Nothing has been committed yet, so transparent retry applies and the failure + surfaces as `MPTMaxRetryError`, not as a truncated stream. +- **After the headers arrive** — a cut during the body is `MPTStreamingTruncatedError`, even + if no record reached your loop, because the response had already committed its status. + +See [Timeouts](usage.md#timeouts) for the full per-phase timeout model. + +## Streaming Errors + +All streaming-specific failures derive from `MPTStreamingError`, so one handler covers them: + +| Exception | Raised when | +|---|---| +| `MPTStreamingNotEnabledError` | The response does not echo `MPT-Streaming`, so the body is an ordinary paged response | +| `MPTStreamingFormatMismatchError` | The response `Content-Type` names a media type other than the requested wire format | +| `MPTStreamingNotSupportedError` | `501` — the resource provides no streaming-capable execution strategy | +| `MPTStreamingNotAcceptableError` | `406` — the requested format cannot be served for this read mode | +| `MPTStreamingOverCapError` | `413` — the result set exceeds the configured `MaxExportKeys` cap | +| `MPTStreamingItemCountMissingError` | The response declares no usable `MPT-Item-Count`, so completeness cannot be verified | +| `MPTStreamingIncompleteError` | The fully consumed stream carried a different number of records than `MPT-Item-Count` declared | +| `MPTStreamingTruncatedError` | The connection was aborted mid-body, so the response ended before the HTTP message completed | + +```python +import logging + +from mpt_api_client.exceptions import MPTStreamingError + +logger = logging.getLogger(__name__) + +try: + for order in client.commerce.orders.stream(): + print(order.id) +except MPTStreamingError as error: + logger.error("Streaming unavailable: %s", error) +``` + +Split them by what a caller can do about them: + +- **Request-shape and endpoint-support failures** — `MPTStreamingNotEnabledError`, + `MPTStreamingFormatMismatchError`, `MPTStreamingNotSupportedError`, + `MPTStreamingNotAcceptableError`, `MPTStreamingOverCapError` and + `MPTStreamingItemCountMissingError`. Retrying the same call against the same endpoint fails + the same way. Change the request, or fall back to `iterate()`. A `406` now genuinely can + mean a format you asked for and the endpoint cannot serve, so check `stream_format` before + assuming the endpoint is at fault. +- **Incomplete-export failures** — `MPTStreamingIncompleteError` and + `MPTStreamingTruncatedError`. A retry can succeed, but only as a fresh export; see + [Restart, Do Not Resume](#3-restart-do-not-resume). + +`MPTStreamingNotEnabledError` and `MPTStreamingItemCountMissingError` are raised before the +body is read, so no partial data is consumed. The three HTTP-backed types also subclass +`MPTHttpError`, so existing `except MPTHttpError` handlers keep working and `status_code` +remains available. Any other HTTP status passes through unchanged. + +A body the client cannot parse is not a streaming error at all but a `json.JSONDecodeError` — +a malformed record line in the line-delimited format, a malformed or unterminated envelope in +the envelope format. A body cut short usually loses records before it loses its closing +tokens, so a truncated export normally reports the more precise `MPTStreamingIncompleteError` +instead. + +### Over-Cap Exports + +The API answers `413` when the result set is larger than the configured `MaxExportKeys` cap. +`MPTStreamingOverCapError` keeps the `problem+json` body as structured data on `payload` +rather than flattening it into the message, because the configured cap is the value a caller +acts on: + +```python +import logging + +from mpt_api_client.exceptions import MPTStreamingOverCapError + +logger = logging.getLogger(__name__) + +try: + records = list(client.commerce.orders.stream()) +except MPTStreamingOverCapError as error: + # The body names the configured cap. Read it defensively: payload is {} when the + # response carried no JSON, and the member names are the server's, not the client's. + cap = error.payload.get("maxExportKeys") + logger.error("Export refused (configured cap: %s): %s", cap, error.payload) + # A bounded retry is a different read: the first N of the sort order, not the export. + records = list(client.commerce.orders.stream(limit=10_000)) +``` + +`payload` is an empty mapping when the response carries no JSON body, so treat every member +as optional. The ways forward are the ones the body names: narrow the filter, set an explicit +`limit=N`, or split the export into key or date ranges. + +Two things about that retry are worth being deliberate about: + +- **A bounded retry does not get you the export.** `limit=N` returns the first `N` records of + the sort order, and `MPT-Item-Count` reports the capped `K`, so the result is complete as a + stream but is a prefix of the set you asked for. If you need all of it, narrow the filter or + split by key or date range instead — those return the whole set in pieces. +- **Do not derive the limit from the cap.** A limit equal to `maxExportKeys` asks for the + largest export the server will permit, which is the request that just failed for being too + big to be useful. Pick a size the consumer can actually process; the cap tells you the + request was refused, not what to ask for next. + +## Async Streaming + +The async form is the same contract over an async generator. Consume it inside a coroutine — +a top-level `async for` is a syntax error: + +```python +import asyncio +import uuid + +from mpt_api_client import AsyncMPTClient, BearerTokenAuthentication +from mpt_api_client.models import DeletionStub + + +async def export_orders() -> None: + client = AsyncMPTClient.from_config( + authentication=BearerTokenAuthentication(""), + base_url="https://api.s1.show/public", + ) + attempt_id = uuid.uuid4().hex + + try: + async for result in client.commerce.orders.stream(): + if isinstance(result, DeletionStub): + await stage_delete(attempt_id, result.id) + else: + await stage_upsert(attempt_id, result) + + await promote(attempt_id) + except Exception: + await discard(attempt_id) + raise + + +asyncio.run(export_orders()) +``` + +Everything above applies unchanged: the same headers, the same completeness check, the same +stubs, the same restart rule. Only the `Async*` mixins and an `await`ed body differ. + +## Reporting Progress + +A long export gives no feedback by default. `stream()` accepts an optional `progress` +receiver, called once per consumed record — including a stub withheld by `skip_deleted` — +and once on completion: + +```python +from mpt_api_client.models import ConsoleProgress + +for order in client.commerce.orders.stream(progress=ConsoleProgress()): + print(order.id) +``` + +Whether a total is reported depends on the wire format. In the envelope format +`set_total_items` receives `$meta.pagination.total` as soon as `$meta` arrives, so +`ConsoleProgress` can render a real percentage. In the default line-delimited format there is +no envelope and no total, so `ConsoleProgress` renders its total as `0` — the record count is +then the meaningful half of the line. + +A progress receiver never sees `MPT-Item-Count`. The client reads that header to verify the +export and does not pass it on, and `set_total_items` is called only when the *response* +reports a total. So a percentage needs either `StreamFormat.JSON`, or a total you already +know from elsewhere. Implement the `Progress` protocol (or `AsyncProgress` for the async +client) to route progress somewhere other than the console. + +## Related Documents + +- [usage.md](usage.md): installation, client construction, timeouts, and general usage +- [architecture.md](architecture.md): where the streaming mixins and exceptions live +- [rql.md](rql.md): building the filters a stream applies diff --git a/docs/usage.md b/docs/usage.md index 55ededbd..fe96332c 100644 --- a/docs/usage.md +++ b/docs/usage.md @@ -100,15 +100,18 @@ Two things are worth knowing: - The **read** timeout, not the connect timeout, governs how long the client waits for a response to start arriving. A server that accepts the connection and then thinks before replying is bounded by `read_timeout`. -- Streaming requests use `stream_read_timeout` (default `120.0`) in place of the regular read - timeout, because a streamed response commits its status only after the server has built the - result set — so the first byte can be deferred far longer than for a regular call. The - effective streaming read timeout is never lower than `read_timeout`, so raising that raises - both. +- A streaming request's read phase is bounded by the **larger** of `stream_read_timeout` + (default `120.0`) and `read_timeout`, because a streamed response commits its status only + after the server has built the result set — so the first byte can be deferred far longer + than for a regular call. Raising `read_timeout` therefore raises the streaming budget too. No total-duration timeout is applied. A long export runs for as long as the server keeps sending; the limits are per phase, not overall. +Getting this wrong is the most commonly misdiagnosed streaming failure, because a working +export then looks like a broken one. See +[The Client-Timeout Trap](streaming.md#the-client-timeout-trap). + ## Synchronous Usage Patterns Read a single resource: @@ -162,8 +165,12 @@ and `BatchProgressReport` once every `batch_size` records. Both track the count total for you — implement only `report(current, total, *, completed)`: ```python +import logging + from mpt_api_client.models import BatchProgressReport +logger = logging.getLogger(__name__) + class LogProgress(BatchProgressReport): def report(self, current, total, *, completed): @@ -184,8 +191,9 @@ The `progress` parameter is also accepted by `stream()` and `stream_jsonl()` des [Streaming Large Result Sets](#streaming-large-result-sets); there `set_total_items` is called only when the envelope wire format reports `$meta.pagination.total`, and never in the line-delimited format, which carries no envelope — see -[Choosing The Wire Format](#choosing-the-wire-format) — so design progress implementations -to work while the total is still unknown. The async `iterate()`, `stream()` and `stream_jsonl()` accept an +[Choosing The Wire Format](streaming.md#choosing-the-wire-format) — so design progress +implementations to work while the total is still unknown. The async `iterate()`, `stream()` +and `stream_jsonl()` accept an `AsyncProgress` implementation whose methods are `async def` and are awaited — `AsyncConsoleProgress` is the shipped counterpart, with `AsyncProgressReport`, `AsyncTimeProgressReport`, and `AsyncBatchProgressReport` as the async abstract bases. @@ -225,11 +233,11 @@ time without buffering the whole body, so memory stays flat regardless of result out of the box — no extra composition is needed: ```python -from mpt_api_client import MPTClient, BearerTokenAuthentication, RQLQuery +from mpt_api_client import BearerTokenAuthentication, MPTClient, RQLQuery client = MPTClient.from_config( - authentication=BearerTokenAuthentication("your-token"), - base_url="https://api.example.com", + authentication=BearerTokenAuthentication(""), + base_url="https://api.s1.show/public", ) service = client.commerce.orders @@ -241,70 +249,20 @@ Streaming mixins extend `QueryableMixin`, so `filter()`, `order_by()` and `selec before `stream()` exactly as they do before `iterate()`. Membership is fixed when the stream opens: records created afterwards are not included. -The minimal loop above reads `id`, which every streamed object carries. Anything that touches -other fields must first check for a deletion stub: a member deleted after the snapshot arrives -as a `DeletionStub`, not a model. See [Deletion Stubs](#deletion-stubs) — the async example -below shows the branch. - -### Choosing The Wire Format - -The same records travel in either of two formats, chosen per request with `Accept`: - -| `stream_format` | `Accept` | Body | -|---|---|---| -| `StreamFormat.JSONL` (default) | `application/jsonl` | one record object per line, no envelope | -| `StreamFormat.JSON` | `application/json` | the standard `{$meta, data}` envelope, the same shape `iterate()` reads | - -```python -from mpt_api_client.http.mixins import StreamFormat - -for order in service.stream(stream_format=StreamFormat.JSON): - print(order.id) -``` - -Both formats are parsed as the body arrives, so a record is yielded while the rest of the -export is still on the wire and the whole body is never held in memory. In envelope format -that means the JSON is tokenized incrementally: each record is deserialized when its own -closing brace arrives, not when the envelope completes. The insignificant whitespace a -streaming response emits between tokens as a keep-alive is consumed while tokenizing — the -envelope-format equivalent of the blank keep-alive lines of the line-delimited format — so it -never reaches your loop. - -Only the envelope carries `$meta.pagination.total`, which equals the `MPT-Item-Count` value -and is likewise the capped `min(matches, N)` under a bounded `limit=N`. It is reported to a -`progress` receiver through `set_total_items`, exactly as `iterate()` reports the total of -each page, so a progress report can render a percentage of a streamed export. The -line-delimited format carries no envelope and therefore never calls `set_total_items`. The -total is reported as soon as `$meta` arrives, which precedes the records only when the server -sends `$meta` first. - -Everything else is format-independent: query state, `limit` and `offset`, deletion stubs, the -completeness check against `MPT-Item-Count`, and every streaming error. +The minimal loop above reads `id`, which every streamed object carries; anything that touches +other fields must first branch on `DeletionStub`, as the async example does — or declare that +deletions are irrelevant with `skip_deleted=True`. -### Bounding An Export - -By default `stream()` sends no `limit`, which exports the full snapshot; passing `limit=-1` -requests the same thing explicitly. An explicit `limit=N` bounds the export to the first `N` -records of the stream order, for a "first 100K by this sort" read that does not page: - -```python -for order in service.order_by("-audit.created.at").stream(limit=100_000): - print(order.id) -``` - -Under a bounded limit the counts the response reports — the `MPT-Item-Count` header and -`$meta.pagination.total` — describe the stream itself, `min(matches, N)`, not the uncapped -number of matches. - -`stream()` also accepts `offset`. Pagination inputs are sent exactly as given and are never -checked locally, because the server owns their validation: it currently rejects `offset` in -streaming mode with `400`, and support for it is scheduled. Passing through is correct either -way, so no client release is coupled to that change. +The wire format is a per-request choice — `stream_format=StreamFormat.JSONL` by default, or +`StreamFormat.JSON` for the `{$meta, data}` envelope. Both are parsed incrementally and carry +the same records; see +[Choosing The Wire Format](streaming.md#choosing-the-wire-format) for what differs. The async form yields from an async generator: ```python import asyncio +import uuid from mpt_api_client import AsyncMPTClient, BearerTokenAuthentication from mpt_api_client.models import DeletionStub @@ -316,216 +274,41 @@ async def main(): base_url="https://api.s1.show/public", ) service = client.commerce.orders + attempt_id = uuid.uuid4().hex - async for result in service.stream(): - if isinstance(result, DeletionStub): - await delete_local_record(result.id) - else: - await upsert_local_record(result) - - -asyncio.run(main()) -``` - -### Deletion Stubs - -A member of the snapshot whose row is hard-deleted before the stream reaches it is still a -member of the export, so the platform emits it as a deletion stub instead of a record: - -```json -{"id": "ORD-1234-5678", "$meta": {"deleted": true}} -``` - -`$meta.deleted` is the platform's metadata channel for that signal. A **truthy** `deleted` -marker is what identifies a stub; anything else — no `$meta`, no `deleted` key, or -`"deleted": false` — is data rather than a deletion. In practice the platform omits `$meta` -entirely on a normal record, so the falsy cases are defensive rather than expected. Only `id` -is guaranteed on a stub: no other property of the deleted row is carried. + try: + async for result in service.stream(): + if isinstance(result, DeletionStub): + await stage_delete(attempt_id, result.id) + else: + await stage_upsert(attempt_id, result) -`stream()` yields these as `DeletionStub`, never as a model, so the object cannot be handed -to code that expects a record. Deserializing a stub as a model would instead produce an -instance whose every declared field is `None` — indistinguishable from a record whose values -really are unset — and a sync job writing that back would overwrite the stored record with -nulls. Branch on the type before ingesting the object; there is nothing else on a stub to -inspect: + # Reached only once stream() has verified the export; see the streaming guide. + await promote(attempt_id) + except Exception: + # Any failure strands the staged attempt, not only MPTStreamingError. + await discard(attempt_id) + raise -```python -from mpt_api_client.models import DeletionStub - -for result in service.stream(): - if isinstance(result, DeletionStub): - delete_local_record(result.id) - else: - upsert_local_record(result) -``` - -A stub is not the same thing as a `DELETED` domain status. `DELETED` arrives on a full, -existing record and is a state of that record, so it stays a model; a stub marks a row that -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; 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: - -| Exception | Raised when | -|---|---| -| `MPTStreamingNotEnabledError` | The response does not echo `MPT-Streaming`, so the body is an ordinary paged response | -| `MPTStreamingFormatMismatchError` | The response `Content-Type` names a media type other than the requested wire format | -| `MPTStreamingNotSupportedError` | `501` — the resource provides no streaming-capable execution strategy | -| `MPTStreamingNotAcceptableError` | `406` — the requested format cannot be served for this read mode | -| `MPTStreamingOverCapError` | `413` — the result set exceeds the configured `MaxExportKeys` cap | -| `MPTStreamingItemCountMissingError` | The response declares no usable `MPT-Item-Count`, so completeness cannot be verified | -| `MPTStreamingIncompleteError` | The fully consumed stream yielded a different number of records than `MPT-Item-Count` declared | -| `MPTStreamingTruncatedError` | The connection was aborted mid-body, so the response ended before the HTTP message completed | - -Completeness is verified for you: streaming commits the `MPT-Item-Count` response header -with the status — the number of records the stream will carry, `min(matches, N)` under a -bounded `limit=N` — and `stream()` compares it with the number of records actually yielded -when the body ends. The header is the contract's only completeness signal and it does not -survive persisting the payload, so without this check a truncated export would end as a -silently short result. - -```python -from mpt_api_client.exceptions import MPTStreamingError - -try: - for order in service.stream(): - print(order.id) -except MPTStreamingError as error: - logger.error("Streaming unavailable: %s", error) -``` -Catch the specific types when the response should differ — for example falling back to -`iterate()` on `MPTStreamingNotSupportedError`, but treating -`MPTStreamingNotAcceptableError` as a bug in the request: - -```python -from mpt_api_client.exceptions import MPTStreamingNotSupportedError - -try: - records = list(service.stream()) -except MPTStreamingNotSupportedError: - records = list(service.iterate()) -``` - -Treat the request-shape and endpoint-support failures — `MPTStreamingNotEnabledError`, -`MPTStreamingFormatMismatchError`, `MPTStreamingNotSupportedError`, -`MPTStreamingNotAcceptableError`, `MPTStreamingOverCapError` -and `MPTStreamingItemCountMissingError` — as exactly that rather than transient failures: -retrying the same call against the same endpoint fails the same way, and an over-cap export -needs a narrower request, not a retry. `MPTStreamingIncompleteError` is different: it reports a -stream that terminated gracefully but did not match its declared count, for example after an -intermediary swallowed part of the body. Discard the partial records and re-run the export. -The three HTTP-backed types also subclass `MPTHttpError`, so existing `except MPTHttpError` -handlers keep working and `status_code` remains available. Any other HTTP status passes -through unchanged. - -A body the client cannot parse is not a streaming error but a `json.JSONDecodeError`, in -either format: a malformed record line in the line-delimited format, a malformed or -unterminated envelope in the envelope format. Because a body cut short loses records before -it loses its closing tokens, a truncated envelope normally reports the more precise -`MPTStreamingIncompleteError` instead. - -`MPTStreamingNotEnabledError`, `MPTStreamingFormatMismatchError` and -`MPTStreamingItemCountMissingError` are raised before the -body is read, so no partial data is consumed. A response that omits `Content-Type` is not -treated as a mismatch. `MPTStreamingIncompleteError` can only be raised -once the body has been consumed to the end — records already yielded have been processed by -then, which is why the count must pass before the result set is treated as complete. Closing -the stream early on purpose, such as breaking out of the loop, does not raise: the check -applies only to a stream consumed to completion. - -`MPTStreamingTruncatedError` is the opposite case: the API signals an internal mid-stream -failure by aborting the connection without completing the HTTP message, so it is raised after -records have already been yielded. It is not retry exhaustion — transparent retry runs while -the response is opened, and cannot re-request once the body has started, so `MPTMaxRetryError` -stays reserved for a request that never delivered a body at all. - -Resume is a non-goal: a new request opens a new snapshot, so records from a failed attempt -cannot be spliced onto a later one. Discard everything the failed attempt produced and restart -the export from scratch: - -```python -from mpt_api_client.exceptions import MPTStreamingTruncatedError - -try: - records = list(service.stream()) -except MPTStreamingTruncatedError: - records = list(service.stream()) # a new snapshot, not a continuation -``` - -### Over-Cap Exports - -The API answers `413` when the result set is larger than the configured `MaxExportKeys` cap. -`MPTStreamingOverCapError` keeps the `problem+json` body as structured data on `payload` -rather than flattening it into the message, because the configured cap is the value a caller -acts on: - -```python -from mpt_api_client.exceptions import MPTStreamingOverCapError - -try: - records = list(service.stream()) -except MPTStreamingOverCapError as error: - logger.error("Export refused: %s", error.payload) - records = list(service.stream(limit=10_000)) +asyncio.run(main()) ``` -`payload` is an empty mapping when the response carries no JSON body, so read it defensively. -The ways forward are the ones the body names: narrow the filter, set an explicit `limit=N`, or -split the export into key or date ranges. +**Read [the streaming guide](streaming.md) before shipping a stream consumer.** It covers +when to stream instead of paging, both wire formats, `limit` semantics, the streaming +exceptions, and the three obligations a consumer cannot skip — verifying completeness against +`MPT-Item-Count`, handling deletion stubs (and when +[opting out](streaming.md#opting-out-of-deletion-stubs) is legitimate), and restarting rather +than resuming a failed export — plus the timeout setting that decides whether a large export +works at all. > **Note:** `StreamJSONLMixin` exposes the separately named `stream_jsonl()` for endpoints > that assign `application/jsonl` their own meaning outside streaming mode, such as billing > statement charges. It sends no `MPT-Streaming` header and performs no confirmation check. > The distinct names let a service compose both streaming mixins side by side. See -> [architecture.md](architecture.md) for the distinction. +> [the streaming guide](streaming.md#do-not-confuse-stream-with-stream_jsonl) for the +> distinction. + ## Navigate The API Surface @@ -583,6 +366,7 @@ for product in ( ## Related Documents - [testing.md](testing.md): validation and test command behavior +- [streaming.md](streaming.md): streaming guide — access pattern, obligations, timeouts - [rql.md](rql.md): RQL builder guide - [architecture.md](architecture.md): repository structure and abstractions - [local-development.md](local-development.md): repository-local Docker workflow for contributors