streaming-xml-parser 1.0.0
High-performance streaming XML parser for real-time applications
License: MIT
Requires Python: >=3.8
Keywords: xml, parser, streaming, real-time, performance, llm

# xmlstream

A high-performance, configurable streaming XML parser designed for real-time processing of XML content in streaming applications. Perfect for handling XML output from language models, API responses, and other streaming data sources.

## Features

- 🚀 **High Performance**: Optimized for minimal latency and maximum throughput
- 🔄 **Real-time Streaming**: Process XML content as it arrives, not after completion
- 🎯 **Configurable Behaviors**: Different handling modes for different XML tags
- 📚 **Stack-based Nesting**: Proper XML nesting support with streaming-aware rules
- ⚡ **Zero Dependencies**: Pure Python implementation with no external requirements
- 🛡️ **Production Ready**: Comprehensive error handling and safety limits
- 🏗️ **Event-driven Architecture**: Clean separation of parsing and output handling

## Tag Behaviors

The parser supports three distinct behaviors for XML tags:

- **STREAMING**: Content is streamed immediately as it arrives (blocks nesting)
- **PLACEHOLDER**: Shows status messages during processing (allows nesting)
- **SILENT**: Processes content without user feedback (allows nesting)

## Installation

```bash
pip install xmlstream
```

## Quick Start

```python
from xmlstream import StreamingXMLParser, TagConfig, TagBehavior

# Configure tag behaviors
tag_configs = {
    "reply": TagConfig(
        name="reply",
        behavior=TagBehavior.STREAMING,  # Stream content immediately
    ),
    "thinking": TagConfig(
        name="thinking", 
        behavior=TagBehavior.PLACEHOLDER,  # Show status message
        placeholder_message="🤔 Thinking..."
    ),
    "variables": TagConfig(
        name="variables",
        behavior=TagBehavior.SILENT,  # Process silently
    )
}

# Create parser
parser = StreamingXMLParser(tag_configs)

# Process streaming content
xml_content = "<reply>Hello, this is streaming content!</reply>"
for event in parser.process_chunk(xml_content):
    if event.event_type == "tag_content" and event.tag_name == "reply":
        print(event.content, end="", flush=True)
```

## Advanced Usage

### Real-time Processing with Callbacks

```python
from xmlstream import (
    StreamingXMLParser, TagConfig, TagBehavior, 
    StreamingOutputHandler
)

def on_reply_start(tag_name):
    print(f"\n🤖 {tag_name.title()}: ", end="", flush=True)

def on_reply_complete(tag_name, content):
    print(f"\n✅ {tag_name} completed ({len(content)} chars)")

# Configure with callbacks
reply_config = TagConfig(
    name="reply",
    behavior=TagBehavior.STREAMING,
    start_callback=on_reply_start,
    complete_callback=on_reply_complete
)

parser = StreamingXMLParser({"reply": reply_config})
output_handler = StreamingOutputHandler()

# Process with output handler
for event in parser.process_chunk("<reply>Streaming response...</reply>"):
    output_handler.handle_event(event)
```

### Multiple Output Handlers

```python
from xmlstream import CollectingOutputHandler, CallbackOutputHandler

# Collecting handler for testing
collector = CollectingOutputHandler()

# Custom callback handler
def handle_content(content):
    print(f"Received: {content}")

callback_handler = CallbackOutputHandler(on_content=handle_content)

# Use both handlers
for event in parser.process_chunk(xml_data):
    collector.handle_event(event)
    callback_handler.handle_event(event)

# Get collected content
full_content = collector.get_content()
all_events = collector.get_events()
```

### Configuration Validation

```python
from xmlstream import TagConfig, TagBehavior
from xmlstream.exceptions import ConfigurationError

try:
    # This will raise ConfigurationError
    invalid_config = TagConfig(
        name="",  # Empty name not allowed
        behavior=TagBehavior.PLACEHOLDER,
        placeholder_message=None  # Required for PLACEHOLDER
    )
except ConfigurationError as e:
    print(f"Configuration error: {e}")
```

## API Reference

### Core Classes

#### `StreamingXMLParser`

Main parser class for processing streaming XML content.

**Constructor:**
```python
StreamingXMLParser(tag_configs: Dict[str, TagConfig], max_buffer_size: int = 1024*1024)
```

**Key Methods:**
- `process_chunk(chunk: str) -> Generator[StreamingEvent, None, None]`
- `add_tag_config(config: TagConfig) -> None`
- `remove_tag_config(tag_name: str) -> bool`
- `reset() -> None`

#### `TagConfig`

Configuration for XML tag behavior.

```python
TagConfig(
    name: str,
    behavior: TagBehavior,
    placeholder_message: Optional[str] = None,
    start_callback: Optional[Callable[[str], Any]] = None,
    content_callback: Optional[Callable[[str], Any]] = None,
    complete_callback: Optional[Callable[[str, str], Any]] = None
)
```

#### `StreamingEvent`

Immutable event object representing parsing progress.

**Attributes:**
- `event_type: str` - Type of event (content, tag_start, tag_content, tag_complete)
- `tag_name: Optional[str]` - Associated tag name
- `content: Optional[str]` - Content data
- `data: Optional[Any]` - Additional payload

**Methods:**
- `is_content_event() -> bool`
- `is_tag_event() -> bool`
- `has_content() -> bool`

### Output Handlers

#### `StreamingOutputHandler`

Default handler for streaming output with placeholder support.

#### `CollectingOutputHandler`

Collects all output for batch processing or testing.

#### `CallbackOutputHandler`

Flexible handler using custom callbacks for each event type.

## Performance Optimizations

The parser includes several performance optimizations:

- **Conditional Logging**: Debug logging only when enabled
- **Single-pass Scanning**: Optimized buffer scanning algorithm
- **Minimal Allocations**: Efficient memory usage patterns
- **Buffer Management**: Smart buffering with safety limits

## Error Handling

The package includes comprehensive error handling:

```python
from xmlstream.exceptions import (
    StreamingXMLError,
    TagNotFoundError,
    InvalidTagError,
    ConfigurationError,
    BufferOverflowError
)

try:
    parser.process_chunk(malformed_xml)
except StreamingXMLError as e:
    print(f"Parser error: {e}")
```

## Examples

See the `/examples` directory for complete working examples:

- **Basic Streaming**: Simple real-time content processing
- **LLM Integration**: Integration with language model APIs
- **Custom Handlers**: Building custom output processors
- **Error Handling**: Robust error management patterns

## Development

```bash
# Clone repository
git clone https://github.com/example/xmlstream.git
cd xmlstream

# Install development dependencies
pip install -e

# Run tests
pytest

# Code formatting
black xmlstream/

# Type checking
mypy xmlstream/
```

## Performance Benchmarks

- **Latency**: < 1ms for typical chunks
- **Throughput**: > 100MB/s on modern hardware
- **Memory**: Constant memory usage with buffer limits
- **CPU**: Optimized single-pass algorithms

## License

MIT License - see LICENSE file for details.

## Contributing

1. Fork the repository
2. Create a feature branch
3. Add tests for new functionality
4. Ensure all tests pass
5. Submit a pull request

## Changelog

### v1.0.0
- Initial release
- Core streaming parser implementation
- Multiple output handlers
- Comprehensive error handling
- Performance optimizations