sify-queue-kafka 0.1.3
Enterprise Queue management Framework
License: MIT
Requires Python: >=3.8
Keywords: kafka, events, queue, messaging

# Sify Queue Kafka

Enterprise Queue management Framework for building robust, scalable event-driven applications.

## Features

- **Event-Driven Architecture**: Built on Apache Kafka for reliable message streaming
- **Type Safety**: Pydantic models for event validation and serialization
- **Retry & Resilience**: Exponential backoff retry with configurable policies
- **Idempotency**: Built-in duplicate event detection and prevention
- **Dead Letter Queue**: Automatic failed event routing for error handling
- **Stage-Based Processing**: Multi-stage event pipelines with stage-specific handlers
- **Comprehensive Logging**: Structured logging throughout the framework
- **Exception Handling**: Custom exception types for better error management

## Installation

```bash
pip install sify-queue-kafka
```

For development with additional tools:
```bash
pip install sify-queue-kafka[dev]
```

## Quick Start

### Basic Producer

```python
from datetime import datetime, timezone
from queue_kafka import Producer

producer = Producer()

def send_user_created_event():
    message = {
        "eventType": "USER_CREATED",
        "eventVersion": "1.0",
        "source": "user-service",
        "tenantId": "tenant-123",
        "payload": {
            "userId": "user-456",
            "email": "user@example.com",
            "name": "John Doe"
        },
        "metadata": {
            "stage": "SendWelcomeEmail",
            "created_at": datetime.now(timezone.utc).isoformat()
        }
    }

    producer.send("user-topic", message)
    print("USER_CREATED event sent")

if __name__ == "__main__":
    try:
        send_user_created_event()
    finally:
        producer.close()
        print("Producer closed")
```

### Basic Consumer (with stage)

```python
from queue_kafka import Consumer, event_handler, KafkaEvent


@event_handler("USER_CREATED", stage="EMAIL_NOTIFICATION")
def send_welcome_email(event: KafkaEvent):
    print(f"Sending welcome email to: {event.payload['email']}")

if __name__ == "__main__":

    consumer = Consumer(
        topics="user-topic",
        config={
            "stage": "EMAIL_NOTIFICATION",
            "tenant_id": "tenant-123",
            "source": "email-service",
        }
    )

    try:
        print("Consumer running... Press Ctrl+C to stop")
        consumer.start()  
    except KeyboardInterrupt:
        print("\nKeyboard interrupt received")
    finally:
        consumer.stop()
        print("Consumer stopped")
```

### Basic Consumer (without stage)

```python
from queue_kafka import Consumer, event_handler, KafkaEvent

@event_handler("USER_CREATED")
def send_welcome_email(event: KafkaEvent):
    print(f"Sending welcome email to: {event.payload['email']}")


if __name__ == "__main__":

    consumer = Consumer(
        topics="user-topic",
        config={
            "tenant_id": "tenant-123",
            "source": "email-service",
        }
    )

    try:
        print("Consumer running... Press Ctrl+C to stop")
        consumer.start()   
    except KeyboardInterrupt:
        print("\nKeyboard interrupt received")
    finally:
        consumer.stop()
        print("Consumer stopped")
```
### Error Handling and DLQ

```python
from queue_kafka import Consumer, event_handler, KafkaEvent, EventProcessingError

@event_handler("PAYMENT_INIT", stage="ProcessPayment")
def process_payment(event: KafkaEvent):
    print("Payment processed successfully")


if __name__ == "__main__":

    process_consumer = Consumer(
        topics=payment-topic,
        config={
            "stage": "ProcessPayment",
            "tenant_id": "tenant-222",
            "source": "payment-service",
            "dead_letter_topic": payment-dlq-topic
        }
    )

    try:
        process_consumer.start()

    except KeyboardInterrupt:
        print("\nShutting down...")

    finally:
        process_consumer.stop()
        print("Consumer stopped")
```

## Configuration Options

| Option | Type | Default | Description |
|--------|------|---------|-------------|
| `bootstrap_servers` | str | Required | Kafka bootstrap servers |
| `group_id` | str | None | Consumer group ID |
| `enable_auto_commit` | bool | False | Enable auto commit of offsets |
| `auto_offset_reset` | str | "earliest" | Offset reset policy |
| `enable_idempotency` | bool | True | Enable duplicate detection |
| `enable_retry` | bool | True | Enable retry with backoff |
| `dead_letter_topic` | str | None | Topic for failed events |
| `stage` | str | None | Processing stage name |

## Event Model

```python
class KafkaEvent(BaseModel):
    eventId: str 
    eventType: str
    eventVersion: str
    source: str
    tenantId: str
    timestamp: datetime
    correlationId: Optional[str]
    traceId: Optional[str]
    priority: Optional[str]
    retryCount: int
    maxRetries: int
    payload: Dict[str, Any]
    metadata: Optional[Dict[str, Any]]
```

## Exception Types

- `QueueSDKError`: Base exception for all SDK errors
- `ConfigurationError`: Configuration-related errors
- `HandlerNotFoundError`: No handler found for event type/stage
- `EventProcessingError`: Event processing failures
- `ProducerError`: Producer operation failures
- `ConsumerError`: Consumer operation failures
- `SerializationError`: Event serialization/deserialization errors


## Architecture

```
┌─────────────────┐    ┌─────────────────┐    ┌─────────────────┐
│    Producer     │───▶│    Kafka Topic  │───▶│    Consumer     │
└─────────────────┘    └─────────────────┘    └─────────────────┘
                                                       │
                                                       ▼
                                            ┌─────────────────┐
                                            │   Dispatcher    │
                                            └─────────────────┘
                                                       │
                                                       ▼
                                            ┌─────────────────┐
                                            │ Event Registry  │
                                            └─────────────────┘
                                                       │
                                                       ▼
                                            ┌─────────────────┐
                                            │ Event Handler   │
                                            └─────────────────┘
```

## License

This project is licensed under the MIT License - see the [LICENSE](LICENSE) file for details.

## Changelog

### v0.1.3
- Initial release
- Basic producer/consumer functionality
- Event validation with Pydantic
- Retry and idempotency features
- Dead letter queue support
- Comprehensive logging
- Unit test coverage