Skip to content

API

pico_kafka.decorators

Marker decorators for the two sides of Kafka messaging.

Consumer side: @kafka_consumer on a component method subscribes it to a topic; the method receives the JSON-decoded record value.

Producer side: @kafka_producer on a class of @produce stubs turns it into an injectable component whose methods send their message argument as a JSON value (same generated-implementation idiom as pico-httpx clients and pico-rabbitmq publishers).

kafka_consumer(topic, *, group_id='')

Subscribe a component method to topic.

group_id defaults to the kafka.group_id setting; give each independent projection its own group to fan the stream out.

Source code in src/pico_kafka/decorators.py 
22
23
24
25
26
27
28
29
30
31
32
33
34
35
def kafka_consumer(topic: str, *, group_id: str = ""):
    """Subscribe a component method to ``topic``.

    ``group_id`` defaults to the ``kafka.group_id`` setting; give each
    independent projection its own group to fan the stream out.
    """
    if not topic:
        raise ValueError("@kafka_consumer requires a topic")

    def dec(fn):
        setattr(fn, CONSUMER_META, {"topic": topic, "group_id": group_id})
        return fn

    return dec

kafka_producer(cls)

Turn a class of @produce stubs into an injectable component.

Source code in src/pico_kafka/decorators.py 
38
39
40
41
42
43
44
45
46
47
48
49
50
def kafka_producer(cls):
    """Turn a class of ``@produce`` stubs into an injectable component."""
    setattr(cls, PRODUCER_META, True)
    if "__init__" not in cls.__dict__:

        def __init__(self, registrar):
            self._pico_kafka = registrar

        from .registrar import KafkaRegistrar

        __init__.__annotations__ = {"registrar": KafkaRegistrar}
        cls.__init__ = __init__
    return component(cls)

produce(topic)

Generate a producing implementation for a stub method.

The stub's message argument (any JSON-serializable value) is the record value. Sync stubs block until the broker acks; async stubs await it.

Source code in src/pico_kafka/decorators.py 
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
def produce(topic: str):
    """Generate a producing implementation for a stub method.

    The stub's ``message`` argument (any JSON-serializable value) is the
    record value. Sync stubs block until the broker acks; async stubs
    await it.
    """
    if not topic:
        raise ValueError("@produce requires a topic")

    def dec(fn):
        if inspect.iscoroutinefunction(fn):

            @functools.wraps(fn)
            async def async_impl(self, message):
                import asyncio

                future = self._pico_kafka.produce(topic, message)
                return await asyncio.wrap_future(future)

            setattr(async_impl, PRODUCE_META, {"topic": topic})
            return async_impl

        @functools.wraps(fn)
        def sync_impl(self, message):
            future = self._pico_kafka.produce(topic, message)
            return future.result(timeout=self._pico_kafka.produce_timeout)

        setattr(sync_impl, PRODUCE_META, {"topic": topic})
        return sync_impl

    return dec

pico_kafka.registrar

Connection owner and dispatcher.

Runs a dedicated asyncio loop in a daemon thread (same architecture as pico-rabbitmq): consumers and producers work in any app without lifespan wiring. Each @kafka_consumer method gets its own AIOKafkaConsumer task; records resolve their component through the container per message.

Failure policy: a consumer method that raises is logged and its record is skipped — offsets keep advancing so a poison record cannot stall the partition. Reprocess from your monitoring, not from a hot loop.

pico_kafka.config

Settings for pico-kafka (prefix kafka, zero-config).

KafkaSettings dataclass

enabled: false disables consumers and producers entirely.

Source code in src/pico_kafka/config.py 
 8
 9
10
11
12
13
14
15
16
@configured(target="self", prefix="kafka", mapping="tree")
@dataclass
class KafkaSettings:
    """``enabled: false`` disables consumers and producers entirely."""

    bootstrap_servers: str = "localhost:9092"
    enabled: bool = True
    group_id: str = "pico"
    produce_timeout_seconds: float = 10.0