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 (0)