agora-etl-plugins 0.3.1

Official plugins for agora-etl — Kafka, PostgreSQL, Redis, cron scheduling, and distributed coordination.

Agora ETL Plugins

Official plugin collection for agora-etl — Redis, cron scheduling, distributed coordination, Kafka, and PostgreSQL.

---

Overview

agora-etl-plugins extends agora-etl with production-ready integrations. Plugins are auto-discovered via Python entry-points — install the package and they register themselves automatically, no manual wiring needed.

Canonical ecosystem docs live in the Agora docs site:

• Plugin ecosystem overview: https://agora.my-working.com/plugins/
• Core docs home: https://agora.my-working.com/

This README stays focused on package-specific quickstart information.

from agora import DeliveryConfig, Pipeline
from agora_plugins.redis.sources import RedisStreamSource
from agora_plugins.redis.sinks import RedisSink

summary = await (
    Pipeline(RedisStreamSource(url="redis://localhost:6379", stream="events", group="my-group", consumer="worker-1"))
    .build(
        RedisSink(url="redis://localhost:6379", key_fn=lambda r: r["id"]),
        config=DeliveryConfig(batch_size=100),
    )
    .run()
)
print(f"written={summary.records_written}  errors={summary.records_errored}")

---

Install

pip install "agora-etl-plugins[redis]"        # Redis source, sink, state, DLQ, dedup, AI cache
pip install "agora-etl-plugins[cron]"         # Cron schedule support for ScheduledPipeline
pip install "agora-etl-plugins[distributed]"  # Redis-backed distributed worker coordination
pip install "agora-etl-plugins[kafka]"        # Kafka source and sink
pip install "agora-etl-plugins[postgres]"     # PostgreSQL source, sink, DLQ, schema adapter
pip install "agora-etl-plugins[all]"          # Everything in one install

This package now targets agora-etl>=0.2.2.

For plugin sources such as Redis Streams, Kafka, and PostgreSQL, agora-etl 0.2.2 adds two core improvements worth adopting by default:

• DeliveryConfig(batch_size=100) improves throughput on the linear lane.
• BatchMiddleware now works correctly even when the source emits one record at a time.

For long-lived consumers that batch writes, agora-etl 0.2.2 also adds DeliveryConfig(batch_flush_interval_ms=...) so partial sink batches can flush without forcing the worker run to end.

If your pipelines checkpoint frequently, you can also enable the Rust checkpoint hot path from the core package:

pip install "agora-etl[rs]" "agora-etl-plugins[redis]"

---

Available plugins

Redis [redis]

Full Redis integration — streaming ingestion, writes, dead-letter queue, state, deduplication, and LLM response caching.

Component	Type	Description
RedisStreamSource	Source	Consume records from a Redis Stream via XREADGROUP
RedisSink	Sink	Write records to Redis (SET / LPUSH / RPUSH / XADD)
RedisDLQSink	Sink	Route failed records to a Redis-backed dead-letter queue
RedisDLQSource	Source	Replay failed records from the Redis DLQ
RedisBackend	State	Redis-backed state backend with TTL and membership support
RedisStore	Dedup	Exact-match deduplication via Redis SET NX
RedisEmbeddingStore	Dedup	Semantic deduplication using cosine similarity (up to ~10k entries)
RedisLLMCache	AI Cache	Distributed LLM response cache backed by Redis

---

License

Apache 2.0 — see LICENSE.