Pipeline

The Pipeline class is the main entry point for building and running ETL pipelines.

Overview

from pycharter import Pipeline, HTTPExtractor, FileLoader, Rename

# Build with pipe operator
pipeline = (
    Pipeline(HTTPExtractor(url="https://api.example.com/data"))
    | Rename({"old": "new"})
    | FileLoader(path="output.json")
)

# Run
result = await pipeline.run()

API Reference

Pipeline

Pipeline(config: PipelineConfig, steps: list[StepConfig])

DAG-based ETL pipeline.

Pipelines are constructed from YAML config files or dicts and consist of a directed acyclic graph of steps (extract, transform, join, union, load).

Example::

pipeline = Pipeline.from_file("pipeline.yaml")
result = asyncio.run(pipeline.run())

Source code in src/pycharter/pipeline_generator/pipeline.py

def __init__(
    self,
    config: PipelineConfig,
    steps: list[StepConfig],
) -> None:
    self._config = config
    self._steps = steps

run async

run(*, run_id: str | None = None, fail_fast: bool = True) -> PipelineResult

Execute the pipeline.

Parameters:

Name	Type	Description	Default
run_id	str | None	Optional identifier for this run.	None
fail_fast	bool	Abort on the first step failure.	True

Returns:

Type	Description
PipelineResult	Aggregated pipeline result with per-step results.

Source code in src/pycharter/pipeline_generator/pipeline.py

async def run(
    self,
    *,
    run_id: str | None = None,
    fail_fast: bool = True,
) -> PipelineResult:
    """Execute the pipeline.

    Args:
        run_id: Optional identifier for this run.
        fail_fast: Abort on the first step failure.

    Returns:
        Aggregated pipeline result with per-step results.
    """
    return await run_dag(
        self._steps,
        pipeline_name=self._config.name,
        run_id=run_id,
        variables=self._config.variables,
        fail_fast=fail_fast,
    )

from_config_dir classmethod

from_config_dir(directory: str | Path, *, extra_variables: dict[str, str] | None = None) -> Pipeline

Create a pipeline from a directory of config files.

Looks for a pipeline.yaml (or .yml) first. If not found, assembles a section-based config from individual extract.yaml, load.yaml, transform.yaml files. Also loads variables.yaml and settings.yaml if present.

Parameters:

Name	Type	Description	Default
directory	str | Path	Path to the config directory.	required
extra_variables	dict[str, str] | None	Additional variables.	None

Returns:

Type	Description
Pipeline	A validated, ready-to-run pipeline.

Source code in src/pycharter/pipeline_generator/pipeline.py

@classmethod
def from_config_dir(
    cls,
    directory: str | Path,
    *,
    extra_variables: dict[str, str] | None = None,
) -> Pipeline:
    """Create a pipeline from a directory of config files.

    Looks for a ``pipeline.yaml`` (or ``.yml``) first.  If not found,
    assembles a section-based config from individual
    ``extract.yaml``, ``load.yaml``, ``transform.yaml`` files.
    Also loads ``variables.yaml`` and ``settings.yaml`` if present.

    Args:
        directory: Path to the config directory.
        extra_variables: Additional variables.

    Returns:
        A validated, ready-to-run pipeline.
    """
    dirpath = Path(directory)
    if not dirpath.is_dir():
        raise FileNotFoundError(f"Config directory not found: {dirpath}")

    pipeline_file = _find_yaml(dirpath, "pipeline")
    if pipeline_file:
        return cls.from_file(pipeline_file, extra_variables=extra_variables)

    extract_file = _find_yaml(dirpath, "extract")
    load_file = _find_yaml(dirpath, "load")
    if not extract_file or not load_file:
        raise FileNotFoundError(
            f"Config directory must contain pipeline.yaml or "
            f"extract.yaml + load.yaml: {dirpath}"
        )

    raw = _merge_section_files(dirpath, extract_file, load_file)
    return cls.from_dict(raw, extra_variables=extra_variables)

from_config_files classmethod

from_config_files(*paths: str | Path, extra_variables: dict[str, str] | None = None) -> Pipeline

Create a pipeline by merging multiple YAML config files.

Each file is loaded and merged (later files override earlier ones on key collisions).

Parameters:

Name	Type	Description	Default
paths	str | Path	One or more YAML file paths.	()
extra_variables	dict[str, str] | None	Additional variables.	None

Returns:

Type	Description
Pipeline	A validated, ready-to-run pipeline.

Source code in src/pycharter/pipeline_generator/pipeline.py

@classmethod
def from_config_files(
    cls,
    *paths: str | Path,
    extra_variables: dict[str, str] | None = None,
) -> Pipeline:
    """Create a pipeline by merging multiple YAML config files.

    Each file is loaded and merged (later files override earlier ones
    on key collisions).

    Args:
        paths: One or more YAML file paths.
        extra_variables: Additional variables.

    Returns:
        A validated, ready-to-run pipeline.
    """
    if not paths:
        raise ValueError("At least one config file path is required.")

    merged: dict[str, Any] = {}
    for p in paths:
        filepath = Path(p)
        if not filepath.exists():
            raise FileNotFoundError(f"Config file not found: {filepath}")
        with open(filepath) as f:
            data = yaml.safe_load(f) or {}
        if not isinstance(data, dict):
            raise ValueError(
                f"Expected YAML mapping in {filepath}, got {type(data).__name__}"
            )
        merged.update(data)

    return cls.from_dict(merged, extra_variables=extra_variables)

Factory Methods

from_config_dir

Load pipeline from a directory containing extract.yaml, transform.yaml, and load.yaml:

pipeline = Pipeline.from_config_dir("pipelines/users/")

from_config_files

Load from explicit file paths:

pipeline = Pipeline.from_config_files(
    extract="configs/extract.yaml",
    transform="configs/transform.yaml",  # Optional
    load="configs/load.yaml",
    variables={"API_KEY": "secret"}
)

from_config_file

Load from a single combined config file:

pipeline = Pipeline.from_config_file("pipeline.yaml")

PipelineResult

PipelineResult dataclass

PipelineResult(success: bool = True, rows_extracted: int = 0, rows_transformed: int = 0, rows_loaded: int = 0, rows_failed: int = 0, rows_quarantined_extract: int = 0, rows_quarantined_load: int = 0, validation_errors_extract: list[str] = list(), validation_errors_load: list[str] = list(), start_time: datetime | None = None, end_time: datetime | None = None, duration_seconds: float | None = None, batches_processed: int = 0, batch_results: list[BatchResult] = list(), errors: list[str] = list(), pipeline_name: str | None = None, run_id: str | None = None, quality_report: Any = None, lineage_events: list[dict[str, Any]] = list(), step_results: list[StepResult] = list())

Complete result from running an ETL pipeline.

success class-attribute instance-attribute

success: bool = True

rows_extracted class-attribute instance-attribute

rows_extracted: int = 0

rows_transformed class-attribute instance-attribute

rows_transformed: int = 0

rows_loaded class-attribute instance-attribute

rows_loaded: int = 0

rows_failed class-attribute instance-attribute

rows_failed: int = 0

duration_seconds class-attribute instance-attribute

duration_seconds: float | None = None

errors class-attribute instance-attribute

errors: list[str] = field(default_factory=list)

Examples

Basic Pipeline

import asyncio
from pycharter import Pipeline, FileExtractor, FileLoader

pipeline = (
    Pipeline(FileExtractor(path="input.json"))
    | FileLoader(path="output.json")
)

result = asyncio.run(pipeline.run())
print(f"Loaded {result.rows_loaded} rows")

With Transformations

from pycharter import Pipeline, HTTPExtractor, PostgresLoader, Rename, Filter, AddField

pipeline = (
    Pipeline(HTTPExtractor(url="https://api.example.com/users"))
    | Rename({"userName": "user_name"})
    | Filter(lambda r: r.get("active"))
    | AddField("processed_at", "now()")
    | PostgresLoader(connection_string="...", table="users")
)

Error Handling

from pycharter.shared.errors import ErrorMode, ErrorContext

# Collect errors instead of raising
result = await pipeline.run(
    error_context=ErrorContext(mode=ErrorMode.COLLECT)
)

if result.errors:
    for error in result.errors:
        print(f"Error: {error}")

With Variables

pipeline = Pipeline.from_config_dir(
    "pipelines/users/",
    variables={
        "API_KEY": os.environ["API_KEY"],
        "OUTPUT_PATH": "/data/output.json"
    }
)

Message Queue Acknowledgment

When a pipeline uses a messaging extractor (Kafka, RabbitMQ, SQS), the pipeline automatically acknowledges each batch after loading:

from pycharter.pipeline_generator.extractors import KafkaExtractor

pipeline = (
    Pipeline(KafkaExtractor(topics=["orders"], consumer_group="etl"))
    | Rename({"orderId": "order_id"})
    | PostgresLoader(connection_string="...", table="orders")
)

# Pipeline calls extractor.acknowledge() after each batch load
result = await pipeline.run()

# Don't forget to close the consumer
await pipeline.extractor.close()

The pipeline determines success from the LoadResult:

Scenario	Acknowledge call
Load succeeds	acknowledge(batch_index, success=True)
Load fails	acknowledge(batch_index, success=False)
Dry run	acknowledge(batch_index, success=True)
No loader set	acknowledge(batch_index, success=True)

If acknowledge() raises an exception, it is logged as a warning and the pipeline continues.

Incremental Extraction

Track a watermark field across runs so only new/updated records are extracted:

extract.yaml

type: database
query: "SELECT * FROM events WHERE updated_at > '${watermark}'"
connection_string: "${DATABASE_URL}"
incremental:
  enabled: true
  watermark_field: updated_at
  initial_value: "2024-01-01"

settings.yaml

state_store:
  backend: file      # or "sqlite"
  path: ./.state

The pipeline persists the highest watermark value on success and injects it into the next run.

Testing Utilities

PyCharter provides mock classes and assertion helpers for testing pipelines without real I/O:

from pycharter import MockExtractor, MockLoader, PipelineTestHarness

# Mock extractor yields fixture data
extractor = MockExtractor(data=[
    [{"id": 1, "name": "Alice"}, {"id": 2, "name": "Bob"}]
])

# Mock loader captures what was loaded
loader = MockLoader()

pipeline = Pipeline(extractor) | Rename({"name": "full_name"}) | loader
result = await pipeline.run()

# Assert on captured data
from pycharter import assert_record_count, assert_fields_present
assert_record_count(loader.loaded_data, 2)
assert_fields_present(loader.loaded_data, ["id", "full_name"])

See Also

• Streaming and Messaging Guide
• ETL Pipelines Tutorial
• Extractors
• Transformers
• Loaders