json-stream

You can selectively change the persistence of data as you consume it. This is useful when you want to iterate through a large list transiently but need the individual items within that list to be persistent (allowing random access to their keys).

From Transient to Persistent

Call the .persistent() method on a transient list or dict-like object. This makes the children of that object persistent, not the object itself.

From Persistent to Transient

Call the .transient() method on a persistent object. This causes the object to produce transient children.

Note: When using .transient(), accessing data that was already read from the parent persistent object will still work, but accessing data in the new transient child that has already been passed in the stream will raise an exception.

import json_stream

# Iterate transient list, but produce persistent items
data = json_stream.load(f) 
for result in data['results'].persistent():
    # result is now a persistent dict-like object
    print(result['x'])
    print(result['x'])  # This works because 'result' is persistent

The library provides specialized loaders for common Python HTTP libraries to enable streaming directly from a URL.

urllib

Pass the response object directly to json_stream.load().

requests

1. You must pass stream=True to the request.
2. Use json_stream.requests.load(response). Note: This uses a default chunk_size of 10k bytes. For more responsive consumption, you can tune chunk_size to 1 to disable buffering.

httpx

1. Use client.stream(...) to initiate the request.
2. Use json_stream.httpx.load(response). Note: Similar to requests, this uses a 10k byte chunk size by default.

All these methods also support the visitor pattern via json_stream.requests.visit() and json_stream.httpx.visit().

# requests example
import requests
import json_stream.requests

with requests.get('http://example.com/data.json', stream=True) as response:
    data = json_stream.requests.load(response)

# httpx example
import httpx
import json_stream.httpx

with httpx.Client() as client, client.stream('GET', 'http://example.com/data.json') as response:
    data = json_stream.httpx.load(response)

If you are using the json-stream-rs-tokenizer (the default for speed), you can parse files containing JSON followed by binary data by using the correct_cursor=True argument.

Workflow for JSON then Binary

1. Call json_stream.load(f, correct_cursor=True) to read the JSON header.
2. Call header.read_all() to ensure the stream is read up to the end of the JSON.
3. Call header.tokenizer.park_cursor() to 'park' the underlying file cursor at the correct position.
4. Proceed to read the binary data from the file.

Workflow for Binary then JSON

Simply read the binary data from the file using standard file methods before calling json_stream.load(f).

Workflow for JSON -> Binary -> JSON

1. Read the first JSON with correct_cursor=True.
2. Call data1.read_all() and data1.tokenizer.park_cursor().
3. Read the binary data.
4. Call json_stream.load(f) for the second JSON document.

import json_stream

with open('test.bin', 'rb') as f:
    # 1. Read first JSON
    data1 = json_stream.load(f, correct_cursor=True)
    data1.read_all()
    data1.tokenizer.park_cursor()

    # 2. Read binary data
    binary_data = f.read(1024)

    # 3. Read second JSON
    data2 = json_stream.load(f)

If your data consists of a sequence of top-level JSON values (like NDJSON or concatenated JSON), use these functions:

• json_stream.load_many(f): Yields each top-level JSON value as it is parsed.
• json_stream.visit_many(f, visitor): Visits each top-level JSON value using the visitor pattern and yields control after each one.

Concatenated JSON Boundaries

json-stream can unambiguously detect boundaries for objects {...}, arrays [...], and literals true, false, null without delimiters. However, numbers and strings require a delimiter (like a space or newline) between consecutive documents to prevent tokenizer errors.

import json_stream

# Reading NDJSON
with open("events.ndjson", "rb") as f:
    for item in json_stream.load_many(f):
        handle(item)

# Visiting many documents
def visitor(value, path):
    ...

for _ in json_stream.visit_many(open("events.ndjson", "rb"), visitor):
    pass

To re-output (encode) persistent json-stream objects using the standard json.dump() or json.dumps(), you must provide a way for the encoder to handle the streaming types.

Option 1: Using default argument

Pass json_stream.encoding.default to the default parameter of json.dumps().

Option 2: Using JSONStreamEncoder class

Pass json_stream.encoding.JSONStreamEncoder to the cls parameter of json.dumps().

Option 3: Using a Context Manager

If you are using a library that calls json.dumps() internally, you can monkey-patch the encoder for the duration of a with block using JSONStreamEncoder().

Thread Safety: For multi-threaded environments, use json_stream.dump.threading.ThreadSafeJSONStreamEncoder to ensure the patch is applied correctly across threads.

import json
import json_stream
from json_stream.dump import JSONStreamEncoder, default

# Option 1
print(json.dumps(data, default=default))

# Option 2
print(json.dumps(data, cls=JSONStreamEncoder))

# Option 3 (for external libraries)
with JSONStreamEncoder():
    some_library_function_out_of_your_control(data)