Agora ETL

Async-first ETL framework for Python.

agora-etl is the core Agora runtime: a builder-first ETL framework organized around a Source → Middleware chain → Sink(s) model.

The core package owns runtime semantics, public contracts, CLI diagnostics, checkpointing, DLQ behavior, scheduling primitives, health/metrics surfaces, and plugin discovery. Backend integrations such as Redis, Kafka, PostgreSQL, cron scheduling, distributed coordination, and Anthropic live in the official agora-etl-plugins package.

Keep the package story simple:

• agora-etl owns runtime semantics and public framework contracts
• agora-etl-plugins owns official backend implementations and backend runbooks
• agora-etl-rs accelerates selected hot paths through the core acceleration boundary

Source of truth

Use this README for install, quickstart, and package boundary orientation. For canonical docs beyond that:

• runtime contract: Runtime Guarantees
• plugin ecosystem and backend boundaries: Plugins
• support-boundary map: Source Of Truth Map
• CLI behavior: CLI

Install

pip install agora-etl

Optional core extras:

pip install "agora-etl[file]"       # pyarrow + faster JSONL/file paths
pip install "agora-etl[rs]"         # optional Rust acceleration boundary
pip install "agora-etl[benchmark]"  # local benchmarking helpers

Official integrations:

pip install "agora-etl-plugins[redis]"
pip install "agora-etl-plugins[kafka]"
pip install "agora-etl-plugins[postgres]"
pip install "agora-etl-plugins[all]"

agora-etl-plugins 0.4.x targets agora-etl>=0.4.1,<1.

Quickstart

import asyncio

from agora import DeliveryConfig, IterableSource, Pipeline
from agora.core.dlq import SQLiteDLQSink


records = [
    {"id": 1, "city": "Ho Chi Minh City", "confidence": 0.92},
    {"id": 2, "city": "Da Nang", "confidence": 0.41},
    {"id": 3, "city": "Hanoi", "confidence": 0.88},
]


async def main() -> None:
    summary = await (
        Pipeline(IterableSource(records))
        .filter(lambda row: row["confidence"] >= 0.5, name="confidence_gate")
        .build(
            config=DeliveryConfig(
                dlq=SQLiteDLQSink(".agora_dlq.db"),
                batch_size=100,
            ),
        )
        .run()
    )
    print(summary)


asyncio.run(main())

build() without an explicit sink uses the default stdout sink. The example:

• emits in-memory records
• filters low-confidence rows
• writes accepted records to stdout
• routes failed records to a local SQLite DLQ
• returns a PipelineRunSummary with counts, timing, and runtime signals

Capability snapshot

Runtime guarantees

Agora's public runtime contract is documented in Runtime Guarantees. The high-level model:

• source order is preserved at the sink boundary
• checkpoint advancement waits for handled outcomes
• sink delivery is fail-closed by default
• DLQ replay acknowledges only after durable replay success
• batch, Arrow, and process-isolated paths keep the same correctness contract
• at-least-once delivery is the model; exactly-once behavior requires external idempotency or transactional systems

Execution lanes and data planes

Agora selects the execution lane from explicit source, middleware, and sink contracts:

• python_rows
• python_batches
• arrow_batches

Use .explain() before a run to inspect lane selection and sink downgrade decisions:

from agora import Pipeline
from agora.sinks.file.csv import CsvSink


bound = Pipeline(source).build(CsvSink(path="out.csv", row_mapper=lambda row: row))
plan = bound.explain(max_records=1_000)

print(plan)
print(plan.to_dict())

PipelineExplain includes the selected lane, source data plane, writer input plane, middleware compatibility, sink plane, and downgrade markers.

Core vs plugins

Keep the package boundary clear:

If a capability depends on Redis, Kafka, PostgreSQL, cron parsing, distributed lease ownership, or a hosted AI provider, it belongs in a plugin package rather than core.

CLI

The package installs the agora command:

agora --help
agora doctor
agora plugins list
agora dlq replay --help

Use agora doctor and agora plugins list --json in release gates and operator diagnostics when installed package behavior matters.

Documentation map

• Quickstart
• Pipelines
• Runtime Guarantees
• Failure Handling
• Recovery Matrix
• Checkpointing
• Scheduling and Workers
• Observability
• Performance
• Plugins
• Architecture
• CLI
• Change Log

Reference sections:

• Sources
• Sinks
• Middlewares
• Schema
• State

Development

python -m venv .venv
.venv/bin/pip install -e ".[file,dev,benchmark]"
make ci

Useful focused checks:

make docs-check
make test-core
make contracts
make preservation

The GitHub CI matrix runs on Python 3.11, 3.12, and 3.13.

Security

Security reporting instructions live in SECURITY.md. Do not include exploitable details, credentials, payloads, production hostnames, or private stack traces in public issues.

License

Apache 2.0 — see LICENSE.