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. For where the streaming mixins and exceptions sit in the codebase, see architecture.md.

Upgrading From 7.0.0

7.0.0 published the platform streaming read as stream(), taking a name that had meant the endpoint JSONL download since 6.x. Because CollectionMixin inherits the streaming mixin, a 6.x caller's stream() silently became the streaming read: same name, same import, different request and different results. This release separates the two names again.

7.0.0	now
`stream()`	`stream_snapshot()`	the platform streaming read, opted into with `MPT-Streaming`
`stream_jsonl()`	`stream()`	the endpoint JSONL download, unchanged since 6.x
`stream(stream_format=StreamFormat.JSON)`	—	the envelope encoding is held back; see The Wire Format

Upgrading from 6.x needs no change. Upgrading from 7.0.0 means renaming those two call sites.

stream_snapshot() is a transitional name: the next major version gives stream() back to the streaming read and renames the JSONL download, so the two contracts swap labels exactly once more, in a release that says so.

`stream_snapshot()` 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_snapshot()`
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	`application/jsonl`, one record per line
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_snapshot() 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_snapshot() 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_snapshot() is available on every collection service without per-service wiring:

from mpt_api_client import BearerTokenAuthentication, MPTClient, RQLQuery

client = MPTClient.from_config(
    authentication=BearerTokenAuthentication("<token>"),
    base_url="https://api.s1.show/public",
)

for order in client.commerce.orders.filter(RQLQuery(status="Processing")).stream_snapshot():
    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_snapshot() 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:

from mpt_api_client.exceptions import MPTStreamingNotSupportedError

try:
    records = list(client.commerce.orders.stream_snapshot())
except MPTStreamingNotSupportedError:
    records = list(client.commerce.orders.iterate())

Do Not Confuse `stream_snapshot()` With `stream()`

The two methods look alike and mean different things.

	`stream_snapshot()`	`stream()`
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() is its JSONL download, while stream_snapshot() is the streaming-mode read it inherits with every other collection service. Reach for stream() only when you specifically want that endpoint's JSONL contract; for everything else stream_snapshot() is the streaming read.

stream() is the name the JSONL download has carried since 6.x, and it keeps it. In the next major version stream_snapshot() takes the name back for the streaming read and the JSONL download is renamed, so the two contracts swap labels exactly once, in a release that says so.

The line grammar is shared: in both methods a record line must decode to a JSON object, and a line holding any other valid JSON value — a bare scalar, an array, null — raises json.JSONDecodeError rather than producing a bogus model.

Memory Characteristics

stream_snapshot() 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. The body is tokenized incrementally rather than buffered.

Time scales the same way. A record that spans many body chunks is assembled once, when its line ends, so the cost of a record is linear in its own length: doubling the record size doubles the work rather than quadrupling it.

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.

The bound above is asserted, not assumed: tests/e2e/streaming/ fails if the allocation peak scales with the record count, or if buffering the same export stops being markedly more expensive than streaming it. Proving that costs four live reads — a 500-record warmup, a 2,000-record export, a 20,000-record export, and that same export buffered — so 42,500 records per case and roughly 85,000 across the sync and async pair. See Streaming Memory Coverage.

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.

The Wire Format

The wire format is fixed: stream_snapshot() always requests application/jsonl, one record object per line with no envelope. The MPT-Streaming header selects streaming mode, and Accept is not a choice the caller makes.

stream_snapshot() rejects a response whose Content-Type names a different media type, before reading its body. That guard catches a proxy that ignores Accept and labels the body honestly. It does not catch a body mislabeled as application/jsonl: a {$meta, data} envelope on one line is a JSON object, so the line reader accepts it as one record, and only the count check against MPT-Item-Count will notice.

The body is parsed as it arrives, so the memory bound described above holds.

Keep-alives are blank lines and are invisible: whitespace means the four characters RFC 8259 fixes — space, tab, line feed and carriage return — so a line is a keep-alive only when there is nothing else on it. It never reaches your loop and never counts as a record.

Record boundaries are newlines alone: a record ends at a line feed, optionally preceded by a carriage return, and nothing else ends one. The last record is the one exception, and ends at the end of the body whether or not a final line feed arrives. Inside a JSON string value every character from %x20 up is legal unescaped except " and \, so U+0085, U+00A0, U+2028, U+2029, U+3000 and U+FEFF arrive whole in your text fields. Outside a value none of them is whitespace or a separator, so each of these is a malformed body rather than something the reader absorbs, and raises json.JSONDecodeError:

a line carrying only one of them, which is not a keep-alive and does count towards MPT-Item-Count
a record padded with one of them outside its value
a body that frames its records with a lone carriage return, or with U+001E — a separator of application/json-seq, a different media type from the application/jsonl this client requests

A single UTF-8 byte order mark opening the body is dropped — the same tolerance json.loads gives the paged path — so a BOM-emitting producer parses identically everywhere.

A progress receiver gets the declared MPT-Item-Count through set_total_items, exactly once, before the first record arrives, so a progress report can render a percentage of a streamed export.

Note: the {$meta, data} envelope encoding and the incremental parser that reads it are held back for the next major version, where stream_snapshot() becomes stream() again. StreamFormat remains a single-member enum so re-adding the envelope costs no change to any public name.

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: MPT-Item-Count declares K. The completeness check compares against that capped value, so a bounded export verifies exactly like a full one.

for order in client.commerce.orders.order_by("-audit.created.at").stream_snapshot(limit=100_000):
    print(order.id)

stream_snapshot() 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. It is declared with the headers, before the data, so it is a total to check the body against once the body ends, not a claim that the body arrived.

stream_snapshot() 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:

for order in client.commerce.orders.stream_snapshot():
    if order.id == "ORD-0000-0001":
        break  # deliberate early exit, no completeness check

On the async client a deliberate early exit also has to close the stream, or the response stays open until the event loop gets around to finalizing the generator. See Leaving An Async Stream Early.

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.

The supported way to capture it is a progress receiver: stream_snapshot() calls set_total_items with the declared MPT-Item-Count before the first record, so the count is in hand before any record is written. stream_snapshot() still yields records, not headers — there is no need to drop to client.http_client.stream(...) just to read the header.

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:

{"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. A stub that carries no string id breaks the one guarantee the contract makes, so stream_snapshot() raises TypeError rather than yielding a stub that identifies nothing.

stream_snapshot() 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:

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_snapshot():
        if isinstance(result, DeletionStub):
            stage_delete(attempt_id, result.id)
        else:
            stage_upsert(attempt_id, result)

    # Reached only once stream_snapshot() 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_snapshot() 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 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, 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:

for order in client.commerce.orders.stream_snapshot(skip_deleted=True):
    print(order.id)

The flag is typed with overloads, so a type checker resolves stream_snapshot(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:

from mpt_api_client.exceptions import (
    MPTStreamingIncompleteError,
    MPTStreamingTruncatedError,
)

try:
    records = list(client.commerce.orders.stream_snapshot())
except (MPTStreamingIncompleteError, MPTStreamingTruncatedError):
    records = list(client.commerce.orders.stream_snapshot())  # 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:

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("<token>"),
    )
)

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: a blank line is a keep-alive. A line of anything else fails the read instead of extending it.
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 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 `application/jsonl`
`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

import logging

from mpt_api_client.exceptions import MPTStreamingError

logger = logging.getLogger(__name__)

try:
    for order in client.commerce.orders.stream_snapshot():
        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 means the endpoint cannot serve application/jsonl for this read mode, so streaming is not available there at all.
Incomplete-export failures — MPTStreamingIncompleteError and MPTStreamingTruncatedError. A retry can succeed, but only as a fresh export; see 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. The status decides the type, so the mapping holds whatever the error body turns out to be: an endpoint that answers 406 or 413 with a bare JSON string instead of problem+json still raises the typed error, only without the members payload would otherwise carry.

A body the client cannot parse is not a streaming error at all but a json.JSONDecodeError: a malformed record line, or a line whose JSON is valid but is not an object. A record must be a JSON object, so a line holding a bare scalar, array, or null is rejected with that typed decode error rather than surfacing as a crash deeper in record handling. A body cut short usually loses records before the reader runs out of lines to parse, 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:

import logging

from mpt_api_client.exceptions import MPTStreamingOverCapError

logger = logging.getLogger(__name__)

try:
    records = list(client.commerce.orders.stream_snapshot())
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_snapshot(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:

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("<token>"),
        base_url="https://api.s1.show/public",
    )
    attempt_id = uuid.uuid4().hex

    try:
        async for result in client.commerce.orders.stream_snapshot():
            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. The Async* mixins and an awaited body are the visible difference; leaving a stream early is the one behavioural one.

Leaving An Async Stream Early

A loop that runs to the last record needs nothing extra — the generator closes the response on its way out. Leaving early is the one place the async form does not behave like the sync one, so wrap the stream in contextlib.aclosing() whenever the loop can exit before the end: a break, a return, an exception.

from contextlib import aclosing

async with aclosing(client.commerce.orders.stream_snapshot()) as orders:
    async for order in orders:
        if order.id == "ORD-0000-0001":
            break  # the response is released when the async with block exits

Without the wrapper, the break leaves the generator suspended and still holding the open response. Python does not close an abandoned async generator when its last reference goes; it hands it to the event loop's async-generator finalization hook, which runs the aclose() at a moment nothing in your code controls. Until that happens the HTTP connection stays checked out of the httpx pool.

In a short script this is invisible. In a long-lived service that breaks out of many streams, the checked-out connections accumulate against the pool limit, and the debt surfaces at loop shutdown as unclosed-response and unfinalized-async-generator warnings — which point at the shutdown rather than at the loop that caused them.

The sync form needs no wrapper on CPython: dropping the last reference to a suspended generator closes it there and then, so break-ing out of a for releases the response promptly. contextlib.closing() states the same thing explicitly if you would rather not rely on refcounting.

Two things the wrapper does not change:

The completeness check is still skipped. Closing the iterator releases the connection; it does not verify the export. An early exit is not reported as an incomplete stream, as obligation 1 describes.
progress still sees no completed(). The export was abandoned rather than finished, so the receiver ends on the last item_processed it was given.

The JSONL stream() is an async generator over an open response too, so an early exit from one takes the same wrapper.

Reporting Progress

A long export gives no feedback by default. stream_snapshot() accepts an optional progress receiver, called once per consumed record — including a stub withheld by skip_deleted — and once on completion:

from mpt_api_client.models import ConsoleProgress

for order in client.commerce.orders.stream_snapshot(progress=ConsoleProgress()):
    print(order.id)

set_total_items is called exactly once, with the declared MPT-Item-Count, when the response headers are verified — before the first record — so ConsoleProgress renders a real percentage from the start of the export.

The receiver is also where you capture the declared count when the payload is stored for later verification, as obligation 1 describes. Implement the Progress protocol (or AsyncProgress for the async client) to route progress somewhere other than the console.

Navigation Menu

Repository navigation

streaming.md

streaming.md

Streaming

Upgrading From 7.0.0

`stream_snapshot()` Versus `iterate()`

Every Collection Service Streams

Do Not Confuse `stream_snapshot()` With `stream()`

Memory Characteristics

What A Stream Is

The Wire Format

Bounding An Export

Three Obligations You Cannot Skip

1. Verify Completeness Against `MPT-Item-Count`

2. Check For A Deletion Stub Before Ingesting A Record

Opting Out Of Deletion Stubs

3. Restart, Do Not Resume

The Client-Timeout Trap

Streaming Errors

Over-Cap Exports

Async Streaming

Leaving An Async Stream Early

Reporting Progress

Related Documents

FilesExpand file tree

streaming.md

Latest commit

History

streaming.md

File metadata and controls

Streaming

Upgrading From 7.0.0

stream_snapshot() Versus iterate()

Every Collection Service Streams

Do Not Confuse stream_snapshot() With stream()

Memory Characteristics

What A Stream Is

The Wire Format

Bounding An Export

Three Obligations You Cannot Skip

1. Verify Completeness Against MPT-Item-Count

2. Check For A Deletion Stub Before Ingesting A Record

Opting Out Of Deletion Stubs

3. Restart, Do Not Resume

The Client-Timeout Trap

Streaming Errors

Over-Cap Exports

Async Streaming

Leaving An Async Stream Early

Reporting Progress

Related Documents

`stream_snapshot()` Versus `iterate()`

Do Not Confuse `stream_snapshot()` With `stream()`

1. Verify Completeness Against `MPT-Item-Count`