WKafka v1.0.0 LTS 🚀

Professional, Decorator-based Kafka Wrapper for Python.

WKafka simplifies Apache Kafka integration by providing a high-level, intuitive API focused on developer productivity. It includes built-in support for complex data types like JSON, YAML, Images, Files, and Pydantic models, making it ideal for modern microservices, IoT, and Computer Vision pipelines.

🌟 Features

• Decorator-driven API: Minimalistic and clean message handling.
• Modern Python: Fully typed, PEP 8 compliant, supporting Python 3.9 through 3.14.
• Enterprise Security: Built-in support for SASL (PLAIN, SCRAM) and KRaft mode.
• Multimedia & File Native: Seamlessly send and receive images (OpenCV/NumPy/PIL) and arbitrary files (PDF, ZIP, TXT) via format="file".
• Type-safe Pydantic Validation: Automatic schema validation with format="pydantic".
• Manual Offset Commit: Control At-Least-Once delivery semantics with auto_commit=False and msg.commit().
• Immediate Commit on Receive: Built-in support for At-Most-Once / early acknowledgment with commit_on_receive=True.
• Decoupled Background Execution: Non-blocking background worker dispatch with background=True and background_workers=N to prevent Kafka heartbeat timeouts during long-running tasks.
• Partition Auto-scaling & Retries: Automatic topic partition scaling with describe_topics inspection and exponential backoff retry resilience against transient NodeNotReadyError.
• Async/Await Support: Define non-blocking async def consumer handlers.
• Professional Ops: Structured logging via loguru and multi-version testing with tox.

📦 Installation

# Via pip
pip install wkafka

# Via poetry
poetry add wkafka

Optional snappy compression:

pip install wkafka[snappy]

🚀 Quick Start

Basic Producer & Consumer

from wkafka import WKafka

# Configures automatically via KAFKA_SERVER or defaults to localhost:9092
kafka = WKafka(bootstrap_servers="localhost:9092")

@kafka.consumer(topic="orders", format="json")
def handle_order(msg):
    print(f"New order received: {msg.value}")

# Start consumers in a background thread pool
kafka.run_consumers(block=True)

# Produce with context manager safety
with kafka.producer() as p:
    p.send("orders", value={"id": 123, "item": "Coffee"}, format="json")

Manual Offset Commit

@kafka.consumer(topic="transactions", format="json", auto_commit=False)
def handle_tx(msg):
    # Process business logic
    save_to_db(msg.value)
    # Explicitly commit offset only after success
    msg.commit()

Retries & DLQ Routing

@kafka.consumer(
    topic="unstable_events",
    format="json",
    max_retries=3,
    retry_delay=1.0,
    dlq_topic="unstable_events.DLQ"
)
def handle_event(msg):
    process_payload(msg.value)

Immediate Commit on Receive (Early Acknowledgment)

@kafka.consumer(topic="events", format="json", commit_on_receive=True)
def handle_early_ack(msg):
    # Offset is committed immediately upon receipt before entering this function
    process_quick_event(msg.value)

Long-Running Tasks & Decoupled Background Workers

# Perfect for ML inference, video processing, or PDF generation that takes hours
@kafka.consumer(
    topic="reports.generate",
    format="json",
    commit_on_receive=True,  # Commit offset immediately upon receipt
    background=True,         # Offload execution to a background ThreadPool
    background_workers=4,    # Up to 4 parallel background workers
)
def handle_long_task(msg):
    # Runs in a separate thread. Kafka consumer loop never blocks and never triggers max_poll_interval_ms rebalances!
    generate_heavy_report_and_pdf(msg.value)

📂 Project Structure

• wkafka.core: Orchestration and base logic (WKafka, Message).
• wkafka.serializers: Extensible serialization system (JSONSerializer, YAMLSerializer, ImageSerializer, PydanticSerializer, FileSerializer).
• wkafka.controller: Backward compatibility layer for legacy code.
• examples/: 12 complete, production-ready example modules (01_basic through 12_pydantic_validation).
• environment/: Production-ready Docker setups (KRaft, SASL).

🛠️ Tecnologías y Librerías Relevantes

• Python (3.9 - 3.14): Lenguaje principal de desarrollo y ejecución.
• kafka-python-ng / kafka-python: Client library for low-level communication with Apache Kafka.
• OpenCV (opencv-python) & Pillow: Procesamiento, renderizado, serialización y deserialización de imágenes.
• Pydantic: Validación de esquemas y modelos de datos tipados (format="pydantic").
• NumPy: Manejo de estructuras de datos matriciales multidimensionales para imágenes.
• PyYAML: Serialización y deserialización nativa de estructuras YAML.
• Loguru: Sistema avanzado de logging estructurado.

🧪 Unit Testing & Coverage

To run the unit test suite, calculate coverage, or execute tests inside a sandboxed Docker container:

Run Locally (pytest)

poetry run pytest tests/

Coverage Report (80% Minimum Guaranteed)

./run_coverage.sh

El script genera un informe de cobertura HTML (htmlcov/index.html) garantizando una cobertura mínima global de 80%.

Sandboxed Testing with Docker

./run_tests_docker.sh

👤 Autor & Afiliación Oficial

• William Steve Rodriguez Villamizar (Wisrovi)
• Cargo: Principal AI Engineer & Applied AI Solutions Architect | Scientific Researcher
• 🌐 Portal Oficial: wisrovi.dev
• 💼 LinkedIn: wisrovi-rodriguez
• 🆔 ORCID: 0009-0005-0710-1861
• 📦 PyPI: pypi.org/user/wisrovi/
• 🐙 GitHub: @wisrovi

📜 License

Distributed under the MIT License. Open for industrial and research use.