Metadata-Version: 2.4
Name: esio-kinesis
Version: 1.2.0
Summary: Async AWS Kinesis client with consumer decorator, checkpointing, and factory pattern for EdgeStack IoT platform
Project-URL: Homepage, https://github.com/edgestackio/esio-kinesis
Project-URL: Repository, https://github.com/edgestackio/esio-kinesis
Project-URL: Issues, https://github.com/edgestackio/esio-kinesis/issues
Author-email: EdgeStack IO <dev@edgestack.io>
License: MIT
License-File: LICENSE
Keywords: async,aws,edgestack,iot,kinesis,streaming
Classifier: Development Status :: 4 - Beta
Classifier: Framework :: AsyncIO
Classifier: Intended Audience :: Developers
Classifier: License :: OSI Approved :: MIT License
Classifier: Programming Language :: Python :: 3
Classifier: Programming Language :: Python :: 3.11
Classifier: Programming Language :: Python :: 3.12
Classifier: Topic :: Internet
Classifier: Topic :: Software Development :: Libraries :: Python Modules
Requires-Python: >=3.11
Requires-Dist: aioboto3>=13.0.0
Requires-Dist: structlog>=25.0.0
Description-Content-Type: text/markdown

# esio-kinesis

Async AWS Kinesis client for the EdgeStack IoT platform. Provides publish, consume, and stream management with a factory pattern that supports sequential, concurrent, and Enhanced Fan-Out (EFO) consumer strategies.

## Features

- **Async-native** — built on `aioboto3`, safe in `asyncio` task groups
- **Consumer decorator pattern** — `ack()` / `rollback(cause)` for explicit at-least-once processing
- **Redis checkpointing** — resumes from last processed sequence number on restart
- **LocalStack compatible** — `AT_TIMESTAMP` falls back to `TRIM_HORIZON` automatically
- **Factory pattern** — swap `KINESIS_CLIENT_TYPE=simple|concurrent|efo` via env var, zero service code changes

## Installation

```bash
pip install esio-kinesis
```

## Quick start

```python
from esio_kinesis import KinesisClientFactory

consumer = KinesisClientFactory.get(
    client_type='simple',
    endpoint_url='http://localhost:4566',   # omit for real AWS
    region='us-east-1',
    checkpoint_store=redis_client,          # optional Redis for checkpointing
    consumer_id='my-service',
)

# Publish
await consumer.put_record('my-stream', '{"hello": "world"}', partition_key='key')

# Consume — ack/rollback are explicit
async for message in consumer.consume('my-stream'):
    await process(message)
    await consumer.ack()
```

## Consumer types

| `KINESIS_CLIENT_TYPE` | Description | Status |
|---|---|---|
| `simple` | Sequential polling with Redis checkpointing | Available |
| `concurrent` | Semaphore-bounded concurrent batch processing | Planned |
| `efo` | Enhanced Fan-Out via `SubscribeToShard` | Planned |

## License

MIT
