series-kafka 1.0.1

Type-safe async Kafka SDK for Python microservices in the Series platform

Series Kafka SDK - Python

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

pip install series-kafka

Or with Poetry:

poetry add series-kafka

Quick Start

Producer

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="[e-mail]",
            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

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:

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

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

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:

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

Configuration

Producer Configuration

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

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

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:

docker-compose up -d

Run tests:

# 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

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

Code Quality

# 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:

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):

from series_kafka.exceptions import ValidationError

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

FatalError

Unrecoverable errors that go directly to DLQ:

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