DEV Community

William Rodriguez
William Rodriguez

Posted on

Reliable Event Ingestion in Python: Redis Streams and Consumer Groups with WRedis

#streaming

When your application outgrows fire-and-forget Redis Pub/Sub, you need message persistence, consumer offsets, and distributed worker balancing without spinning up a heavy broker.

WRedis v1.0.0 LTS provides first-class primitives for Redis Streams (XADD, XREADGROUP) with declarative consumer decorators, automatic group creation, and clean shutdown handling.

This is Day 03 of the WRedis Open-Source Technical Series (MIT, Python 3.10+, 95%+ test coverage).

Why Redis Streams Over Pub/Sub?

  1. Persistent History: Messages are logged to disk rather than dropped if a worker is momentarily offline.
  2. Consumer Groups: Workloads are balanced across multiple worker processes without double processing.
  3. Pending Entries List (PEL): Failed or unacknowledged messages can be reclaimed and reprocessed reliably.

Production Implementation: Streams with @on_message

1. Producing Structured Events (add_to_stream) 

from wredis.streams import RedisStreamManager

stream_mgr = RedisStreamManager(host="localhost", port=6379)

# Produce event payload with automatic JSON serialization
msg_id = stream_mgr.add_to_stream(
    key="events:transactions",
    data={"tx_id": "TX-9941", "amount": 250.0, "currency": "USD"},
    ttl=86400  # Stream retention TTL
)
print(f"Dispatched event ID: {msg_id}")
Enter fullscreen mode Exit fullscreen mode

2. Declarative Consumer Group Worker 

from wredis.streams import RedisStreamManager

stream_mgr = RedisStreamManager(host="localhost", verbose=False)

# Register worker with auto-created consumer group
@stream_mgr.on_message(
    stream_name="events:transactions",
    group_name="settlement_workers",
    consumer_name="node_alpha"
)
def process_transaction(data):
    print(f"Settling TX: {data['tx_id']} for {data['amount']} {data['currency']}")

# Start daemon consumer threads and listen for SIGINT
stream_mgr.wait()
Enter fullscreen mode Exit fullscreen mode

What Sets WRedis Apart

  • Zero-Boilerplate Groups: Creates consumer groups on demand without crashing if they already exist.
  • Thread-Managed Polling: Background listeners handle backoff and reconnects automatically without blocking your main application loop.
  • Type-Safe & Battle-Tested: 800+ unit tests, 38 integration tests, and 19 stress tests against real Redis instances.

Installation & Repository 

pip install wredis
Enter fullscreen mode Exit fullscreen mode

Author: William Steve Rodríguez Villamizar (Wisrovi)

Top comments (1)

Collapse
 
william_rodriguez_65a5898 profile image
William Rodriguez •

En sistemas distribuidos reactivos con Kafka, la frontera entre control de flujo y resiliencia ante caídas es sutil. El uso de patrones declarativos con commits atómicos permite que la lógica de reintento no degrade el rebalanceo de particiones.

¿Prefieren delegar el manejo de backpressure y offset commits a middleware declarativo o gestionarlo manualmente por handler?