### Visitor Pattern Output Example

Source: https://github.com/daggaz/json-stream/blob/master/README.md

Example output demonstrating the visitor pattern, showing the parsed items and their corresponding paths within the JSON structure.

```text
1 at path ('x',)
{} at path ('y',)
1 at path ('xxxx', 0)
2 at path ('xxxx', 1)
1 at path ('xxxx', 2, 'yyyy')
z at path ('xxxx', 3)
1 at path ('xxxx', 4)
[] at path ('xxxx', 5)
```

--------------------------------

### Collect Specific Data with Visitor Callback

Source: https://context7.com/daggaz/json-stream/llms.txt

Example of using json_stream.visit() to collect specific data points (leaf scalars at depth >= 2) by defining a custom visitor function that filters and stores relevant values.

```python
# --- Collect only leaf scalars at depth >= 2 ---
results = {}
def collect_deep(value, path):
    if len(path) >= 2 and not isinstance(value, (dict, list)):
        results[path] = value

json_stream.visit(StringIO(json_data), collect_deep)
print(results)
# {('data', 0): 10, ('data', 1): 20, ('data', 2, 'z'): 30}
```

--------------------------------

### Generate Nested JSON with streamable_list and streamable_dict

Source: https://github.com/daggaz/json-stream/blob/master/README.md

This example demonstrates generating a JSON object with a nested JSON list. It uses time.sleep() to simulate slow data generation and show that output is written incrementally. Both streamable_list and streamable_dict are used.

```python
import sys
import json
import time

from json_stream.writer import streamable_dict, streamable_list


# define a list data generator that (slowly) yields 
# items that will be written as a JSON list
@streamable_list
def generate_list(n):
    # output n numbers and their squares
    for i in range(n):  # range is itself a generator
        yield i
        time.sleep(1)


# define a dictionary data generator that (slowly) yields 
# key/value pairs that will be written as a JSON dict
@streamable_dict
def generate_dict(n):
    # output n numbers and their squares
    for i in range(n):  # range is itself a generator
        yield i, i ** 2
        time.sleep(1)

    # yield another dictionary item key, with the value
    # being a streamed nested list
    yield "a list", generate_list(n)


# get a streamable generator
data = generate_dict(5)

# use json.dump() to write dict generator to stdout
json.dump(data, sys.stdout, indent=2)

# if you already have an iterable object, you can just
# call streamable_* on it to make it writable
data = streamable_list(range(10))
json.dump(data, sys.stdout)
```

--------------------------------

### Encode json-stream Objects using json.dumps

Source: https://github.com/daggaz/json-stream/blob/master/README.md

Re-output persistent json-stream objects back to JSON using `json.dumps`. This example shows using `json_stream.encoding.default` as the `default` argument.

```python
import json

import json_stream
from json_stream.dump import JSONStreamEncoder, default

data = json_stream.load(f, persistent=True)

# Option 1: supply json_stream.encoding.default as the default argument
print(json.dumps(data, default=default))
```

--------------------------------

### Encode json-stream Objects using JSONStreamEncoder

Source: https://github.com/daggaz/json-stream/blob/master/README.md

Re-output persistent json-stream objects back to JSON using `json.dumps`. This example shows using `json_stream.encoding.JSONStreamEncoder` as the `cls` argument.

```python
import json

import json_stream
from json_stream.dump import JSONStreamEncoder, default

data = json_stream.load(f, persistent=True)

# Option 2: supply json_stream.encoding.JSONStreamEncoder as the cls argument
print(json.dumps(data, cls=JSONStreamEncoder))
```

--------------------------------

### Mixed JSON + binary data with `correct_cursor=True` and `park_cursor()`

Source: https://context7.com/daggaz/json-stream/llms.txt

Explains how to handle files containing both JSON and binary data by using `correct_cursor=True` and `park_cursor()` to manage the file cursor position accurately.

```APIDOC
# Mixed JSON + binary data with `correct_cursor=True` and `park_cursor()`

When a file contains JSON followed by binary data, pass `correct_cursor=True` so the Rust tokenizer tracks the exact byte position, then call `park_cursor()` to reposition the file cursor at the end of the JSON before reading the binary section. Requires `json-stream-rs-tokenizer`.

```python
import io
import json
import json_stream

# Build a binary file: JSON header + raw binary payload
json_header = json.dumps({"version": 1, "length": 4})
binary_payload = b'\x00\x01\x02\x03'
file_data = json_header.encode() + binary_payload

with io.BytesIO(file_data) as f:
    # Step 1: read JSON header with cursor tracking
    header = json_stream.load(f, correct_cursor=True)
    version = header["version"]         # 1
    length  = header["length"]          # 4
    header.read_all()                   # drain any remaining JSON tokens

    # Step 2: reposition file cursor to end of JSON
    header.tokenizer.park_cursor()
    print(f"Cursor after JSON: {f.tell()}")   # == len(json_header)

    # Step 3: read binary payload normally
    binary = f.read(length)
    print(binary)  # b'\x00\x01\x02\x03'
```
```

--------------------------------

### HTTP integrations: requests and httpx

Source: https://context7.com/daggaz/json-stream/llms.txt

Shows how to use json-stream with the `requests` and `httpx` libraries to process streaming JSON responses from HTTP requests.

```APIDOC
# HTTP integrations: `json_stream.requests` and `json_stream.httpx`

Drop-in wrappers for the `requests` and `httpx` libraries that convert HTTP response streams into json-stream-compatible iterables, exposing the same `load`, `load_many`, `visit`, and `visit_many` API.

```python
# --- requests integration ---
import requests
import json_stream.requests

with requests.get("https://api.example.com/large-dataset.json", stream=True) as response:
    data = json_stream.requests.load(response, persistent=False)
    for item in data["results"]:
        process(item)   # processes each result as it arrives

# With custom chunk size (default is 10 KB; set to 1 to disable buffering)
with requests.get("https://api.example.com/events.ndjson", stream=True) as response:
    for doc in json_stream.requests.load_many(response, chunk_size=1):
        handle_event(doc)

# Visitor over HTTP response
with requests.get("https://api.example.com/data.json", stream=True) as response:
    def visitor(value, path):
        if path == ("user", "id"):
            print(f"User ID: {value}")
    json_stream.requests.visit(response, visitor)

# --- httpx integration ---
import httpx
import json_stream.httpx

with httpx.Client() as client:
    with client.stream("GET", "https://api.example.com/large-dataset.json") as response:
        data = json_stream.httpx.load(response)
        for item in data["results"]:
            process(item)
```
```

--------------------------------

### Visit JSON from URL using httpx

Source: https://github.com/daggaz/json-stream/blob/master/README.md

Apply a visitor function to JSON data streamed from a URL using the `httpx` library with `json_stream.httpx.visit`.

```python
import httpx
import json_stream.httpx

def visitor(item, path):
    print(f"{item} at path {path}")
    
with httpx.Client() as client, client.stream('GET', 'http://example.com/data.json') as response:
    json_stream.httpx.visit(response, visitor)
```

--------------------------------

### Httpx Integration for Streaming JSON Responses

Source: https://context7.com/daggaz/json-stream/llms.txt

Demonstrates using `json_stream.httpx.load` to process large JSON responses from the `httpx` library as streams. This allows for efficient processing of large datasets without loading the entire response into memory.

```python
# --- httpx integration ---
import httpx
import json_stream.httpx

with httpx.Client() as client:
    with client.stream("GET", "https://api.example.com/large-dataset.json") as response:
        data = json_stream.httpx.load(response)
        for item in data["results"]:
            process(item)
```

--------------------------------

### Using a Custom JSON Tokenizer

Source: https://github.com/daggaz/json-stream/blob/master/README.md

Provide a custom JSON tokenizer implementation by passing it as the 'tokenizer' argument to the json_stream.load() or json_stream.visit() methods.

```python
json_stream.load(f, tokenizer=some_tokenizer, **tokenizer_kwargs)
```

--------------------------------

### Handling Mixed JSON and Binary Data

Source: https://context7.com/daggaz/json-stream/llms.txt

Explains how to parse files containing JSON followed by binary data using `correct_cursor=True` and `park_cursor()`. This ensures accurate file cursor positioning after reading JSON, enabling subsequent reading of binary data.

```python
import io
import json
import json_stream

# Build a binary file: JSON header + raw binary payload
json_header = json.dumps({"version": 1, "length": 4})
binary_payload = b'\x00\x01\x02\x03'
file_data = json_header.encode() + binary_payload

with io.BytesIO(file_data) as f:
    # Step 1: read JSON header with cursor tracking
    header = json_stream.load(f, correct_cursor=True)
    version = header["version"]         # 1
    length  = header["length"]          # 4
    header.read_all()                   # drain any remaining JSON tokens

    # Step 2: reposition file cursor to end of JSON
    header.tokenizer.park_cursor()
    print(f"Cursor after JSON: {f.tell()}")   # == len(json_header)

    # Step 3: read binary payload normally
    binary = f.read(length)
    print(binary)  # b'\x00\x01\x02\x03'
```

--------------------------------

### Visit JSON from URL using requests

Source: https://github.com/daggaz/json-stream/blob/master/README.md

Apply a visitor function to JSON data streamed from a URL using the `requests` library with `json_stream.requests.visit`. The `chunk_size` note applies here as well.

```python
import requests
import json_stream.requests

def visitor(item, path):
    print(f"{item} at path {path}")
    
with requests.get('http://example.com/data.json', stream=True) as response:
    json_stream.requests.visit(response, visitor)
```

--------------------------------

### Requests Integration for Streaming JSON Responses

Source: https://context7.com/daggaz/json-stream/llms.txt

Shows how to use `json_stream.requests.load` and `json_stream.requests.load_many` to process large JSON responses from the `requests` library as streams. It also demonstrates using `json_stream.requests.visit` for targeted data extraction.

```python
# --- requests integration ---
import requests
import json_stream.requests

with requests.get("https://api.example.com/large-dataset.json", stream=True) as response:
    data = json_stream.requests.load(response, persistent=False)
    for item in data["results"]:
        process(item)   # processes each result as it arrives
```

```python
# With custom chunk size (default is 10 KB; set to 1 to disable buffering)
with requests.get("https://api.example.com/events.ndjson", stream=True) as response:
    for doc in json_stream.requests.load_many(response, chunk_size=1):
        handle_event(doc)
```

```python
# Visitor over HTTP response
with requests.get("https://api.example.com/data.json", stream=True) as response:
    def visitor(value, path):
        if path == ("user", "id"):
            print(f"User ID: {value}")
    json_stream.requests.visit(response, visitor)
```

--------------------------------

### Visit JSON from URL using urllib

Source: https://github.com/daggaz/json-stream/blob/master/README.md

Apply a visitor function to JSON data streamed directly from a URL using `urllib.request` and `json_stream.visit`.

```python
import urllib.request
import json_stream

def visitor(item, path):
    print(f"{item} at path {path}")
    
with urllib.request.urlopen('http://example.com/data.json') as response:
    json_stream.visit(response, visitor)
```

--------------------------------

### Load JSON with Persistent Mode and Convert to Transient

Source: https://github.com/daggaz/json-stream/blob/master/README.md

Load JSON data in persistent mode and then convert it to a transient object to manage memory. Accessing data that has already been passed in the transient list will raise an exception.

```python
# JSON: {"a": 1, "x": ["long", "list", "I", "don't", "want", "in", "memory"], "b": 2}
data = load(StringIO(json), persistent=True).transient()
# data is a persistent dict-list object that produces transient children

print(data["a"])  # prints 1
x = data["x"]  # x is a transient list, you can use it accordingly
print(x[0])  # prints long

# access earlier data from memory
print(data["a"])  # this would have raised an exception if data was transient

print(data["b"])  # prints 2

# we have now moved past all the data in the transient list
print(x[0])  # will raise exception
```

--------------------------------

### ThreadSafeJSONStreamEncoder

Source: https://context7.com/daggaz/json-stream/llms.txt

Demonstrates the use of ThreadSafeJSONStreamEncoder for safely encoding JSON streams in multi-threaded environments using a context manager.

```APIDOC
## ThreadSafeJSONStreamEncoder

Thread-safe variant of `JSONStreamEncoder` that uses locks to ensure concurrent uses of the context manager only apply and remove the monkey-patch once (first-in / last-out), preventing races in multi-threaded servers.

```python
import json
import json_stream
from json_stream.dump.threading import ThreadSafeJSONStreamEncoder
from io import StringIO
from concurrent.futures import ThreadPoolExecutor

json_data = '{"count": 1, "items": [42]}'

def serialize_in_thread(_):
    data = json_stream.load(StringIO(json_data))
    with ThreadSafeJSONStreamEncoder():
        return json.dumps(data)

with ThreadPoolExecutor(max_workers=4) as pool:
    results = list(pool.map(serialize_in_thread, range(4)))

for r in results:
    print(r)
# {"count": 1, "items": [42]}  (printed 4 times, safely)
```
```

--------------------------------

### Load JSON Document (Transient Mode)

Source: https://context7.com/daggaz/json-stream/llms.txt

Use json_stream.load() in transient mode for memory-efficient, forward-only parsing of a single JSON document. Accessing already-read keys will raise a TransientAccessException.

```python
import json_stream
from io import StringIO
from json_stream.base import TransientAccessException

# --- Transient mode (default): memory-efficient, forward-only ---
json_data = '{"count": 3, "results": ["a", "b", "c"]}'

data = json_stream.load(StringIO(json_data))  # TransientStreamingJSONObject
results = data["results"]                     # TransientStreamingJSONList

for item in results:
    print(item)  # prints: a, b, c

try:
    count = data["count"]  # raises: key was already passed in the stream
except TransientAccessException as e:
    print(f"Transient error: {e}")
```

--------------------------------

### Reading Mixed JSON and Binary Data with Rust Tokenizer

Source: https://github.com/daggaz/json-stream/blob/master/README.md

When using the Rust tokenizer, enable correct_cursor=True to parse mixed data. Ensure all JSON data is read using read_all() and park the cursor before reading binary data.

```python
import json_stream

with open('test.bin', 'rb') as f:
    # read JSON header
    header = json_stream.load(f, correct_cursor=True)
    # ... process JSON header ...
    header.read_all()

    # ensure the tokenizer has "parked" the file 
    # cursor at the end of the JSON data
    header.tokenizer.park_cursor()

    # now we can read binary data from the same file
    binary_start = f.tell()
    data = f.read()
```

--------------------------------

### Stream JSON from URL using httpx

Source: https://github.com/daggaz/json-stream/blob/master/README.md

Stream JSON data from a URL using the `httpx` library. Use `client.stream()` and `json_stream.httpx.load()` with the response object. Similar `chunk_size` considerations apply as with `requests`.

```python
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)
```

--------------------------------

### Visit multiple JSON documents with json_stream.visit_many()

Source: https://context7.com/daggaz/json-stream/llms.txt

Applies a visitor function to each top-level JSON document within a single stream. The generator yields after each document, allowing per-document logic. Ensure the visitor function is defined before use.

```python
import json_stream
from io import StringIO

payload = "\n".join([
    '{"a": 1, "b": 2}',
    '[10, 20, 30]',
    '"standalone string"',
])

batch = []

def visitor(value, path):
    batch.append((path, value))

for _ in json_stream.visit_many(StringIO(payload), visitor):
    print(f"Document done, visited {len(batch)} leaf(s): {batch}")
    batch.clear()

# Document done, visited 2 leaf(s): [(('a',), 1), (('b',), 2)]
# Document done, visited 3 leaf(s): [((0,), 10), ((1,), 20), ((2,), 30)]
# Document done, visited 1 leaf(s): [((), 'standalone string')]
```

--------------------------------

### Load JSON Document (Persistent Mode)

Source: https://context7.com/daggaz/json-stream/llms.txt

Use json_stream.load() with persistent=True for random access to JSON data. Previously read data is cached in memory. Note that len() may trigger reading the entire document.

```python
# --- Persistent mode: random access, stores previously-read data ---
data = json_stream.load(StringIO(json_data), persistent=True)

# Access in any order; already-parsed keys are served from memory
print(data["results"][1])  # b
print(data["count"])       # 3  (retrieved from memory cache)
print(len(data))           # 2  (triggers read_all internally)
```

--------------------------------

### Reading Binary Data Before JSON

Source: https://github.com/daggaz/json-stream/blob/master/README.md

To read binary data followed by JSON, read the binary data directly from the file before calling json_stream.load().

```python
with open('test.bin', 'rb') as f:
    binary_data = f.read(1024)
    data = json_stream.load(f)
    # ... process JSON ...
```

--------------------------------

### streamable_list() and streamable_dict()

Source: https://context7.com/daggaz/json-stream/llms.txt

Wrap Python generators (or any iterable) so they can be passed to `json.dump()` for iterative JSON encoding without pre-materializing the entire document.

```APIDOC
## `streamable_list()` and `streamable_dict()`

Wrap Python generators (or any iterable) so they can be passed to `json.dump()` for iterative JSON encoding without pre-materializing the entire document.

```python
import json
import sys
from json_stream import streamable_list, streamable_dict

# --- As decorator on a generator function ---
@streamable_list
def generate_numbers(n):
    for i in range(n):
        yield i * i  # could be an expensive computation

@streamable_dict
def generate_report(title, n):
    yield "title", title
    yield "squares", generate_numbers(n)
    yield "total", n

report = generate_report("squares report", 5)
json.dump(report, sys.stdout, indent=2)
# {
#   "title": "squares report",
#   "squares": [0, 1, 4, 9, 16],
#   "total": 5
# }

# --- As a wrapper around an existing iterable ---
data = streamable_list(range(4))
print(json.dumps(data))      # [0, 1, 2, 3]

pairs = streamable_dict([("a", 1), ("b", 2)])
print(json.dumps(pairs))     # {"a": 1, "b": 2}
```
```

--------------------------------

### Mixed mode: `.persistent()` and `.transient()`

Source: https://context7.com/daggaz/json-stream/llms.txt

Transient streaming objects can be switched to produce persistent children (and vice versa) mid-stream, giving fine-grained control over which parts of the document are cached in memory.

```APIDOC
## Mixed mode: `.persistent()` and `.transient()`

Transient streaming objects can be switched to produce persistent children (and vice versa) mid-stream, giving fine-grained control over which parts of the document are cached in memory.

```python
import json_stream
from io import StringIO

# --- Transient list that yields persistent objects ---
# Useful when list items have unordered keys
json_data = '{"results": [{"y": 3, "x": 1}, {"x": 2, "y": 4}]}'

data = json_stream.load(StringIO(json_data))        # transient dict
results = data["results"]                           # transient list

for item in results.persistent():                   # items are persistent
    print(item["x"], item["y"])                     # access in any order
# 1 3
# 2 4

# --- Persistent root with transient children (saves memory for large lists) ---
json_data2 = '{"a": 1, "items": ["big", "list", "here"], "b": 2}'
root = json_stream.load(StringIO(json_data2), persistent=True).transient()

print(root["a"])                 # 1 (cached)
items = root["items"]           # TransientStreamingJSONList
print(root["b"])                 # 2 (retrieved from cache, not re-read)

try:
    _ = items[0]                 # raises: already passed in transient stream
except Exception as e:
    print(f"Expected: {e}")
```
```

--------------------------------

### Visit Many Values in Many Documents

Source: https://github.com/daggaz/json-stream/blob/master/README.md

Use `visit_many` to apply a visitor function to each value within each top-level JSON text in a file.

```python
import json_stream

def visitor(value, path):
    # called depth-first for each value within the current top-level JSON text
    ...

for _ in json_stream.visit_many(open("events.ndjson", "rb"), visitor):
    # loop advances one top-level JSON text at a time
    pass
```

--------------------------------

### Visitor Pattern for JSON Parsing

Source: https://github.com/daggaz/json-stream/blob/master/README.md

Parse JSON using a visitor pattern where a supplied function is called for each data item as it is parsed depth-first. This uses a transient parser, conserving memory.

```python
import json_stream

# JSON: {"x": 1, "y": {}, "xxxx": [1,2, {"yyyy": 1}, "z", 1, []]}

def visitor(item, path):
    print(f"{item} at path {path}")

json_stream.visit(f, visitor)
```

--------------------------------

### Visit JSON Document with Callback

Source: https://context7.com/daggaz/json-stream/llms.txt

Use json_stream.visit() to parse a JSON document and invoke a callback function for each scalar value, empty object, or empty list encountered during a depth-first traversal.

```python
import json_stream
from io import StringIO

json_data = '{"x": 1, "y": {}, "data": [10, 20, {"z": 30}, []]}'

def visitor(value, path):
    print(f"{path!r:40s} -> {value!r}")

json_stream.visit(StringIO(json_data), visitor)
# ('x',)                                   -> 1
# ('y',)                                   -> {}
# ('data', 0)                              -> 10
# ('data', 1)                              -> 20
# ('data', 2, 'z')                         -> 30
# ('data', 3)                              -> []
```

--------------------------------

### json_stream.visit()

Source: https://context7.com/daggaz/json-stream/llms.txt

Parses a single JSON document using a visitor callback invoked depth-first for every scalar value, empty object, or empty list encountered.

```APIDOC
## json_stream.visit()

### Description
Parses a single JSON document using a visitor callback that is invoked depth-first for every scalar value, empty object, or empty list. The callback receives `(value, path)` where `path` is a tuple of string keys and integer indices.

### Parameters
- `fp` (file-like object or iterable): The source of the JSON data.
- `visitor` (callable): A function to be called for each visited element. It accepts `(value, path)` as arguments.

### Request Example
```python
import json_stream
from io import StringIO

json_data = '{"x": 1, "y": {}, "data": [10, 20, {"z": 30}, []]}'

def visitor(value, path):
    print(f"{path!r:40s} -> {value!r}")

json_stream.visit(StringIO(json_data), visitor)
```

### Response
- This function does not return a value directly but executes the `visitor` callback for each element in the JSON structure.
```

--------------------------------

### Handling Multiple JSON Documents with Interspersed Binary Data

Source: https://github.com/daggaz/json-stream/blob/master/README.md

For streams with binary data between JSON documents, use correct_cursor=True for JSON documents followed by binary data. Park the cursor after reading each JSON document before reading binary data.

```python
with open('test.bin', 'rb') as f:
    # 1. Read first JSON
    data1 = json_stream.load(f, correct_cursor=True)
    # ... process data1 ...
    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)
    # ... process data2 ...
```

--------------------------------

### JSONStreamEncoder and default

Source: https://context7.com/daggaz/json-stream/llms.txt

Allows re-encoding streaming JSON objects back to JSON strings using the standard `json.dumps()` / `json.dump()` interface via three usage patterns: `default=` argument, `cls=` argument, or context manager monkey-patch.

```APIDOC
## `JSONStreamEncoder` and `default`

Allows re-encoding streaming JSON objects back to JSON strings using the standard `json.dumps()` / `json.dump()` interface via three usage patterns: `default=` argument, `cls=` argument, or context manager monkey-patch.

```python
import json
import json_stream
from json_stream.dump import JSONStreamEncoder, default
from io import StringIO

json_data = '{"count": 3, "results": ["a", "b", "c"]}'

# Option 1: default= argument
data = json_stream.load(StringIO(json_data))
print(json.dumps(data, default=default))
```
```

--------------------------------

### Switch between persistent and transient objects mid-stream

Source: https://context7.com/daggaz/json-stream/llms.txt

Allows fine-grained control over memory caching by switching between persistent (cached) and transient (on-demand) streaming objects. Use `.persistent()` on lists when items have unordered keys, and `.transient()` on roots to save memory for large lists while keeping other parts cached.

```python
import json_stream
from io import StringIO

# --- Transient list that yields persistent objects ---
# Useful when list items have unordered keys
json_data = '{"results": [{"y": 3, "x": 1}, {"x": 2, "y": 4}]}'

data = json_stream.load(StringIO(json_data))        # transient dict
results = data["results"]                           # transient list

for item in results.persistent():                   # items are persistent
    print(item["x"], item["y"])                     # access in any order
# 1 3
# 2 4

# --- Persistent root with transient children (saves memory for large lists) ---
json_data2 = '{"a": 1, "items": ["big", "list", "here"], "b": 2}'
root = json_stream.load(StringIO(json_data2), persistent=True).transient()

print(root["a"])                 # 1 (cached)
items = root["items"]           # TransientStreamingJSONList
print(root["b"])                 # 2 (retrieved from cache, not re-read)

try:
    _ = items[0]                 # raises: already passed in transient stream
except Exception as e:
    print(f"Expected: {e}")
```

--------------------------------

### Thread-Safe JSON Encoding with ThreadSafeJSONStreamEncoder

Source: https://context7.com/daggaz/json-stream/llms.txt

Illustrates the use of `ThreadSafeJSONStreamEncoder` for thread-safe JSON serialization of json-stream objects within a multi-threaded environment. It ensures that concurrent context manager usage is handled safely.

```python
import json
import json_stream
from json_stream.dump.threading import ThreadSafeJSONStreamEncoder
from io import StringIO
from concurrent.futures import ThreadPoolExecutor

json_data = '{"count": 1, "items": [42]}'

def serialize_in_thread(_):
    data = json_stream.load(StringIO(json_data))
    with ThreadSafeJSONStreamEncoder():
        return json.dumps(data)

with ThreadPoolExecutor(max_workers=4) as pool:
    results = list(pool.map(serialize_in_thread, range(4)))

for r in results:
    print(r)
```

--------------------------------

### Create streamable iterables with streamable_list() and streamable_dict()

Source: https://context7.com/daggaz/json-stream/llms.txt

Wrap Python generators or iterables to be compatible with `json.dump()` for iterative JSON encoding without pre-materializing the entire document. This is useful for large datasets or computationally intensive data generation.

```python
import json
import sys
from json_stream import streamable_list, streamable_dict

# --- As decorator on a generator function ---
@streamable_list
def generate_numbers(n):
    for i in range(n):
        yield i * i  # could be an expensive computation

@streamable_dict
def generate_report(title, n):
    yield "title", title
    yield "squares", generate_numbers(n)
    yield "total", n

report = generate_report("squares report", 5)
json.dump(report, sys.stdout, indent=2)
# {
#   "title": "squares report",
#   "squares": [0, 1, 4, 9, 16],
#   "total": 5
# }

# --- As a wrapper around an existing iterable ---
data = streamable_list(range(4))
print(json.dumps(data))      # [0, 1, 2, 3]

pairs = streamable_dict([("a", 1), ("b", 2)])
print(json.dumps(pairs))     # {"a": 1, "b": 2}
```

--------------------------------

### Stream JSON from URL using requests

Source: https://github.com/daggaz/json-stream/blob/master/README.md

Stream JSON data from a URL using the `requests` library. Ensure `stream=True` is set and use `json_stream.requests.load()` with the response object. Note the `chunk_size` caveat for responsiveness.

```python
import requests
import json_stream.requests

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

--------------------------------

### Iterate Transient List, Produce Persistent Items

Source: https://github.com/daggaz/json-stream/blob/master/README.md

Use the `.persistent()` method on a transient list to produce persistent items, allowing random access to child objects without loading the entire list into memory.

```python
import json_stream

# JSON: {"results": [{"x": 1, "y": 3}, {"y": 4, "x": 2}]}
# note that the keys of the inner objects are not ordered 
data = json_stream.load(f)  # data is a transient dict-like object 

# iterate transient list, but produce persistent items
for result in data['results'].persistent():
    # result is a persistent dict-like object
    print(result['x'])  # print x
    print(result['y'])  # print y (error on second result without .persistent())
    print(result['x'])  # print x again (error without .persistent())
```

--------------------------------

### Partially Consume Documents with load_many

Source: https://context7.com/daggaz/json-stream/llms.txt

Demonstrates partially consuming JSON documents yielded by json_stream.load_many(). The generator allows for selective processing of parts of each document.

```python
# --- Partially consuming documents ---
stream = StringIO('{"results": [1, 2, 3]} {"meta": {"total": 3}} 42')
gen = json_stream.load_many(stream, persistent=True)

first = next(gen)
print(list(first["results"]))          # [1, 2, 3]

second = next(gen)
print(dict(second["meta"].items()))    # {"total": 3}

third = next(gen)
print(third)                           # 42
```

--------------------------------

### Thread-Safe JSON Encoding Context Manager

Source: https://github.com/daggaz/json-stream/blob/master/README.md

Utilize ThreadSafeJSONStreamEncoder for thread-safe JSON encoding. It ensures the patch is applied only when the first thread enters and removed when the last thread exits.

```python
from json_stream.dump.threading import ThreadSafeJSONStreamEncoder

# your code
with ThreadSafeJSONStreamEncoder():
   some_library_function_out_of_your_control(data)
```

--------------------------------

### json_stream.visit_many()

Source: https://context7.com/daggaz/json-stream/llms.txt

Applies the visitor pattern across multiple top-level JSON documents in a single stream. The generator yields once per document, allowing interleaved per-document logic after each batch of visits.

```APIDOC
## `json_stream.visit_many()`

Applies the visitor pattern across multiple top-level JSON documents in a single stream. The generator yields once per document, allowing interleaved per-document logic after each batch of visits.

```python
import json_stream
from io import StringIO

payload = "\n".join([
    '{"a": 1, "b": 2}',
    '[10, 20, 30]',
    '"standalone string"',
])

batch = []

def visitor(value, path):
    batch.append((path, value))

for _ in json_stream.visit_many(StringIO(payload), visitor):
    print(f"Document done, visited {len(batch)} leaf(s): {batch}")
    batch.clear()

# Document done, visited 2 leaf(s): [(('a',), 1), (('b',), 2)]
# Document done, visited 3 leaf(s): [((0,), 10), ((1,), 20), ((2,), 30)]
# Document done, visited 1 leaf(s): [((), 'standalone string')]
```
```

--------------------------------

### Re-encode streaming JSON with JSONStreamEncoder and default

Source: https://context7.com/daggaz/json-stream/llms.txt

Allows re-encoding streaming JSON objects back to JSON strings using standard `json.dumps()` or `json.dump()`. This can be achieved via the `default=` argument, the `cls=` argument with `JSONStreamEncoder`, or by using a context manager to monkey-patch the default encoder.

```python
import json
import json_stream
from json_stream.dump import JSONStreamEncoder, default
from io import StringIO

json_data = '{"count": 3, "results": ["a", "b", "c"]}'

# Option 1: default= argument
data = json_stream.load(StringIO(json_data))
print(json.dumps(data, default=default))
```

--------------------------------

### read_all() and is_streaming / .streaming

Source: https://context7.com/daggaz/json-stream/llms.txt

Explicitly drain all remaining data from a streaming object, and check whether it still has unread content.

```APIDOC
## `read_all()` and `is_streaming` / `.streaming`

Explicitly drain all remaining data from a streaming object, and check whether it still has unread content.

```python
import json_stream
from io import StringIO

data = json_stream.load(StringIO('{"a": 1, "b": [1, 2, 3]}'), persistent=True)

print(data.streaming)    # True – not yet fully consumed
data.read_all()
print(data.streaming)    # False – fully consumed

# read_all() is also called automatically by load_many() between documents
```
```

--------------------------------

### Stream JSON from URL using urllib

Source: https://github.com/daggaz/json-stream/blob/master/README.md

Load JSON data directly from a URL using Python's built-in `urllib.request`. The response object is file-like and can be passed directly to `json_stream.load`.

```python
import urllib.request
import json_stream

with urllib.request.urlopen('http://example.com/data.json') as response:
    data = json_stream.load(response)
```

--------------------------------

### Wrap Iterable with streamable_list

Source: https://github.com/daggaz/json-stream/blob/master/README.md

Use streamable_list to wrap an existing iterable, making it compatible with json.dump() for streaming output.

```python
import sys
import json

from json_stream import streamable_list

# wrap existing iterable
data = streamable_list(range(10))

# consume iterable with standard json.dump()
json.dump(data, sys.stdout)
```

--------------------------------

### json_stream.load()

Source: https://context7.com/daggaz/json-stream/llms.txt

Parses a single JSON document from a file-like object or iterable, returning a streaming dict-like or list-like object. Supports transient (memory-efficient) and persistent (random access) modes.

```APIDOC
## json_stream.load()

### Description
Parses a single JSON document from a file-like object or iterable and returns a streaming dict-like or list-like object. The `persistent` flag (default `False`) controls whether already-read data is kept in memory for random access or discarded after consumption.

### Parameters
- `fp` (file-like object or iterable): The source of the JSON data.
- `persistent` (bool, optional): If `True`, previously-read data is cached in memory for random access. Defaults to `False` (transient mode).

### Request Example
```python
import json_stream
from io import StringIO

json_data = '{"count": 3, "results": ["a", "b", "c"]}'

# Transient mode (default)
data_transient = json_stream.load(StringIO(json_data))
results_transient = data_transient["results"]
for item in results_transient:
    print(item)

# Persistent mode
data_persistent = json_stream.load(StringIO(json_data), persistent=True)
print(data_persistent["results"][1])
print(data_persistent["count"])
```

### Response
- Returns a streaming JSON object (dict-like or list-like).

### Response Example
```json
{
  "count": 3,
  "results": [
    "a",
    "b",
    "c"
  ]
}
```
```

--------------------------------

### Stream JSON from an Iterable

Source: https://github.com/daggaz/json-stream/blob/master/README.md

Load JSON data from any iterable that yields byte or string chunks of encoded JSON. This is how `requests` and `httpx` extensions function internally.

```python
import json_stream

def some_iterator():
    yield b'{"some":'
    yield b' "JSON"}'

data = json_stream.load(some_iterator())
assert data['some'] == "JSON"
```

--------------------------------

### Read NDJSON from File

Source: https://github.com/daggaz/json-stream/blob/master/README.md

Use `load_many` to iterate over JSON objects in a file where each object is on a new line (NDJSON format).

```python
import json_stream

with open("events.ndjson", "rb") as f:  # NDJSON: one JSON object per line
    for item in json_stream.load_many(f):
        # each item is the top-level item in each JSON text
        handle(item)
```

--------------------------------

### Load JSON from Bytes Iterator

Source: https://context7.com/daggaz/json-stream/llms.txt

Load JSON data from an iterator yielding bytes, suitable for chunked network data. The library handles assembling the chunks into a complete JSON document.

```python
# --- Load from bytes iterator (e.g. chunked network data) ---
def chunked():
    yield b'{"key": '
    yield b'"value"}'

data = json_stream.load(chunked())
print(data["key"])  # value
```

--------------------------------

### Load Multiple JSON Documents (NDJSON)

Source: https://context7.com/daggaz/json-stream/llms.txt

Use json_stream.load_many() to parse a stream containing multiple JSON documents, such as NDJSON (one JSON value per line). Each document is yielded in order.

```python
# --- NDJSON (one JSON value per line) ---
ndjson = "\n".join([
    '{"event": "click", "id": 1}',
    '{"event": "scroll", "id": 2}',
    '[1, 2, 3]',
    'true',
    '"hello"',
])

for doc in json_stream.load_many(StringIO(ndjson), persistent=True):
    print(to_standard_types(doc))
# {"event": "click", "id": 1}
# {"event": "scroll", "id": 2}
# [1, 2, 3]
# True
# hello
```

--------------------------------

### Drain stream with read_all() and check streaming status

Source: https://context7.com/daggaz/json-stream/llms.txt

Explicitly consumes all remaining data from a streaming object using `read_all()`. The `.streaming` attribute (or `.streaming` property) indicates whether the object still has unread content. `read_all()` is also automatically called by `load_many()` between documents.

```python
import json_stream
from io import StringIO

data = json_stream.load(StringIO('{"a": 1, "b": [1, 2, 3]}'), persistent=True)

print(data.streaming)    # True – not yet fully consumed
data.read_all()
print(data.streaming)    # False – fully consumed

# read_all() is also called automatically by load_many() between documents
```

--------------------------------

### Load JSON data from a file-like object

Source: https://github.com/daggaz/json-stream/blob/master/README.md

Use json_stream.load() to parse JSON data from a file-like or iterable object. This method does not require the entire document to be loaded into memory upfront.

```python
import json_stream
data = json_stream.load(f)
```

--------------------------------

### Custom JSON Encoding with JSONStreamEncoder

Source: https://context7.com/daggaz/json-stream/llms.txt

Demonstrates using `JSONStreamEncoder` as a custom encoder for `json.dumps` to handle json-stream objects. It can be used directly by passing `cls=JSONStreamEncoder` or globally within a `with` block.

```python
import json
from io import StringIO
from json_stream.dump import JSONStreamEncoder

json_data = '{"count": 3, "results": ["a", "b", "c"]}'

# Option 2: cls= argument (subclassable for custom encoding)
data = json_stream.load(StringIO(json_data))
print(json.dumps(data, cls=JSONStreamEncoder))
```

```python
# Option 3: context manager (patches JSONEncoder globally within the block)
data = json_stream.load(StringIO(json_data))
with JSONStreamEncoder():
    result = json.dumps(data)   # any call to json.dumps works here
print(result)
```

```python
# Works with persistent mode too
data = json_stream.load(StringIO(json_data), persistent=True)
_ = data["count"]  # partially consumed; encoder handles the rest
print(json.dumps(data, cls=JSONStreamEncoder))
```

--------------------------------

### Convert json-stream types to standard Python types

Source: https://github.com/daggaz/json-stream/blob/master/README.md

Use json_stream.to_standard_types to recursively convert json-stream dict-like and list-like objects to standard Python dicts and lists.

```python
# JSON: {"round": 1, "results": [1, 2, 3]}
data = json_stream.load(f)
results = data["results"]
print(results)  # prints <TransientStreamingJSONList: TRANSIENT, STREAMING>
converted = json_stream.to_standard_types(results)
print(converted)  # prints [1, 2, 3]
```

--------------------------------

### Convert streaming JSON to standard Python types with to_standard_types()

Source: https://context7.com/daggaz/json-stream/llms.txt

Recursively converts streaming JSON objects and lists into standard Python `dict` and `list` types, fully consuming the stream. Scalars are returned unchanged. This is useful when you need to access all data at once or perform operations that require fully materialized types.

```python
import json_stream
from json_stream import to_standard_types
from io import StringIO

json_data = '{"round": 1, "scores": [100, 200, {"bonus": 50}]}'

data = json_stream.load(StringIO(json_data))
scores = data["scores"]

# Still a TransientStreamingJSONList here
print(type(scores))  # <class 'TransientStreamingJSONList'>

standard = to_standard_types(scores)
print(standard)       # [100, 200, {'bonus': 50}]
print(type(standard)) # <class 'list'>

# Scalars pass through unchanged
print(to_standard_types(42))      # 42
print(to_standard_types("hello")) # hello
```

--------------------------------

### Load Multiple JSON Documents (Concatenated)

Source: https://context7.com/daggaz/json-stream/llms.txt

Use json_stream.load_many() to parse a stream of concatenated JSON documents without delimiters. The library automatically consumes content after each document is yielded.

```python
# --- Concatenated JSON (no delimiters for objects/arrays/literals) ---
concatenated = b'{"a":1}[1,2]truenull{}'
for doc in json_stream.load_many(BytesIO(concatenated)):
    print(to_standard_types(doc))
# {"a": 1}, [1, 2], True, None, {}
```

--------------------------------

### Transient mode JSON parsing

Source: https://github.com/daggaz/json-stream/blob/master/README.md

In transient mode (default), json_stream.load() parses JSON data iteratively. Only the currently accessed data is held in memory, and attempting to access previously skipped data raises a TransientAccessException. This mode is essential for processing large JSON datasets with low memory usage.

```python
import json_stream

# JSON: {"count": 3, "results": ["a", "b", "c"]}
data = json_stream.load(f)  # data is a transient dict-like object 
# stream has been read up to "{" 

# use data like a dict
results = data["results"]  # results is a transient list-like object
# stream has been read up to "[", we now cannot read "count"

# iterate transient list
for result in results:
    print(result)  # prints a, b, c
# stream has been read up to "]"

# attempt to read "count" from earlier in stream
# count = data["count"]  # will raise exception
# stream is now exhausted

# attempt to read from list that has already been iterated
# for result in results:  # will raise exception
#    pass

```

--------------------------------

### Persistent mode JSON parsing

Source: https://github.com/daggaz/json-stream/blob/master/README.md

In persistent mode, json_stream.load(f, persistent=True) stores all parsed data in memory, allowing access to any key or index. Requests for data not yet parsed will block until it is available in the stream. This mode provides a standard dict/list-like interface for the streamed data.

```python
import json_stream

# JSON: {"count": 1, "results": ["a", "b", "c"]}
data = json_stream.load(f, persistent=True)
# data is a streaming  dict-like object 
# stream has been read up to "{" 

# use data like a dict
results = data["results"]  # results is a streaming list-like object
# stream has been read up to "["
# count has been stored data

# use results like a list
a_result = results[1]  # a_result = "b"
# stream has been read up to the middle of list

```

--------------------------------

### Decorate Generator with streamable_dict

Source: https://github.com/daggaz/json-stream/blob/master/README.md

Use the streamable_dict decorator on a generator function to automatically make its output streamable as a JSON object. This is useful when the generator produces key-value pairs.

```python
import json
import sys

from json_stream import streamable_dict

# declare a new streamable dict generator function
@streamable_dict
def generate_dict_of_squares(n):
    for i in range(n):
        # this could be some memory intensive operation
        # or just a really large value of n
        yield i, i ** 2

# data is will already be Streamable because
# of the decorator
data = generate_dict_of_squares(10)
json.dump(data, sys.stdout)
```