GoldenPipe

GoldenPipe

Orchestrator for the Golden Suite. Wires GoldenCheck → GoldenFlow → GoldenMatch into a single declarative pipeline driven by YAML, with adaptive logic that decides which stages to run based on data quality findings and dataset shape.

Source: packages/python/goldenpipe · PyPI: goldenpipe

Install

pip install goldenpipe[full]   # brings Check + Flow + Match together

What it does

Capability	Notes
Stage discovery	Reads goldencheck.scan, goldenflow.transform, goldenmatch.dedupe via entry points
YAML config	pipeline.yaml declares pipeline name + stage list + per-stage settings
Adaptive flow	decide_flow(findings) skips Flow if Check found nothing actionable; decide_match(findings, row_count, strategy_override) picks dedupe vs agent-mode
Result	PipeResult with status, stages, artifacts, errors, reasoning, timing

DQBench Pipeline Score: 88.07 (without LLM).

Programmatic API

import goldenpipe as gp

# YAML-driven
pipeline = gp.Pipeline.from_yaml("pipeline.yaml")
result = pipeline.run("customers.csv")

# Inspect stage decisions
for name, stage_result in result.stages.items():
    print(f"{name}: {stage_result.status} — {result.reasoning.get(name, '')}")

API gotcha: Pipeline.run() does not return the output DataFrame — only the PipeResult summary. The PipeContext.df holds the final data internally but isn't exposed in the result. To get the cleaned + deduped data, run the stages directly (goldenflow.transform_df then goldenmatch.dedupe_df) or read the artifact path GoldenPipe wrote to.

Suite role

                 raw rows
                    │
                    ▼
       ┌─────────────────────────┐
       │  goldencheck.scan       │ ─► findings
       └────────────┬────────────┘
                    │ decide_flow(findings)
                    ▼
       ┌─────────────────────────┐
       │  goldenflow.transform   │ ─► cleaned df
       └────────────┬────────────┘
                    │ decide_match(findings, row_count, strategy)
                    ▼
       ┌─────────────────────────┐
       │  goldenmatch.dedupe     │ ─► golden records
       └─────────────────────────┘

For pipelines that need stage-level retries, mid-pipeline branching, or custom logic between stages, prefer the Airflow DAGs which compose the same packages with full observability. GoldenPipe is the right answer when "Check → Flow → Match" is the whole story.

YAML config

pipeline: customer-dedupe
stages:
  - use: goldencheck.scan
  - use: goldenflow.transform
    config: flow.yaml
  - use: goldenmatch.dedupe
    config: match.yaml

Use PipelineConfig(stages=[...]) programmatically to skip stages or override them.

CLI

goldenpipe run data.csv                         # use pipeline.yaml in cwd
goldenpipe run data.csv --pipeline custom.yaml
goldenpipe stages                               # list discovered stages
goldenpipe explain custom.yaml                  # explain what a config does
goldenpipe mcp-serve --transport http --port 8250

MCP

ghcr.io/benzsevern/goldenpipe-mcp:latest. 4 tools: list_stages, validate_pipeline, run_pipeline, explain_pipeline. Surfaced under the goldensuite-mcp aggregator.

See also

• Containers
• Airflow DAGs — production-shape alternative for non-trivial pipelines
• Master MCP Aggregator
• Full README: packages/python/goldenpipe/README.md