flowforge

A small, config-driven, pluggable ETL pipeline toolkit for Python.

Describe a pipeline as YAML — a source, a chain of transform steps, a destination — and flowforge runs it as a single streaming pass, so records flow through extract → transform → load without loading the whole dataset into memory.

source:
  type: csv
  params:
    path: examples/data/people.csv

steps:
  - type: drop_nulls
    params:
      columns: ["age"]
  - type: filter_rows
    params:
      column: city
      op: eq
      value: NY
  - type: rename_columns
    params:
      mapping:
        name: full_name

destination:
  type: sqlite
  params:
    path: examples/output/people.db
    table: people

flowforge run examples/pipeline.yaml

Why

Most small ETL scripts hardcode their source, transforms, and destination together, so every new dataset means copy-pasting the script and editing it in place. flowforge separates those concerns into three interchangeable roles and lets a config file wire them together, so adding a new pipeline is a YAML change, not a new script.

Architecture

Three abstract roles, each with a name-based registry so concrete implementations can be referenced by string from config:

Extractor.extract()      -> Iterator[Record]
Transformer.transform()  -> Iterator[Record]
Loader.load()             -> int (records written)

• flowforge.registry.Registry — a generic decorator-based registry. Each of extractors/transformers/loaders owns one, so @extractor_registry.register("csv") is how a class becomes addressable as type: csv in a config file.
• flowforge.config — Pydantic models (PipelineConfig, StepConfig) that validate the YAML shape before anything runs.
• flowforge.pipeline.Pipeline — looks up the registered classes for source, each step, and destination, and chains them as lazy generators: extractor → transformer → transformer → ... → loader, so records stream through one at a time.
• flowforge.cli — a thin Typer CLI (flowforge run <config>) over Pipeline.

Every extractor/transformer/loader validates its own params with a small Pydantic model, so a bad config fails fast with a clear error instead of an AttributeError three steps into a run.

Built-in steps

Role	type	Params
Extractor	csv	path, encoding
Extractor	json	path, encoding
Transformer	rename_columns	mapping: {old: new}
Transformer	select_columns	columns: [...]
Transformer	filter_rows	column, op (eq/ne/gt/lt/ge/le/contains), value
Transformer	drop_nulls	columns (optional; defaults to all)
Loader	csv	path, encoding
Loader	sqlite	path, table, if_exists (replace/append)

Adding a new step

Register a class against the right registry and it's immediately usable from config — no changes needed anywhere else:

from flowforge.transformers.base import Transformer, transformer_registry

@transformer_registry.register("uppercase")
class UppercaseTransformer(Transformer):
    def __init__(self, **params):
        self.column = params["column"]

    def transform(self, records):
        for record in records:
            record[self.column] = str(record[self.column]).upper()
            yield record

Development

pip install -e ".[dev]"
ruff check .
mypy src
pytest --cov=flowforge

License

MIT