csrd.message
Transport-agnostic message abstractions for publish/consume workflows, with optional RabbitMQ and Redis Streams adapters.
Why this package exists
- Provide a stable API for messaging behavior independent of broker choice
- Enable optional broker installs with extras (
rabbit,redis,kafka) - Keep service code portable across RabbitMQ, Redis Streams and Kafka backends
Installation
# Core abstractions only
uv add csrd-message
# RabbitMQ support
uv add 'csrd-message[rabbit]'
# Redis Streams support (redis-py >= 5, redis.asyncio)
uv add 'csrd-message[redis]'
# Kafka support (dependency only; no adapter yet)
uv add 'csrd-message[kafka]'
# Everything
uv add 'csrd-message[all]'
Important extras note
Python extras are part of the same PyPI project, not a separate project.
csrd-messageandcsrd-message[redis]resolve to the same package name (csrd-message)- The extra only adds optional dependencies declared by this package
- Extras do not create new package names on PyPI
Core API surface
Message- immutable message envelopeMessagePublisher- protocol for publishing messagesMessageConsumer- protocol for consuming messages (consume(message) -> Acknowledgement)Acknowledgement-ACK/RETRY/REJECTencode_message/decode_message(andmessage_to_dict/message_from_dict) - the canonical JSON envelope (see Wire encoding)
from csrd.message import Acknowledgement, Message, MessagePublisher
async def send(publisher: MessagePublisher) -> None:
await publisher.publish(Message(topic="orders.created", payload={"order_id": "o-1"}))
Redis Streams adapter (csrd-message[redis])
from csrd.message.redis import RedisConsumer, RedisLifespan, RedisPublisher, StreamConfig
At-least-once queueing over Redis Streams consumer groups. Use it when a
csrd service already runs Redis and needs a job/event queue without adding
RabbitMQ. See docs/MESSAGING_AND_WORKERS.md for how it relates to Celery.
Publisher
RedisPublisher(
redis, # redis.asyncio.Redis
*,
stream_prefix="", # stream key = f"{stream_prefix}{message.topic}"
maxlen=10_000, # XADD MAXLEN; None = never trim
approximate=True, # MAXLEN ~ (recommended)
)
await publisher.publish(message) # MessagePublisher protocol
entry_id = await publisher.publish_entry(message) # same, returns the stream entry id
Consumer
RedisConsumer(
redis, topic, group, consumer, # consumer: any MessageConsumer
*,
consumer_name=None, # default: socket.gethostname(); must be stable across restarts
stream_prefix="",
concurrency=10, # max messages processed at once
block_ms=5_000, # XREADGROUP BLOCK
batch_size=10, # COUNT for XREADGROUP / XAUTOCLAIM
claim_idle_ms=60_000, # XAUTOCLAIM min idle time (also the RETRY delay)
claim_interval_ms=30_000, # how often to XAUTOCLAIM; must be < claim_idle_ms
max_deliveries=5, # dead-letter after this many deliveries; None = unlimited
dead_letter_suffix=":dead",
dead_letter_maxlen=None,
group_start_id="0", # XGROUP CREATE id; "0" also consumes entries published before the group existed
shutdown_timeout=30.0, # seconds stop() waits for in-flight messages
)
await consumer.run()consumes untilawait consumer.stop()(use from a standalone worker process; callstop()from a signal handler).consumer.start()runs it as a background task;await consumer.stop()stops it.- Connection errors are retried with exponential backoff (0.5 s up to 30 s);
the group is recreated if it disappears (
NOGROUP).
Processing loop:
XGROUP CREATE <stream> <group> <group_start_id> MKSTREAM.BUSYGROUPis ignored.- Restart recovery:
XREADGROUP GROUP <group> <consumer_name> STREAMS <stream> 0pages through this consumer's own pending entries and dispatches them again. - Every
claim_interval_ms:XCLAIM ... JUSTIDthe in-flight ids to self. This resets their idle time, so other consumers never steal a message that is still being processed. ThenXAUTOCLAIMentries idle longer thanclaim_idle_ms: those left by crashed consumers and those that gotRETRY. XREADGROUP ... BLOCK <block_ms> COUNT <n> STREAMS <stream> >, wherenis never more than the free concurrency slots.
Entry ids already in flight in this process are never dispatched a second time, even when a reclaim returns them.
Acknowledgement mapping
Outcome of consume(message) |
Redis action |
|---|---|
Acknowledgement.ACK |
XACK |
Acknowledgement.REJECT |
XADD <stream><dead_letter_suffix>, then XACK |
Acknowledgement.RETRY |
none; entry stays pending, re-delivered by XAUTOCLAIM after claim_idle_ms |
| raises an exception | logged, treated as RETRY |
delivery count > max_deliveries |
dead-lettered (reason=max_deliveries) without calling consume |
entry is not valid JSON or lacks the message field |
dead-lettered immediately (reason=undecodable) |
| entry was trimmed while pending | XACK (nothing to process) |
cancelled at shutdown (after shutdown_timeout) |
stays pending, recovered on restart |
The exception and poison handling mirrors the Rabbit adapter. There, a handler error is dead-lettered to a TTL queue and re-delivered, and a message rejected 5 or more times is dropped. The Redis adapter dead-letters it rather than dropping it.
Delivery is at least once: consume must be idempotent, keyed for
example on message.message_id.
Lifespan (FastAPI)
from csrd.lifespan import lifespan_stack
from csrd.message.redis import RedisLifespan, StreamConfig
redis_lifespan = RedisLifespan(
"redis://:password@redis:6379/0", # or client=existing_redis (not closed on shutdown)
streams=[StreamConfig(topic="jobs.render", group="render-workers", consumer=RenderConsumer())],
stream_prefix="",
maxlen=10_000,
consumer_name=None, # default for StreamConfig.consumer_name
state_key="redis_streams",
)
app = FastAPI(lifespan=lifespan_stack(db_lifespan, redis_lifespan))
# in a dependency:
state = request.app.state.redis_streams # RedisStreamsState
await state.publisher.publish(Message(topic="jobs.render", payload={...}))
StreamConfig has the same fields and defaults as the RedisConsumer keyword
arguments. If you omit streams, the lifespan is publish-only. The state is
stored under redis_streams, not redis, so it does not clash with a
service's own app.state.redis client. On shutdown every consumer is stopped
gracefully, and the client is closed only if the lifespan created it.
Wire encoding
This section is the cross-language contract.
- Stream key:
<stream_prefix><topic>. With no prefix, the stream key is the topic itself, for examplejobs.render. - Entry fields: exactly one field,
message, whose value is a UTF-8 JSON object (the envelope). Entries without it are dead-lettered. - Consumer groups: a normal Redis consumer group. Any language may join the same group or read the stream with its own group.
- Dead-letter stream:
<stream><dead_letter_suffix>(defaultjobs.render:dead). Its fields aremessage(the original envelope, unchanged),source_stream,source_id(the original entry id),group,consumer,reason(rejected|max_deliveries|undecodable) and, when known,deliveries.
Envelope (JSON Schema):
{
"$schema": "https://json-schema.org/draft/2020-12/schema",
"title": "csrd.message.Message",
"type": "object",
"required": ["topic", "payload"],
"properties": {
"topic": {"type": "string", "minLength": 1},
"payload": {"type": "object"},
"key": {"type": ["string", "null"], "default": null},
"headers": {"type": "object", "additionalProperties": {"type": "string"}, "default": {}},
"message_id": {"type": ["string", "null"], "default": null},
"timestamp": {"type": "string", "format": "date-time",
"description": "ISO-8601; producers SHOULD include a UTC offset. Naive values are read as UTC. Defaults to the time of decoding."}
}
}
Example entry, as created by XADD jobs.render MAXLEN ~ 10000 * message '<json>':
{"topic":"jobs.render","payload":{"card_id":"c-1"},"key":null,"headers":{"trace-id":"4bf9"},"message_id":"9a3e...","timestamp":"2026-10-16T12:00:00.123456+00:00"}
The Python encoder writes compact JSON and refuses payloads that the standard
json module cannot serialise. Convert datetimes, UUIDs and similar values to
strings yourself. topic may be omitted by foreign producers: the consumer
then uses the topic it is subscribed to. Unknown envelope keys are ignored.
Observability
The Redis adapter logs through the standard library, like the Rabbit adapter,
using the logger names csrd.message.redis.*. Records carry extra fields
stream, entry_id, group and, once decoded, message_id, so
csrd-logging JSON formatters pick them up when that package is installed.
csrd-message does not import csrd-logging, csrd-context or
csrd-metrics. To bind context or count outcomes, wrap your
MessageConsumer.
RabbitMQ adapter (csrd-message[rabbit])
from csrd.message.rabbit import RabbitPublisher, RabbitConsumer, RabbitMessageHandler, RabbitLifespan, QueueConfig, AMQPState
See docs/PACKAGE_CAPABILITIES.md and docs/MESSAGING_AND_WORKERS.md.
1 maintainer
Metadata
1 maintainer
Release files for csrd-message 0.8.0
For a detailed explanation of source distributions (sdists) and built distributions (wheels), please see the package formats documentation.
Source distribution (sdist)
| File | Size | Uploaded | |
|---|---|---|---|
| csrd_message-0.8.0.tar.gz | 27.7 kB | Details |
Built distribution (wheel)
| File | Interpreter | ABI | Platform | Reset |
|---|---|---|---|---|
| csrd_message-0.8.0-py3-none-any.whl | Python 3 | none | any | Details |
Total release size: 55.2 kB
Release files / csrd_message-0.8.0.tar.gz
| Download URL | csrd_message-0.8.0.tar.gz |
|---|---|
| Size | 27.7 kB |
| Tags | Source |
|
SHA-256 checksum How to use checksums |
0977063444ca36b6c032ba35e0e9e6e669d50a542a923c84ba589adc1875dea6
|
|
BLAKE2b-256 checksum How to use checksums |
3a9ef182df3f05d6d85504ea73b7633a1369f17c0d105e5e4a3c465b8a391777
|
| Upload date | |
|
Uploaded using Trusted Publishing? What is trusted publishing? |
Yes |
| Uploaded via |
twine/7.0.0 CPython/3.13.14
|
Provenance
Provenance describes where a file came from. On PyPI, provenance is shared via attestations, which provide a verifiable record of the build or publishing details. View details, limitations and caveats.
PyPI Publish Attestation
PyPI verified that this artifact, at this checksum, originated from the publisher listed below.
Signed by GitHub Actions, verified by PyPI on Oct 16, 2026.
Transparency logRelease files / csrd_message-0.8.0-py3-none-any.whl
| Download URL | csrd_message-0.8.0-py3-none-any.whl |
|---|---|
| Size | 27.5 kB |
| Tags | Python 3 |
|
SHA-256 checksum How to use checksums |
a4797a543e07a52a6f7e60da4b15288638ebcf3ae6292024b63cff504ba3a3ec
|
|
BLAKE2b-256 checksum How to use checksums |
ab55949899d9f01f15feaabb0e3cb68645ada6e8c40db148af6f00ca6e9d3b54
|
| Upload date | |
|
Uploaded using Trusted Publishing? What is trusted publishing? |
Yes |
| Uploaded via |
twine/7.0.0 CPython/3.13.14
|
Provenance
Provenance describes where a file came from. On PyPI, provenance is shared via attestations, which provide a verifiable record of the build or publishing details. View details, limitations and caveats.
PyPI Publish Attestation
PyPI verified that this artifact, at this checksum, originated from the publisher listed below.
Signed by GitHub Actions, verified by PyPI on Oct 16, 2026.
Transparency log
