Metadata-Version: 2.1
Name: wkafka
Version: 1.5.1
Summary: A professional, decorator-based Kafka wrapper for Python with built-in support for multimedia and JSON.
Home-page: https://github.com/wisrovi/wkafka
License: MIT
Keywords: kafka,wrapper,decorators,multimedia,vision,opencv
Author: William Steve Rodriguez Villamizar
Author-email: wisrovi.rodriguez@gmail.com
Requires-Python: >=3.9,<3.15
Classifier: Development Status :: 5 - Production/Stable
Classifier: Intended Audience :: Developers
Classifier: License :: OSI Approved :: MIT License
Classifier: Operating System :: OS Independent
Classifier: Programming Language :: Python :: 3
Classifier: Programming Language :: Python :: 3.9
Classifier: Programming Language :: Python :: 3.10
Classifier: Programming Language :: Python :: 3.11
Classifier: Programming Language :: Python :: 3.12
Classifier: Programming Language :: Python :: 3.13
Classifier: Programming Language :: Python :: 3.14
Classifier: Topic :: Software Development :: Libraries :: Python Modules
Provides-Extra: all
Provides-Extra: redis
Provides-Extra: snappy
Requires-Dist: Pillow (>=10.2.0,<11.0.0)
Requires-Dist: PyYAML (>=6.0.1,<7.0.0)
Requires-Dist: kafka-python-ng (>=2.2.2,<3.0.0)
Requires-Dist: loguru (>=0.7.2,<0.8.0)
Requires-Dist: numpy (>=1.26.0,<3.0.0)
Requires-Dist: opencv-python (>=4.9.0,<5.0.0)
Requires-Dist: python-snappy (>=0.7.1,<0.8.0) ; extra == "snappy" or extra == "all"
Requires-Dist: wredis (>=1.1.0) ; extra == "redis" or extra == "all"
Project-URL: Repository, https://github.com/wisrovi/wkafka
Description-Content-Type: text/markdown

<img width="1040" height="582" alt="image" src="https://github.com/user-attachments/assets/e03505ea-5bb0-4e99-b80d-c0d5c261a322" />

<p align="center">
  <a href="https://pypi.org/project/wkafka/"><img src="https://img.shields.io/pypi/v/wkafka?style=for-the-badge&logo=pypi&color=3b82f6" alt="PyPI version" /></a>
  <a href="https://pypi.org/project/wkafka/"><img src="https://img.shields.io/pypi/dm/wkafka?style=for-the-badge&color=10b981" alt="PyPI Downloads" /></a>
  <a href="https://pypi.org/project/wkafka/"><img src="https://img.shields.io/pypi/pyversions/wkafka?style=for-the-badge&logo=python&color=ffd343&logoColor=black" alt="Python Versions" /></a>
  <a href="https://linkedin.com/in/wisrovi-rodriguez"><img src="https://img.shields.io/badge/LinkedIn-0077B5?style=for-the-badge&logo=linkedin&logoColor=white" alt="LinkedIn" /></a>
  <a href="https://wisrovi.dev"><img src="https://img.shields.io/badge/Author-wisrovi.dev-111827?style=for-the-badge&logo=google-chrome&logoColor=white" alt="Portal" /></a>
  <a href="https://orcid.org/0009-0005-0710-1861"><img src="https://img.shields.io/badge/ORCID-0009--0005--0710--1861-A6CE39?style=for-the-badge&logo=orcid&logoColor=white" alt="ORCID" /></a>
  <a href="https://opensource.org/licenses/MIT"><img src="https://img.shields.io/badge/License-MIT-yellow.svg?style=for-the-badge" alt="License" /></a>
</p>

# WKafka 🚀

**Production-Grade, Decorator-Driven Apache Kafka Framework for Python.**

**WKafka** simplifies event-driven architectures by eliminating boilerplate code. It transforms raw, low-level Apache Kafka clients into an intuitive, elegant, and resilient development experience. With first-class native support for **JSON**, **YAML**, **Computer Vision Images (OpenCV / Pillow)**, **Arbitrary Binary Files (PDF, ZIP, TXT)**, and **Strict Pydantic Validation**, WKafka powers high-throughput microservices, IoT telemetry, real-time multimedia, and AI/ML inference pipelines.

---

## 🌟 Why WKafka?

Raw Kafka clients require dozens of lines of repetitive boilerplate: manually serializing bytes, managing poll loops, handling heartbeat timeouts during heavy tasks, orchestrating thread pools, decoding headers, and configuring retries. 

**WKafka solves these challenges out of the box:**

* 🎯 **Clean Decorator Syntax**: Register consumers with a single `@kafka.consumer(...)` annotation.
* 🧬 **5 Native Serializers**: Seamlessly transfer JSON, YAML, OpenCV/NumPy images, raw files, and Pydantic schemas without manual byte conversion.
* ⚡ **Decoupled Heavy Computation**: Prevent Kafka's infamous `CommitFailedError` / `max_poll_interval_ms` rebalance loops during hours-long ML tasks via `background=True` and `commit_on_receive=True`.
* 🛡️ **Built-in Resilience**: Automatic retries with exponential backoff and automated Dead Letter Queue (DLQ) routing with diagnostic error headers.
* 🚀 **Dynamic Partition Auto-Scaling**: Automatically detect consumer concurrency and scale topic partitions on the broker via `KafkaAdminClient` (`partition_scale=True`).
* ⚡ **High-Throughput Payload Offloading**: Stream high-FPS computer vision matrices or heavy files directly in Redis RAM with automated TTL via `wredis` while Kafka routes ultra-lightweight Claim-Check references (~120 bytes).
* 🔒 **Enterprise Security**: Native support for SASL (PLAIN, SCRAM-SHA-256, SCRAM-SHA-512), SSL/TLS, and SOCKS5 proxies.
* ⚡ **Async/Await Ready**: Native execution of asynchronous coroutine handlers (`async def`).

---

## 📦 Installation

```bash
# Basic installation via pip
pip install wkafka

# Via poetry
poetry add wkafka

# With high-speed Redis payload offload support
pip install "wkafka[redis]"

# High-performance snappy compression support
pip install "wkafka[snappy]"

# All optional enterprise features
pip install "wkafka[all]"
```

---

## 🚀 5-Minute Quick Start

### 1. Minimal Producer & Consumer
```python
from wkafka import WKafka

# Connects to KAFKA_SERVER environment variable or defaults to localhost:9092
kafka = WKafka(bootstrap_servers="localhost:9092")

# Register consumer handler
@kafka.consumer(topic="orders.created", format="json")
def on_order_created(msg):
    print(f"📦 [Consumer] Order received: {msg.value}")
    print(f"   Topic: {msg.topic}, Partition Offset: {msg.offset}")

# Start consumers in a managed thread pool
kafka.run_consumers(block=False)

# Produce message safely within a context manager
with kafka.producer() as p:
    p.send(
        topic="orders.created",
        value={"order_id": "ORD-9842", "customer": "Alice", "total": 149.99},
        format="json",
    )
```

---

## 🧬 Built-in Serializers Deep Dive

WKafka includes 5 specialized serializers. No manual `.encode("utf-8")`, `json.loads()`, or image decoding is required:

### 1. JSON (`format="json"`)
Serializes standard Python dictionaries, lists, primitives, and nested objects.
```python
with kafka.producer() as p:
    p.send("iot.telemetry", value={"device_id": "sensor-12", "temp": 23.5}, format="json")

@kafka.consumer(topic="iot.telemetry", format="json")
def handle_telemetry(msg):
    print(f"Temperature: {msg.value['temp']} °C")
```

### 2. YAML (`format="yaml"`)
Ideal for configuration payloads, declarative job descriptions, and Kubernetes-style manifests.
```python
job_config = {
    "version": "v1",
    "pipeline": "face_clustering",
    "resources": {"gpu": 1, "memory": "8Gi"}
}
with kafka.producer() as p:
    p.send("jobs.deploy", value=job_config, format="yaml")

@kafka.consumer(topic="jobs.deploy", format="yaml")
def handle_job(msg):
    # msg.value is deserialized back into a Python dict via PyYAML safe_load
    print(f"Pipeline: {msg.value['pipeline']}")
```

### 3. Images & Video Frames (`format="image"`)
Native integration with **OpenCV (`cv2`)**, **NumPy**, and **Pillow (`PIL.Image`)**. Produces compressed JPEG byte streams and deserializes directly into OpenCV BGR numpy arrays (`np.ndarray`) ready for deep learning models.
```python
import cv2
import numpy as np
from PIL import Image

# Produce an OpenCV BGR frame or Pillow Image
frame = cv2.imread("frame_001.jpg")  # np.ndarray
with kafka.producer() as p:
    # Optional 'quality' parameter (1-100, default: 80)
    p.send("camera.stream", value=frame, format="image", quality=90)

@kafka.consumer(topic="camera.stream", format="image")
def process_frame(msg):
    # msg.value is a decoded cv2 BGR numpy array ready for inference
    image_bgr = msg.value
    h, w, c = image_bgr.shape
    print(f"📸 Frame received: {w}x{h}, {c} channels")
```

### 4. Arbitrary Binary Files & Documents (`format="file"`)
Transmit entire PDF reports, ZIP archives, models, audio, or text files directly. You can pass:
- A local file system path string (e.g. `"/path/to/report.pdf"`).
- Raw `bytes`.
- A dictionary containing a `"content"` key with bytes.

```python
# Producer sends a PDF document directly by file path
with kafka.producer() as p:
    p.send("reports.generated", value="/tmp/quarterly_audit.pdf", format="file")

@kafka.consumer(topic="reports.generated", format="file")
def handle_pdf(msg):
    pdf_bytes = msg.value  # Raw binary bytes
    with open("/storage/received_audit.pdf", "wb") as f:
        f.write(pdf_bytes)
    print(f"📄 Saved PDF file ({len(pdf_bytes)} bytes)")
```

### 5. Pydantic Models (`format="pydantic"`)
Strict typed validation for both Pydantic v1 and v2. Guarantees that consumer handlers only receive valid, validated domain models.
```python
from pydantic import BaseModel, Field

class UserEvent(BaseModel):
    user_id: int
    email: str
    active: bool = True

# Producer sends Pydantic model instance
event = UserEvent(user_id=101, email="wisrovi@example.com")
with kafka.producer() as p:
    p.send("users.events", value=event, format="pydantic")

# Consumer automatically validates and parses back into the model class
@kafka.consumer(topic="users.events", format="pydantic", model=UserEvent)
def handle_user(msg):
    user: UserEvent = msg.value
    print(f"👤 User: {user.user_id}, Email: {user.email}")
```

---

## 📤 Producer Guide

### Context Manager & Safe Flushing
The `with kafka.producer() as p:` pattern guarantees that messages buffered in memory are completely flushed (`p.flush()`) to the Kafka broker and sockets are cleanly closed even if exceptions occur.

```python
with kafka.producer() as p:
    for i in range(10):
        p.send("metrics", value={"cpu": i * 10}, format="json")
# Guaranteed flush & close upon exiting the block
```

### Message Keys & Partition Routing
Provide `key="entity_id"` to ensure all events with the same key land on the same partition, guaranteeing strict chronological ordering.

```python
with kafka.producer() as p:
    p.send("sensor.data", key="sensor-alpha", value={"reading": 42.1}, format="json")
```

### Headers & Metadata
Pass rich metadata dictionaries via `headers`. WKafka automatically serializes dictionary and list headers into JSON bytes, and automatically parses them back on the receiving consumer end:

```python
headers = {
    "trace_id": "7b8f9e2d-3c4a-4e89",
    "tenant_id": 502,
    "routing_info": {"datacenter": "eu-central-1", "priority": "high"},
}

with kafka.producer() as p:
    p.send("orders", value={"order": 1}, headers=headers, format="json")
```

### High-Throughput Snappy Compression
If `python-snappy` is installed (`pip install "wkafka[snappy]"`), WKafka automatically enables Snappy compression for maximum throughput and minimal network bandwidth. If not installed, it gracefully falls back to `gzip` compression with an informative log.

---

## 📥 Consumer Guide

### 1. Manual Offset Commits (At-Least-Once Delivery)
By default, Kafka uses auto-commit. When you require strict At-Least-Once guarantees where an offset must only be marked as completed after successful database persistence, set `auto_commit=False` and call `msg.commit()`:

```python
@kafka.consumer(topic="payments", format="json", auto_commit=False)
def process_payment(msg):
    try:
        charge_credit_card(msg.value)
        record_in_database(msg.value)
        # Commit offset only after business operations succeed
        msg.commit()
    except DatabaseConnectionError:
        # Message is not committed; will be safely re-consumed
        raise
```

### 2. Automatic Retries & Dead Letter Queue (DLQ)
When external services fail temporarily, configure retries with exponential backoff. If all attempts fail, the message is automatically routed to a Dead Letter Queue topic with diagnostic headers:

```python
@kafka.consumer(
    topic="external.sync",
    format="json",
    max_retries=3,          # Retry up to 3 times
    retry_delay=1.0,        # Delays: 1s, 2s, 4s (exponential)
    dlq_topic="external.sync.DLQ"
)
def sync_handler(msg):
    call_flaky_third_party_api(msg.value)
```

Failed messages routed to `dlq_topic` automatically receive the following headers:
- `x-error-message`: Exact Python exception text that caused the final failure.
- `x-original-topic`: Source topic where the message originated.
- `x-original-offset`: Source partition offset.

### 3. Multi-Topic & Regex Pattern Subscriptions
Subscribe to multiple explicit topics or dynamically match topics using regular expressions:

```python
# Multiple explicit topics
@kafka.consumer(topic=["telemetry.eu", "telemetry.us"], format="json")
def handle_global_telemetry(msg):
    print(f"Data from {msg.topic}: {msg.value}")

# Dynamic Regex pattern matching
@kafka.consumer(pattern="iot\\..*\\.sensors", format="json")
def handle_dynamic_sensors(msg):
    print(f"Regex matched event on {msg.topic}: {msg.value}")
```

### 4. Partition Key Filtering
Filter incoming messages by partition key directly at the decorator level, ignoring unrelated keys without business logic overhead:

```python
@kafka.consumer(topic="logs", format="json", key_filter="CRITICAL")
def handle_critical_logs_only(msg):
    send_pagerduty_alert(msg.value)
```

### 5. Asynchronous Coroutines (`async / await`)
Handlers can be native Python coroutines. WKafka manages event loops and thread safety transparently:

```python
import aiohttp

@kafka.consumer(topic="webhooks.dispatch", format="json")
async def dispatch_webhook(msg):
    async with aiohttp.ClientSession() as session:
        async with session.post(msg.value["url"], json=msg.value["payload"]) as resp:
            print(f"Webhook response status: {resp.status}")
```

---

## ⚡ Long-Running Tasks & Decoupled Workers (The ML Dilemma)

### The Problem: Heartbeat Timeouts & Rebalance Storms
In Apache Kafka, if a consumer handler takes longer than `max_poll_interval_ms` (default: 5 minutes) to return:
1. The Kafka broker assumes the consumer has crashed or deadlocked.
2. The broker evicts the consumer from the consumer group (**Consumer Group Rebalance**).
3. The message offset is never committed, and another consumer picks up the same message in an **infinite duplicate loop**.

### The Solution: WKafka Decoupled Architecture
WKafka provides two native solutions that can be used independently or combined:

1. **`commit_on_receive=True`**: Immediately commits the partition offset to the broker the moment the message is dequeued, before running heavy processing logic.
2. **`background=True`**: Immediately submits the message to an internal `ThreadPoolExecutor`. The main Kafka polling thread instantly returns to its `poll()` loop, continually answering broker heartbeats. Heavy ML model inference, large video rendering, or PDF generation can run for hours without triggering a rebalance!
3. **`background_workers=N`**: Controls the maximum number of concurrent tasks executed in parallel.

```python
from wkafka import WKafka
import time

kafka = WKafka(bootstrap_servers="localhost:9092")

# Long-running ML inference or heavy document generation
@kafka.consumer(
    topic="ai.document.ocr",
    format="json",
    commit_on_receive=True,  # 1. Acknowledge offset immediately upon receipt
    background=True,         # 2. Offload to background worker pool
    background_workers=4,    # 3. Allow up to 4 parallel heavy jobs
)
def heavy_ocr_pipeline(msg):
    # Runs in an isolated thread.
    # The Kafka consumer poll loop remains active, sending heartbeats continuously!
    print(f"🚀 Processing document {msg.value['doc_id']}...")
    time.sleep(600)  # Simulating a 10-minute AI task
    print(f"✅ Finished document {msg.value['doc_id']}")

kafka.run_consumers(block=True)
```

### Architecture Diagram
```
Kafka Topic ────────► KafkaConsumer Loop Thread (poll)
                            │
                            ├─► commit_on_receive=True ──► [Immediate Broker Offset Commit]
                            │
                            └─► background=True ─────────► [ThreadPoolExecutor]
                                                               ├─► Worker Thread 1 (Heavy ML / PDF)
                                                               ├─► Worker Thread 2 (Heavy ML / PDF)
                                                               └─► Polling Thread Continues (Heartbeats Active!)
```

---

## 🚀 Dynamic Partition Auto-Scaling (`partition_scale=True`)

In Apache Kafka, the number of parallel consumers in a consumer group cannot exceed the number of partitions on the topic. If a topic has 1 partition and you run 4 consumer threads, 3 threads remain completely idle.

When `partition_scale=True` is enabled in `WKafka()` or `run_consumers(partition_scale=True)`:
1. WKafka counts the number of registered consumer handlers for each topic.
2. It queries the broker via `KafkaAdminClient.describe_topics()`.
3. If the topic exists with fewer partitions than required, it dynamically scales the partition count via `KafkaAdminClient.create_partitions()`.
4. If the topic does not exist, it automatically creates it with the desired partition count.
5. All worker threads can now consume partitions in true parallel fashion!

```python
kafka = WKafka(bootstrap_servers="localhost:9092", partition_scale=True)
```

---

## 🚀 High-Throughput RAM Payload Offloading (Claim-Check Pattern with WRedis)

### The Vision & Large-File Bottleneck
Streaming high-frequency multimedia (e.g. 30 FPS at 1080p OpenCV matrices) or multi-megabyte binary documents directly through Apache Kafka brokers leads to severe problems:
1. **Broker Memory & Disk Saturation:** Heavy messages bloat the JVM heap, trigger garbage collection stalls, and monopolize disk pagecaches.
2. **Commit Latency & Rebalances:** Transporting heavy binaries slows down consumer polling and offset commits.
3. **Stale Frames Accumulation:** If a downstream consumer lags during deep learning inference (e.g., YOLO or segmentation), Kafka queues up gigabytes of outdated frames instead of maintaining real-time video state.

### The Solution: Transparent Claim-Check via WRedis
WKafka natively implements the **Claim-Check Pattern** powered by `WRedis`:
1. **Producer Side:** Heavy payloads (e.g. `format="image"` or `format="file"`) are compressed and written directly to Redis in RAM with a short Time-To-Live (TTL = 5 seconds).
2. **Kafka Broker:** Transfers only an ultra-lightweight JSON Claim-Check pointer (~120 bytes) containing headers and routing keys.
3. **Consumer Side:** Intercepts the Claim-Check reference, retrieves the binary payload directly from Redis RAM, and decodes it back to the original object (`numpy.ndarray` for images) transparently.
4. **Drop-on-Expired Policy:** If a consumer lags beyond the TTL (5s), stale frames evaporate from RAM and are skipped automatically without crashing (`on_expired="drop"`), keeping your pipeline strictly real-time!

```mermaid
sequenceDiagram
    autonumber
    actor P as Producer (Camera Extractor)
    participant WK as WKafka Engine
    participant R as Redis RAM (WRedis)
    participant K as Kafka Broker
    actor C as Consumer (Vision AI / YOLO)

    P->>WK: send("camera-stream", frame, format="image")
    Note over WK: Serializes frame to JPEG bytes.<br/>Detects offload active
    WK->>R: SETEX("wkafka:claim:<uuid>", 5s, raw_bytes)
    WK->>K: send(Topic, Header & JSON pointer: ~166 bytes)
    Note over K: Broker handles sub-millisecond routing
    K->>WK: raw_message
    Note over WK: Intercepts x-wkafka-offload header
    WK->>R: GET("wkafka:claim:<uuid>")
    alt Frame alive in RAM (within 5s TTL)
        R-->>WK: raw_bytes
        Note over WK: Deserializes to np.ndarray
        WK->>C: callback(msg) with original frame
    else Frame expired (> 5s consumer lag)
        Note over WK: on_expired="drop": Drops stale frame safely
        WK-->>C: Skips callback, preserves 0 lag
    end
```

### 📊 Empirical Performance Benchmark: HD 1080p Video Transmission

To quantify real-world efficiency gains, an empirical benchmark was conducted running Apache Kafka (`apache/kafka:latest` in KRaft mode) and Redis (`redis:7.2-alpine`) via Docker Compose. The benchmark measured the transmission of **Full HD 1080p frames** (1920×1080 RGB `numpy.ndarray`, 6.22 MB uncompressed raw buffer) across standard Direct Kafka mode versus WKafka RAM Offload mode (`WRedis` Claim-Check).

#### Comparative Performance Table

| Performance Metric | Traditional Direct Kafka | WKafka RAM Offload (WRedis) | Improvement / Architectural Impact |
| :--- | :--- | :--- | :--- |
| **Broker Wire Payload** | **106.0 KB** | **166 bytes** | **654× smaller payload** (99.85% reduction) |
| **Broker Disk Usage (1 Hr @ 30 FPS)** | **10.92 GB** | **17.1 MB** | **99.8% broker disk savings** per camera |
| **Mean End-to-End Latency** | **24.89 ms** | **16.49 ms** | **34% faster round-trip processing** |
| **P95 End-to-End Latency** | **35.83 ms** | **27.84 ms** | **22% lower tail latency** |
| **Producer Send Latency (Mean)** | 7.80 ms | 9.07 ms | Sub-10ms with concurrent Redis RAM write |
| **Producer Send Latency (P95)** | 5.07 ms | 5.04 ms | Highly predictable burst ingestion |
| **Broker Ingress (100 Cams @ 30 FPS)** | **~318 MB/s** *(Network Bottleneck)* | **~498 KB/s** *(Negligible)* | **650× bandwidth reduction** across Kafka cluster |
| **Consumer Lagging Behavior** | Broker disk fills up with stale frames | Stale frames auto-expire in RAM (TTL) | Zero broker crashes; strictly real-time inference |
| **Developer Experience** | Manual multi-step serialization | **100% transparent**: Receives native `np.ndarray` | Zero code changes to existing consumer callbacks |

> [!TIP]
> **Key Architectural Takeaway:** By decoupling routing metadata (~166 bytes) from heavy multimedia buffers stored in volatile RAM, Kafka brokers operate at pure wire speed without GC pauses or disk I/O bottlenecks. Consumers lagging behind automatically skip obsolete frames due to Redis TTL expiration (`on_expired="drop"`), ensuring computer vision models always process the latest frame.

To reproduce these benchmarks locally using Docker Compose:
```bash
# 1. Spin up the benchmark environment (Kafka in KRaft mode + Redis)
docker compose -f enviroment/docker-compose.yml up -d

# 2. Run the automated benchmark script
poetry run python benchmark_offload.py
```

### Flexible Configuration (Options A, B, C, & D)

#### Option A: Load from Full Application YAML (`config.yaml`)
Ideal for Kubernetes ConfigMaps where infrastructure settings are updated without altering Python code:
```yaml
kafka:
  bootstrap_servers: "192.168.1.68:9092"
  client_id: "vision-pipeline-extended"
  offload:
    backend: "redis"
    host: "192.168.1.68"
    port: 15963
    password: null
    db: 0
    ttl: 5
    on_expired: "drop"
    formats:
      - "image"
      - "file"
    threshold_kb: 50
    key_prefix: "wkafka:claim:camera_1"
```

In Python:
```python
kafka = WKafka(config="config.yaml")
# Or directly as first positional argument:
kafka = WKafka("config.yaml")
```

#### Option B: Load Offload Block from Dedicated YAML
```python
kafka = WKafka(
    bootstrap_servers="localhost:9092",
    offload="offload_config.yaml"
)
```

#### Option C: In-Memory Dictionary (Minimal or Extended)
```python
# Minimal configuration (uses defaults and localhost:6379 / REDIS_HOST)
kafka = WKafka(
    bootstrap_servers="localhost:9092",
    offload={"backend": "redis", "ttl": 5}
)

# Extended configuration with custom network host/port and drop policy
kafka = WKafka(
    bootstrap_servers="localhost:9092",
    offload={
        "backend": "redis",
        "host": "192.168.1.68",
        "port": 15963,
        "password": "<your_redis_password>",
        "db": 0,
        "ttl": 5,
        "on_expired": "drop",
        "formats": ["image", "file"],
        "threshold_kb": 50,
        "key_prefix": "wkafka:claim:cam1",
    }
)
```

#### Option D: Environment Variables (Docker & Kubernetes)
```bash
export REDIS_HOST="192.168.1.68"
export REDIS_PORT="15963"
```
```python
kafka = WKafka(offload={"backend": "redis", "ttl": 5})
```

### Producer & Consumer Implementation (100% Transparent)

```python
# 1. Producer (Camera Extractor)
kafka = WKafka(bootstrap_servers="localhost:9092", offload={"backend": "redis", "ttl": 5})

# Send as usual - offloading is transparent:
kafka.send("camera-stream", value=frame, format="image")

# 2. Consumer (Inference / Analytics)
@kafka.consumer("camera-stream", format="image")
def process_frame(msg):
    frame = msg.value  # Seamlessly receives the numpy.ndarray from Redis RAM
```

---

## 🔒 Enterprise Security & Proxies

### SASL Authentication (PLAIN / SCRAM)
WKafka seamlessly passes authentication credentials down to `KafkaConsumer`, `KafkaProducer`, and `KafkaAdminClient`:

```python
sasl_config = {
    "security_protocol": "SASL_PLAINTEXT",  # or SASL_SSL
    "sasl_mechanism": "SCRAM-SHA-512",      # PLAIN, SCRAM-SHA-256, SCRAM-SHA-512
    "sasl_plain_username": "enterprise-service",
    "sasl_plain_password": "super-secure-password",
}

kafka = WKafka(
    bootstrap_servers="kafka-secure.internal:9093",
    **sasl_config
)
```

### SOCKS5 Proxy Normalization
If operating behind corporate firewalls or SOCKS proxies, WKafka automatically normalizes deprecated proxy settings (`socks5_proxy` to `proxy_url`) without triggering client warnings.

---

## 📋 Comprehensive API Reference

### 1. `WKafka` Constructor Reference
```python
WKafka(bootstrap_servers=None, client_id=None, dynamic_group_id=False, partition_scale=False, **extra_config)
```

| Parameter | Type | Default | Description |
| :--- | :--- | :--- | :--- |
| `bootstrap_servers` | `str \| List[str]` | `None` | Kafka broker address. Falls back to `KAFKA_SERVER` env var or `"localhost:9092"`. |
| `client_id` | `str` | `"wkafka-client"` | Identifier passed to Kafka broker for logging and tracing. |
| `dynamic_group_id` | `bool` | `False` | Generates a random unique `group_id` per run (useful for fan-out pub/sub). |
| `partition_scale` | `bool` | `False` | Automatically scales topic partitions to match registered consumer count. |
| `**extra_config` | `Any` | - | Additional low-level settings forwarded to underlying `KafkaConsumer` and `KafkaProducer`. |

### 2. `@kafka.consumer(...)` Decorator Reference
```python
@kafka.consumer(topic=None, pattern=None, group_id=None, format="json", auto_commit=True, commit_on_receive=False, background=False, background_workers=1, max_retries=0, retry_delay=1.0, dlq_topic=None, model=None, key_filter=None, **kafka_kwargs)
```

| Parameter | Type | Default | Description |
| :--- | :--- | :--- | :--- |
| `topic` | `str \| List[str]` | `None` | Single topic name or list of topic names to subscribe to. |
| `pattern` | `str` | `None` | Regular expression string for dynamic topic pattern matching. |
| `group_id` | `str` | Auto-generated | Consumer group identifier. |
| `format` | `str` | `"json"` | Deserializer: `"json"`, `"yaml"`, `"image"`, `"file"`, `"pydantic"`. |
| `auto_commit` | `bool` | `True` | Standard periodic background offset commits by Kafka. |
| `commit_on_receive` | `bool` | `False` | Commits offset immediately upon receipt before executing handler. |
| `background` | `bool` | `False` | Offloads handler execution to a decoupled thread pool to keep polling loop alive. |
| `background_workers` | `int` | `1` | Max concurrent worker threads when `background=True`. |
| `max_retries` | `int` | `0` | Number of retry attempts on exception before routing to DLQ. |
| `retry_delay` | `float` | `1.0` | Initial delay in seconds for exponential backoff retries. |
| `dlq_topic` | `str` | `None` | Dead Letter Queue topic to route permanently failed messages. |
| `model` | `BaseModel` | `None` | Pydantic model class for typed validation when `format="pydantic"`. |
| `key_filter` | `str` | `None` | Only executes handler for messages matching this exact key. |

### 3. Producer Reference (`kafka.send(...)`)
```python
kafka.send(topic, value, key=None, format="json", headers=None, **kwargs)
```

| Parameter | Type | Default | Description |
| :--- | :--- | :--- | :--- |
| `topic` | `str` | Required | Target topic to publish message to. |
| `value` | `Any` | Required | Payload to serialize and send. |
| `key` | `str` | `None` | Partition routing key. |
| `format` | `str` | `"json"` | Target serializer: `"json"`, `"yaml"`, `"image"`, `"file"`, `"pydantic"`. |
| `headers` | `Dict[str, Any]` | `None` | Key-value dictionary of metadata headers (auto-serialized to JSON). |
| `**kwargs` | `Any` | - | Serializer options (e.g. `quality=90` for `format="image"`). |

### 4. `Message` Object Reference
Received by consumer callback functions:

| Attribute / Method | Type | Description |
| :--- | :--- | :--- |
| `msg.value` | `Any` | Deserialized message content (dict, ndarray, bytes, Pydantic model). |
| `msg.topic` | `str` | Topic from which the message was consumed. |
| `msg.offset` | `int` | Partition offset number. |
| `msg.key` | `Optional[str]` | Message partition key (decoded as string if bytes). |
| `msg.headers` | `Dict[str, Any]` | Decoded headers and metadata dictionary. |
| `msg.group_id` | `str` | Consumer group identifier. |
| `msg.commit()` | `Callable` | Manually commits this message offset to the Kafka broker. |

---

## 🗂️ 15 Production-Ready Examples

WKafka includes a comprehensive catalog of 15 fully functional example modules in the `examples/` directory:

| Directory | Level | Architectural Concept |
| :--- | :--- | :--- |
| [`01_basic`](examples/01_basic) | 🟢 Easy | Basic JSON Producer and Consumer setup. |
| [`02_advanced`](examples/02_advanced) | 🟡 Intermediate | Custom metadata Headers and Partition routing Keys. |
| [`03_multimedia`](examples/03_multimedia) | 🟠 Advanced | Continuous streaming of OpenCV / NumPy image frames. |
| [`04_security_sasl`](examples/04_security_sasl) | 🔴 Pro | Enterprise security with SASL authentication (PLAIN / SCRAM). |
| [`05_patterns`](examples/05_patterns) | 🔥 Architect | Distributed microservices worker pattern. |
| [`06_images`](examples/06_images) | 🖼️ Multimedia | Dual support for Pillow (PIL) and OpenCV computer vision images. |
| [`07_files`](examples/07_files) | 📁 Documents | Binary file transfers (PDFs, ZIPs, TXTs) via `format="file"`. |
| [`08_manual_commit`](examples/08_manual_commit) | ⚙️ Resilience | Manual offset commits (`auto_commit=False` & `msg.commit()`). |
| [`09_retries_and_dlq`](examples/09_retries_and_dlq) | 🛡️ Resilience | Automatic retries with exponential backoff and DLQ routing. |
| [`10_multi_topic_regex`](examples/10_multi_topic_regex) | 🎯 Routing | Multi-topic subscription lists and Regex pattern matching. |
| [`11_async_handlers`](examples/11_async_handlers) | ⚡ Async | Non-blocking asynchronous handlers using `async / await`. |
| [`12_pydantic_validation`](examples/12_pydantic_validation) | 🛡️ Validation | Strict schema validation with Pydantic v1 & v2 models. |
| [`13_partition_scale`](examples/13_partition_scale) | 🚀 Scaling | Dynamic partition auto-scaling for true consumer parallelism. |
| [`14_interactive_producer`](examples/14_interactive_producer) | 🎮 Tooling | Interactive CLI console for publishing messages in real time. |
| [`15_long_running_background_tasks`](examples/15_long_running_background_tasks) | ⚡ Compute | Decoupled background workers preventing heartbeat timeouts. |
| [`16_offload_basic`](examples/16_offload_basic) | 🚀 Offload | High-throughput Claim-Check pattern: frames in Redis RAM with 5s TTL. |
| [`17_offload_extended`](examples/17_offload_extended) | 🌐 Enterprise | Production offloading via external `config.yaml` and custom network ports. |

---

## 🛠️ Key Technologies and Libraries

- **Python (3.9 - 3.14)**: Core programming language and execution runtime.
- **kafka-python-ng / kafka-python**: Low-level client driver for Apache Kafka and KRaft protocol communication.
- **WRedis (`wredis`)**: High-speed in-memory database and caching layer for ephemeral multimedia offloading.
- **OpenCV (`opencv-python`) & Pillow**: Computer vision matrix manipulation, image decoding, rendering, and JPEG compression.
- **NumPy**: Multidimensional arrays and tensor buffers for real-time video matrices.
- **Pydantic (v1 / v2)**: Declarative validation and strictly-typed data contracts.
- **PyYAML**: Safe parsing and serialization of declarative configuration manifests.
- **Loguru**: Structured logging framework for distributed event traceability.
- **Pytest & Coverage**: Automated unit test suites and comprehensive coverage reporting.
- **Docker**: Isolated sandboxing for consistent testing across environments.

---

## 🧪 Unit Testing & Coverage

The project includes an exhaustive unit test suite verifying serializers, retry policies, DLQ routing, manual offsets, decoupled execution, dynamic partition scaling, and Claim-Check payload offload.

### 1. Run Tests Locally (pytest)
```bash
poetry run pytest tests/ -v
```

### 2. Generate Coverage Report (Guaranteed >80%)
```bash
./run_coverage.sh
```
This script runs `pytest --cov=wkafka --cov-fail-under=80` and outputs a comprehensive HTML report in `htmlcov/index.html`. It currently guarantees over **88% code coverage**.

### 3. Run Tests in Isolated Docker Container
To ensure strict parity with CI/CD and production environments:
```bash
./run_tests_docker.sh
```
The script runs the full test suite inside an isolated `python:3.13-slim` container with all operating system and Python dependencies installed.

---

## 👤 Author & Affiliation

* **Author:** William Steve Rodriguez Villamizar (Wisrovi)
* **Title:** AI Leader & Solutions Architect at eCaptureDtech
* **Location:** Badajoz, Extremadura, Spain
* 🌐 **Official Website:** [wisrovi.dev](https://wisrovi.dev)
* 💼 **LinkedIn:** [wisrovi-rodriguez](https://www.linkedin.com/in/wisrovi-rodriguez/)
* 🆔 **ORCID:** [0009-0005-0710-1861](https://orcid.org/0009-0005-0710-1861)
* 📦 **PyPI:** [pypi.org/user/wisrovi/](https://pypi.org/user/wisrovi/)
* 🐙 **GitHub:** [@wisrovi](https://github.com/wisrovi)

---

## 📜 License

Distributed under the **MIT License**. Free for commercial, enterprise, and academic research use.

