jsonversation 0.1.3
Fast and type-safe JSON stream parser
Requires Python: >=3.13

# jsonversation

A Python library for streaming JSON parsing that enables real-time processing of JSON data as it arrives. Perfect for handling large JSON responses, streaming APIs, or any scenario where you need to process JSON data progressively.



## Installation

Coming soon!


## Key Features

-   **Streaming JSON Parsing**: Process JSON data as it arrives, without waiting for complete payloads
-   **Type-Safe Models**: Define strongly-typed data structures with automatic validation
-   **Real-time Callbacks**: React to data changes as they happen with customizable callbacks
-   **Progressive Updates**: Handle partial JSON data and incremental updates seamlessly



## Quick Start

### Basic Usage

```python
import jsonversation as jv


# Define your data structure
class ChatMessage(jv.Object):
    role: jv.String
    content: jv.String
    timestamp: jv.String
    token_count: jv.Atomic[int]
    is_complete: jv.Atomic[bool]


# Create a parser
message = ChatMessage()

# Process streaming JSON data
json_chunk1 = '{"role": "assistant", "content": "Hello'
json_chunk2 = ' world!", "timestamp": "2024-01-01T12:00:00Z", "token_count": 42, "is_complete": true}'

with jv.Parser(message) as parser:
    parser.push(json_chunk1)
    print(message.content.get_value())  # "Hello"

    parser.push(json_chunk2)
    print(message.content.get_value())  # "Hello world!"
    print(message.role.get_value())  # "assistant"
    print(message.token_count.get_value())  # 42
    print(message.is_complete.get_value())  # True
```


### Real-time Callbacks

```python
import jsonversation as jv


class StreamingResponse(jv.Object):
    message: jv.String
    tokens: jv.List[jv.String]


response = StreamingResponse()


# Set up real-time callbacks
def on_message_update(chunk: str):
    print(f"New text: {chunk}")


def on_new_token(token: jv.String):
    print(f"New token: {token.get_value()}")


def on_message_complete(full_message: str):
    print(f"Complete message: {full_message}")


response.message.on_append(on_message_update)
response.tokens.on_append(on_new_token)
response.message.on_complete(on_message_complete)

# Simulate streaming data
with jv.Parser(response) as parser:
    parser.push('{"message": "The quick')
    parser.push(' brown fox", "tokens": ["The", "quick"]}')
```

### Working with Atomic Types

The `Atomic` class handles primitive types that arrive as complete values rather than streaming text:

```python
import jsonversation as jv


class APIResponse(jv.Object):
    message: jv.String
    status_code: jv.Atomic[int]
    success: jv.Atomic[bool]
    response_time: jv.Atomic[float]


response = APIResponse()


# Set up callbacks for atomic values
def on_status_change(status: int | None):
    if status:
        print(f"Status code: {status}")
        if status >= 400:
            print("Error response detected!")


def on_completion(success: bool | None):
    if success is not None:
        print(f"Request {'succeeded' if success else 'failed'}")


response.status_code.on_complete(on_status_change)
response.success.on_complete(on_completion)

# Process JSON with atomic values
with jv.Parser(response) as parser:
    parser.push('{"message": "Request completed", "status_code": 200')
    parser.push(', "success": true, "response_time": 1.23}')
    # Callbacks will be triggered when atomic values are completed

print(response.status_code.get_value())  # 200
print(response.success.get_value())  # True
print(response.response_time.get_value())  # 1.23
```


### Complex Nested Structures

```python
import jsonversation as jv


class Author(jv.Object):
    name: jv.String
    email: jv.String


class Comment(jv.Object):
    id: jv.String
    text: jv.String
    author: Author


class BlogPost(jv.Object):
    title: jv.String
    content: jv.String
    author: Author
    comments: jv.List[Comment]
    tags: jv.List[jv.String]


# Create and use the parser
post = BlogPost()

# Process complex JSON
complex_json = """
{
    "title": "Streaming JSON in Python",
    "content": "This is a comprehensive guide...",
    "author": {
        "name": "Jane Doe",
        "email": "jane@example.com"
    },
    "comments": [
        {
            "id": "1",
            "text": "Great article!",
            "author": {
                "name": "John Smith",
                "email": "john@example.com"
            }
        }
    ],
    "tags": ["python", "json", "streaming"]
}
"""

with jv.Parser(post) as parser:
    parser.push(complex_json)
    # Access nested data
    print(post.title.get_value())  # "Streaming JSON in Python"
    print(post.author.name.get_value())  # "Jane Doe"
    print(post.comments.get_value()[0].text.get_value())  # "Great article!"
    print([tag.get_value for tag in post.tags.get_value()])  # ["python", "json", "streaming"]
```

## AsyncIO

Jsonversation supports async callbacks as well. The API is very similar between the sync/async classes, with the only difference being that the async module supports async callback functions.

### Example

```python
import asyncio
import jsonversation.aio as jv


class StreamingResponse(jv.Object):
    message: jv.String
    tokens: jv.List[jv.String]


response = StreamingResponse()


# Set up real-time callbacks
async def on_message_update(chunk: str):
    await asyncio.sleep(0.001)
    print(f"New text: {chunk}")


async def on_new_token(token: jv.String):
    await asyncio.sleep(0.001)
    print(f"New token: {token.get_value()}")


async def on_message_complete(full_message: str):
    await asyncio.sleep(0.001)
    print(f"Complete message: {full_message}")


response.message.on_append(on_message_update)
response.tokens.on_append(on_new_token)
response.message.on_complete(on_message_complete)


async def run():
    async with jv.Parser(response) as parser:
        await parser.push('{"message": "The quick')
        await parser.push(' brown fox", "tokens": ["The", "quick"]}')


asyncio.run(run())
```

## API Reference

### Core Classes

1.  `Parser(obj: Object)`

    Main parser class for processing streaming JSON data.
    
    **Methods:**
    
    -   `push(chunk: str)` - Process a chunk of JSON data

2.  `Object`

    Base class for defining structured data models.
    
    **Methods:**
    
    -   `update(value: dict)` - Update object with dictionary data

3.  `String`

    String field type with streaming capabilities.
    
    **Properties:**
    
    -   `value` - Get current string value
    
    **Methods:**
    
    -   `on_append(callback)` - Register callback for new text chunks
    -   `on_complete(callback)` - Register callback for completion
    -   `update(value: str)` - Update with new string value

4.  `List[T]`

    List field type for collections of items.
    
    **Properties:**
    
    -   `value` - Get current list of items
    
    **Methods:**
    
    -   `on_append(callback)` - Register callback for new items
    -   `on_complete(callback)` - Register callback for completion
    -   `update(value: list)` - Update with new list data




## Requirements

-   Python 3.9+
-   jiter (for JSON parsing)



## Contributing

Contributions are welcome! Please feel free to submit a Pull Request.



## License

MIT