series-kafka 1.0.1
Type-safe async Kafka SDK for Python microservices in the Series platform
License: MIT
Requires Python: <4.0,>=3.12
Keywords: kafka, async, microservices, event-streaming

# Series Kafka SDK - Python

[![Tests](https://img.shields.io/badge/tests-154%20passing-brightgreen)](https://github.com/series/kafka-sdk-python)
[![Coverage](https://img.shields.io/badge/coverage-81.15%25-green)](https://github.com/series/kafka-sdk-python)
[![Python](https://img.shields.io/badge/python-3.12+-blue)](https://www.python.org/downloads/)
[![License](https://img.shields.io/badge/license-MIT-blue)](LICENSE)

Type-safe async event streaming for Python microservices in the Series platform.

## Features

- ✅ **Async/Await** - Built on aiokafka for non-blocking I/O
- ✅ **Type Safety** - Pydantic models with full validation
- ✅ **Idempotent** - Exactly-once delivery by default
- ✅ **Auto DLQ** - Per-topic dead letter queues with metadata tracking
- ✅ **Observability** - OpenTelemetry tracing and metrics (optional)
- ✅ **Plugin System** - Extensible topic architecture
- ✅ **Schema Validation** - JSON Schema with Pydantic
- ✅ **Subset Fields** - Extract only needed fields for performance
- ✅ **Retry Logic** - Exponential backoff for transient errors
- ✅ **Graceful Shutdown** - Proper offset commits with timeout

## Installation

```bash
pip install series-kafka
```

Or with Poetry:

```bash
poetry add series-kafka
```

## Quick Start

### Producer

```python
import asyncio
from datetime import datetime, timezone
from series_kafka import AsyncProducer, ProducerConfig
from series_kafka_topics.users import UsersTopic, UserCreatedPayload

async def main():
    # Configure producer
    config = ProducerConfig(
        bootstrap_servers="localhost:9092",
        service_name="my-service",
        security_protocol="PLAINTEXT"  # Use SASL_SSL for production
    )
    
    # Create and start producer
    producer = AsyncProducer(config)
    await producer.start()
    
    try:
        # Create payload
        payload = UserCreatedPayload(
            user_id="usr_123",
            email="user@example.com",
            username="johndoe",
            created_at=datetime.now(timezone.utc)
        )
        
        # Produce event
        event_id = await producer.produce(
            topic=UsersTopic(),
            payload=payload,
            event_type="user.created"
        )
        
        print(f"Event produced: {event_id}")
        
    finally:
        await producer.stop()

asyncio.run(main())
```

### Consumer

```python
import asyncio
from series_kafka import AsyncConsumer, ConsumerConfig, BaseMessage
from series_kafka_topics.users import UsersTopic

async def handle_message(message: BaseMessage):
    """Process incoming message."""
    print(f"Received: {message.event_type}")
    print(f"User ID: {message.payload.user_id}")
    print(f"Email: {message.payload.email}")

async def main():
    # Configure consumer
    config = ConsumerConfig(
        bootstrap_servers="localhost:9092",
        group_id="my-consumer-group",
        service_name="my-service",
        security_protocol="PLAINTEXT",
        enable_dlq=True,  # Enable dead letter queue
        max_retries=3
    )
    
    # Create and start consumer
    consumer = AsyncConsumer(
        config=config,
        topics=[UsersTopic()],
        handler=handle_message
    )
    
    await consumer.start()
    await consumer.consume()  # Runs until stopped

asyncio.run(main())
```

## Advanced Features

### Subset Field Extraction

Extract only specific fields for performance and security:

```python
config = ConsumerConfig(
    bootstrap_servers="localhost:9092",
    group_id="my-consumer",
    service_name="my-service",
    security_protocol="PLAINTEXT",
    # Extract only needed fields
    subset_fields=[
        "event_id",
        "payload.user_id",
        "payload.username"
    ]
)

async def handle_fields(fields: dict):
    """Handler receives only extracted fields."""
    print(f"User ID: {fields['user_id']}")
    print(f"Username: {fields['username']}")

consumer = AsyncConsumer(config, [UsersTopic()], handle_fields)
```

### Error Handling with DLQ

```python
from series_kafka.exceptions import RetryableError, ValidationError, FatalError

async def handle_message(message: BaseMessage):
    try:
        # Process message
        result = await process_data(message.payload)
        
    except NetworkError as e:
        # Transient errors will be retried with exponential backoff
        raise RetryableError(f"Network issue: {e}")
        
    except InvalidDataError as e:
        # Validation errors go directly to DLQ (no retry)
        raise ValidationError(f"Bad data: {e}")
        
    except CriticalError as e:
        # Fatal errors go directly to DLQ
        raise FatalError(f"Critical failure: {e}")
```

### Event Type Filtering

```python
config = ConsumerConfig(
    bootstrap_servers="localhost:9092",
    group_id="my-consumer",
    service_name="my-service",
    security_protocol="PLAINTEXT",
    # Only process specific event types
    event_type_filter={"user.created", "user.updated"}
)
```

### Idempotent Producer

Idempotency is enabled by default for exactly-once delivery:

```python
config = ProducerConfig(
    bootstrap_servers="localhost:9092",
    service_name="my-service",
    security_protocol="PLAINTEXT",
    enable_idempotence=True  # Default: True
)
```

## Configuration

### Producer Configuration

```python
from series_kafka import ProducerConfig

config = ProducerConfig(
    # Connection (required)
    bootstrap_servers="localhost:9092",
    service_name="my-service",
    
    # Authentication (optional)
    sasl_username="your-api-key",
    sasl_password="your-api-secret",
    sasl_mechanism="PLAIN",  # PLAIN, SCRAM-SHA-256, SCRAM-SHA-512
    security_protocol="SASL_SSL",  # PLAINTEXT, SSL, SASL_PLAINTEXT, SASL_SSL
    
    # Idempotency (default: True)
    enable_idempotence=True,
    
    # Performance tuning
    compression_type="gzip",  # none, gzip, snappy, lz4, zstd
    batch_size=16384,
    linger_ms=10,
    
    # Observability (default: False)
    enable_metrics=False,
    enable_tracing=False,
)
```

### Consumer Configuration

```python
from series_kafka import ConsumerConfig

config = ConsumerConfig(
    # Connection (required)
    bootstrap_servers="localhost:9092",
    group_id="my-consumer-group",
    service_name="my-service",
    
    # Authentication (optional)
    sasl_username="your-api-key",
    sasl_password="your-api-secret",
    
    # DLQ settings
    enable_dlq=True,
    max_retries=3,
    retry_backoff_ms=1000,
    
    # Event filtering
    event_type_filter={"user.created", "user.updated"},
    
    # Subset field extraction
    subset_fields=["payload.user_id", "payload.username"],
    
    # Consumer settings
    auto_offset_reset="earliest",  # earliest or latest
    shutdown_timeout_seconds=30,
)
```

## Topic Plugins

The SDK includes built-in topic plugins for Series platform:

- **MessagesTopic** - Communication events (SendBlue, Linq, Agent responses)
- **UsersTopic** - User lifecycle events (created, updated, deleted)
- **PostsTopic** - Content lifecycle events (created, updated, deleted)

### Creating Custom Topics

```python
from series_kafka import Topic, BasePayload
from series_kafka.contracts.base import TopicContract
from pydantic import Field

# Define payload
class ProductCreatedPayload(BasePayload):
    product_id: str = Field(..., pattern=r"^prod_")
    name: str
    price: float

# Implement topic plugin
class ProductsTopic(Topic):
    @property
    def name(self) -> str:
        return "products"
    
    @property
    def schema_registry(self) -> dict[str, type[BasePayload]]:
        return {
            "product.created": ProductCreatedPayload,
        }
    
    def get_partition_key(self, message, event_type) -> str | None:
        return message.payload.product_id
    
    def validate_message(self, message) -> bool:
        return message.payload.price > 0
    
    def get_contract(self) -> TopicContract:
        return TopicContract(
            name=self.name,
            version="1.0.0",
            event_types={"product.created": {"type": "object"}}
        )
```

## Testing

### Local Development

Start local Kafka with Docker Compose:

```bash
docker-compose up -d
```

Run tests:

```bash
# Unit tests
pytest tests/unit/ -v

# Integration tests (requires local Kafka)
pytest tests/integration/ -v

# All tests with coverage
pytest --cov=series_kafka --cov=series_kafka_topics
```

### Test Results

- ✅ 154 tests passing
- ✅ 81.15% code coverage
- ✅ Zero bugs
- ✅ All E2E tests validated against real Kafka

## Development

### Setup

```bash
git clone https://github.com/series/kafka-sdk-python
cd kafka-sdk-python
poetry install
```

### Code Quality

```bash
# Format code
black series_kafka series_kafka_topics tests

# Lint code
ruff check series_kafka series_kafka_topics tests

# Type check
mypy series_kafka
```

## Architecture

```
series_kafka/
├── core/              # Core SDK components
│   ├── producer.py    # AsyncProducer
│   ├── consumer.py    # AsyncConsumer
│   ├── message.py     # BaseMessage, BasePayload
│   ├── topic.py       # Topic ABC
│   ├── config.py      # Configuration models
│   ├── dlq.py         # DLQ handler
│   └── field_extractor.py  # Subset field extraction
├── schema/            # Validation & serialization
│   ├── validator.py   # Schema validation
│   └── serializer.py  # Message serialization
├── contracts/         # Data contracts
│   └── base.py        # TopicContract
└── exceptions.py      # Exception hierarchy

series_kafka_topics/
├── messages/          # Messages topic plugin
├── users/             # Users topic plugin
└── posts/             # Posts topic plugin
```

## Error Handling

The SDK provides three error types for proper DLQ routing:

### RetryableError

Transient errors that should be retried with exponential backoff:

```python
from series_kafka.exceptions import RetryableError

# Consumer will retry up to max_retries times
raise RetryableError("Temporary network issue")
```

### ValidationError

Data validation errors that go directly to DLQ (no retry):

```python
from series_kafka.exceptions import ValidationError

# Goes to DLQ immediately
raise ValidationError("Invalid email format")
```

### FatalError

Unrecoverable errors that go directly to DLQ:

```python
from series_kafka.exceptions import FatalError

# Goes to DLQ immediately
raise FatalError("Authentication failed")
```

## Contributing

1. Fork the repository
2. Create a feature branch
3. Make your changes
4. Run tests: `pytest`
5. Format code: `black .`
6. Lint: `ruff check .`
7. Submit pull request

## License

MIT License - See LICENSE file for details.

## Support

For issues and questions:
- GitHub Issues: https://github.com/series/kafka-sdk-python/issues
- Documentation: https://docs.series.io/kafka-sdk-python