fastmesh-redis 0.1.0

Redis adapter for FastMesh — RedisSource and RedisSink via redis-py async

fastmesh-redis

Redis Streams adapter for FastMesh — RedisSource and RedisSink via redis-py async.

Install

pip install fastmesh-redis

Usage

from fastmesh import Bus
from fastmesh_redis import RedisSource, RedisSink

source = RedisSource(
    stream="orders",
    group="pipeline",
    consumer="worker-1",
    url="redis://localhost:6379",
)

sink = RedisSink(
    stream="processed-orders",
    url="redis://localhost:6379",
    maxlen=10_000,
)

await Bus().source(source).sink(sink).run()

RedisSource

Consumes from a Redis Stream using XREADGROUP. Each entry → one Message.

Parameter	Default	Description
stream	(required)	Stream key name
group	(required)	Consumer group name (created if missing)
consumer	"fastmesh"	Consumer name
url	"redis://localhost:6379"	Redis URL
block_ms	1000	XREADGROUP block timeout (ms)
batch_size	10	Max entries per read
pending_on_start	True	Re-process unacknowledged entries on start

Entries are auto-acknowledged (XACK) after yielding. The payload is read from the "payload" field (JSON-decoded) or merged from all fields if absent.

RedisSink

Publishes to a Redis Stream using XADD.

Parameter	Default	Description
stream	(required)	Stream key name
url	"redis://localhost:6379"	Redis URL
maxlen	None	Approximate stream max length
extra_fields_fn	None	(Message) -> dict[str, str] extra entry fields

The Redis connection is lazily initialized and reused. Call await sink.stop() on shutdown.

Requirements

• Python 3.11+
• fastmesh >= 0.2
• redis[asyncio] >= 5.0 (redis-py with async support)