pico-kafka 0.1.0

Kafka for the Pico ecosystem: @kafka_consumer methods and @kafka_producer clients over aiokafka.

pico-kafka

Kafka for the pico ecosystem: @kafka_consumer methods and declarative @kafka_producer clients over aiokafka.

Installation

pip install pico-kafka

Quick start

kafka:
  bootstrap_servers: kafka.internal:9092
  group_id: myapp

Consume — a component method per topic, JSON value decoded for you:

from pico_ioc import component
from pico_kafka import kafka_consumer

@component
class OrderProjection:
    @kafka_consumer("orders")
    async def on_order(self, message: dict):
        ...

    @kafka_consumer("orders", group_id="analytics")   # independent fan-out
    def on_order_analytics(self, message: dict):
        ...

Produce — stubs, like a pico-httpx client:

from pico_kafka import kafka_producer, produce

@kafka_producer
class OrderEvents:
    @produce("orders")
    def order_created(self, message): ...

Semantics:

• Consumers and producers run on a dedicated background loop — works in sync scripts, FastAPI apps and workers alike, no lifespan wiring.
• Each record resolves its component through the container (prototype scope = fresh instance per record).
• A record whose handler raises is logged and skipped — offsets advance, a poison record cannot stall the partition.
• Sync produce stubs block until the broker acks; async stubs await it. Everything stops with the container.

Documentation

Full documentation: https://dperezcabrera.github.io/pico-kafka/

License

MIT