Release v0.1.2 (What’s new?).
Welcome to unistream Documentation¶
unistream is a universal Producer / Consumer abstraction layer for stream systems. It lets you use a single, unified interface to send data to and pull data from any streaming backend — Apache Kafka, AWS Kinesis, AWS CloudWatch Logs, Apache Pulsar, or even local files — while the library handles batching, fault tolerance, retries, and exactly-once consumption behind the scenes.
The core library ships only local-file implementations. Vendor-specific integrations (AWS Kinesis, CloudWatch Logs, DynamoDB checkpoint, etc.) are released as separate plugin packages.
Features¶
Efficient Record Buffering: Groups records into micro-batches with a Write-Ahead Log (WAL) for crash recovery, optimizing network bandwidth without risking data loss.
Non-blocking Exponential Backoff: Automatic retry with configurable backoff schedules. The retry check is time-based (no
sleep), so your application thread is never blocked.Checkpoint & Exactly-Once Consumption: Per-record status tracking (pending → in_progress → succeeded / failed / exhausted), UUID-based concurrency locking with auto-expiry, and stream pointer persistence for fault-tolerant consumption.
Pluggable Architecture: Five core abstractions (Record, Buffer, Producer, CheckPoint, Consumer) with clean ABC → Base → Concrete layering. Implement
send()for a new producer orget_records()for a new consumer — the framework handles everything else.
Core Abstractions¶
AbcRecord/DataClassRecord— the atomic data unit (id + create_at + serialize/deserialize).AbcBuffer/FileBuffer— WAL-backed batch buffer (put → should_i_emit → emit → commit).AbcProducer/BaseProducer/SimpleProducer— non-blocking put() event loop with retry.AbcCheckPoint/BaseCheckPoint/SimpleCheckpoint— per-record state machine + persistence.AbcConsumer/BaseConsumer/SimpleConsumer— consumption loop with tenacity retry + DLQ hook.
AI Agent Skill¶
A self-contained Claude Code skill is included at .claude/skills/unistream/SKILL.md. It contains the complete API reference, protocols, and usage examples — any AI coding agent with this skill loaded can build custom producers, consumers, buffers, checkpoints, and records without reading the source code.
Install¶
unistream is released on PyPI, so all you need is to:
$ pip install unistream
To upgrade to latest version:
$ pip install --upgrade unistream