Skip to content

API

pico_rabbitmq.decorators

Marker decorators for the two sides of RabbitMQ messaging.

Consumer side: @consumer on a component method subscribes it to a queue; the method receives the JSON-decoded message body.

Publisher side: @publisher on a class of @publish stubs turns it into an injectable component whose methods send their message argument as a JSON body (same generated-implementation idiom as pico-httpx clients).

consumer(queue, *, exchange='', routing_key='', exchange_type='topic')

Subscribe a component method to queue.

With exchange empty the queue receives messages published to the default exchange under its own name. Otherwise the exchange is declared (exchange_type) and the queue bound with routing_key.

Source code in src/pico_rabbitmq/decorators.py 
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
def consumer(queue: str, *, exchange: str = "", routing_key: str = "", exchange_type: str = "topic"):
    """Subscribe a component method to ``queue``.

    With ``exchange`` empty the queue receives messages published to the
    default exchange under its own name. Otherwise the exchange is
    declared (``exchange_type``) and the queue bound with ``routing_key``.
    """
    if not queue:
        raise ValueError("@consumer requires a queue name")

    def dec(fn):
        setattr(
            fn,
            CONSUMER_META,
            {"queue": queue, "exchange": exchange, "routing_key": routing_key, "exchange_type": exchange_type},
        )
        return fn

    return dec

publisher(cls)

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

Source code in src/pico_rabbitmq/decorators.py 
43
44
45
46
47
48
49
50
51
52
53
54
55
def publisher(cls):
    """Turn a class of ``@publish`` stubs into an injectable component."""
    setattr(cls, PUBLISHER_META, True)
    if "__init__" not in cls.__dict__:

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

        from .registrar import RabbitRegistrar

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

publish(*, exchange='', routing_key)

Generate a publishing implementation for a stub method.

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

Source code in src/pico_rabbitmq/decorators.py 
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
85
86
87
88
def publish(*, exchange: str = "", routing_key: str):
    """Generate a publishing implementation for a stub method.

    The stub's ``message`` argument (any JSON-serializable value) is the
    body. Sync stubs block until the broker confirms; async stubs await it.
    """
    if not routing_key and not exchange:
        raise ValueError("@publish requires a routing_key (or an exchange)")

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

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

                future = self._pico_rabbitmq.publish(exchange, routing_key, message)
                return await asyncio.wrap_future(future)

            setattr(async_impl, PUBLISH_META, {"exchange": exchange, "routing_key": routing_key})
            return async_impl

        @functools.wraps(fn)
        def sync_impl(self, message):
            future = self._pico_rabbitmq.publish(exchange, routing_key, message)
            return future.result(timeout=self._pico_rabbitmq.publish_timeout)

        setattr(sync_impl, PUBLISH_META, {"exchange": exchange, "routing_key": routing_key})
        return sync_impl

    return dec

pico_rabbitmq.registrar

Connection owner and dispatcher.

Runs a dedicated asyncio loop in a daemon thread so consumers and publishers work in any app — sync scripts, FastAPI, celery workers — without lifespan wiring. Consumers resolve their component through the container per message (prototype scope = fresh instance per message).

Delivery semantics: ack on success; on exception the message is rejected WITHOUT requeue (log + drop) so a poison message cannot spin the consumer. Configure a dead-letter exchange on the queue if you need to keep failures.

pico_rabbitmq.config

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

RabbitSettings dataclass

enabled: false disables consumers and publishers entirely.

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

    url: str = "amqp://guest:guest@localhost/"
    enabled: bool = True
    prefetch_count: int = 10
    publish_timeout_seconds: float = 10.0