Metadata-Version: 2.4
Name: chunkstream
Version: 0.1.0
Summary: Zero-dependency memory-efficient streaming chunk buffer, delimiter-aware frame accumulator, and SSE parser for Python.
Author-email: Hamidreza <h4midreza.ir@gmail.com>
License-Expression: MIT
Project-URL: Homepage, https://github.com/mrhamidrezam/chunkstream
Project-URL: Repository, https://github.com/mrhamidrezam/chunkstream
Project-URL: Issues, https://github.com/mrhamidrezam/chunkstream/issues
Keywords: streaming,buffer,sse,chunk,llm,frame,async,zero-dependency
Classifier: Development Status :: 4 - Beta
Classifier: Intended Audience :: Developers
Classifier: Programming Language :: Python :: 3
Classifier: Programming Language :: Python :: 3.10
Classifier: Programming Language :: Python :: 3.11
Classifier: Programming Language :: Python :: 3.12
Classifier: Programming Language :: Python :: 3.13
Classifier: Programming Language :: Python :: 3.14
Classifier: Topic :: Software Development :: Libraries :: Python Modules
Classifier: Typing :: Typed
Requires-Python: >=3.10
Description-Content-Type: text/markdown
License-File: LICENSE
Dynamic: license-file

# chunkstream 🌊

[![PyPI version](https://img.shields.io/pypi/v/chunkstream.svg)](https://pypi.org/project/chunkstream/)
[![Python versions](https://img.shields.io/pypi/pyversions/chunkstream.svg)](https://pypi.org/project/chunkstream/)
[![CI](https://github.com/mrhamidrezam/chunkstream/actions/workflows/ci.yml/badge.svg)](https://github.com/mrhamidrezam/chunkstream/actions/workflows/ci.yml)
[![License: MIT](https://img.shields.io/badge/License-MIT-blue.svg)](https://github.com/mrhamidrezam/chunkstream/blob/main/LICENSE)
[![Zero Dependency](https://img.shields.io/badge/dependencies-0-success.svg)](pyproject.toml)

**Zero-dependency, memory-efficient streaming chunk buffer, delimiter-aware frame accumulator, and Server-Sent Events (SSE) parser for Python.**

Designed specifically for streaming LLM tokens (OpenAI, Anthropic, Gemini, Ollama, DeepSeek), HTTP chunked transfer decoders, WebSocket packet framing, and async network pipelines.

---

## ⚡ Why chunkstream?

When consuming real-world streaming APIs, network sockets deliver data in **arbitrary packet slices** (e.g. `{"mess`, `age": "hel`, `lo"}`). Handling fragmented chunks, multibyte UTF-8 characters across boundaries, and Server-Sent Events without external dependencies or memory bloat is tedious.

`chunkstream` solves this with:
- **Zero Dependencies:** Pure standard library, compatible with Python 3.10 through 3.14.
- **Dual Sync & Async Buffers:** `ChunkBuffer` for synchronous streams and `AsyncChunkBuffer` for `asyncio` event-driven loops.
- **Spec-Compliant SSE Parser:** Parses `data:`, `event:`, `id:`, and `[DONE]` markers across fragmented chunks in both `parse_sse` and `parse_sse_async`.
- **Delimiter & Fixed-Length Framers:** Split byte streams by arbitrary delimiters (`\n`, `\r\n\r\n`, `\0`, or multipart boundaries).
- **Backpressure & Overflow Protection:** Configurable `max_size` prevents runaway memory allocation on slow consumers.
- **Sliding Stream Window:** Lookbehind and deduplication window for stream inspection.

---

## 📦 Installation

```bash
pip install chunkstream
```

---

## 🚀 Quickstart

### 1. Parsing Streaming LLM Server-Sent Events (SSE)

Seamlessly parse fragmented chunks from any LLM or HTTP client (`httpx`, `aiohttp`, `urllib`):

```python
from chunkstream import parse_sse

def mock_llm_stream():
    yield b'data: {"id": "c1", "choices": [{"delta": {"con'
    yield b'tent": "Hello"}}]}\n\n'
    yield b'data: {"id": "c1", "choices": [{"delta": {"content": " world!"}}]}\n\n'
    yield b'data: [DONE]\n\n'

for event in parse_sse(mock_llm_stream()):
    if event.is_done:
        print("\n[Generation Complete]")
        break
    print(f"Token: {event.data}")
```

### 2. Asynchronous SSE Stream Parsing

```python
import asyncio
from chunkstream import parse_sse_async

async def main():
    async for event in parse_sse_async(async_byte_generator()):
        if event.is_done:
            break
        print(event.data)

asyncio.run(main())
```

---

### 3. Delimiter Framing (ChunkBuffer)

Accumulate incoming network chunks and read complete lines or frames:

```python
from chunkstream import ChunkBuffer

buffer = ChunkBuffer(max_size=1024 * 1024) # 1MB limit

buffer.feed(b"frame_one\0frame_two\0partial_th")

# Read frame by frame
while frame := buffer.read_until(b"\0"):
    print("Frame:", frame.decode())

# Remaining data stays safely buffered
print("Unconsumed:", buffer.read()) # b"partial_th"
```

---

### 4. Sliding Stream Window

Track rolling history over an ongoing stream:

```python
from chunkstream import StreamWindow

window = StreamWindow(size=3)
for item in ["A", "B", "C", "D"]:
    window.feed(item)

print(window.snapshot()) # ['B', 'C', 'D']
```

---

## 🧪 Running Tests

```bash
pip install pytest
pytest
```

---

## 📄 License

[MIT](https://github.com/mrhamidrezam/chunkstream/blob/main/LICENSE) © 2026 Hamidreza ([mrhamidrezam](https://github.com/mrhamidrezam))
