pipeline-tracer 0.1.0

Lightweight decorator-based pipeline tracing for ML/data projects

Pipeline Tracer

Lightweight decorator-based pipeline tracing for ML/data projects.

Installation

pip install -e /path/to/pipeline_tracer

Quick Start

In your script (init tracer):

from pipeline_tracer import Tracer, set_tracer

tracer = Tracer("my_pipeline")
set_tracer(tracer)

In your src (add decorators):

from pipeline_tracer import trace

@trace(count="patient_id", desc="Remove null values")
def filter_nulls(df):
    return df.drop_nulls()

@trace(count="user_id", desc="Keep active users only")
def filter_active(df):
    return df.filter(pl.col("status") == "active")

@trace(count="user_id", desc="Aggregate to user level")
def aggregate(df):
    return df.group_by("user_id").agg(...)

Back in script (save results):

# Run pipeline
df = filter_nulls(raw_df)
df = filter_active(df)
df = aggregate(df)

# Save trace
tracer.to_csv("trace.csv")
tracer.to_json("trace.json")

# Generate reports (optional)
from pipeline_tracer import MarkdownReporter, MermaidReporter, ConsortReporter

MarkdownReporter(tracer.to_dict()).generate()  # -> Markdown with Mermaid diagram
MermaidReporter(tracer.to_dict()).generate()   # -> Mermaid flowchart
ConsortReporter(tracer.to_dict()).to_text()    # -> CONSORT text diagram

Output

trace.csv:

step	function	input_rows	output_rows	removed	retention	desc
1	filter_nulls	10,000	8,000	2,000	80.0%	Remove null values
2	filter_active	8,000	6,000	2,000	75.0%	Keep active users
3	aggregate	6,000	500	5,500	8.3%	Aggregate to user level

trace.json: Full metadata including counts per column.

Mermaid flowchart:

flowchart TD
    S0["filter_nulls<br/>n=8,000"] -->|"-2,000"| S1
    S1["filter_active<br/>n=6,000"] -->|"-2,000"| S2
    S2["aggregate<br/>n=500"]

Key Features

• Zero config in src - just @trace() decorator
• Count any column - count="patient_id" tracks unique values
• Auto row tracking - input/output rows captured automatically
• Multiple output formats - CSV, JSON, Markdown, Mermaid, CONSORT
• No tracer = no-op - decorators do nothing if set_tracer() not called

API

Decorator

@trace(
    count="column_name",           # Column to count unique values (optional)
    count=["col1", "col2"],        # Multiple columns (optional)
    desc="Description",            # Step description (optional)
)
def my_function(df):
    ...

Tracer

tracer = Tracer("pipeline_name")
set_tracer(tracer)                 # Set as global tracer

tracer.source("name", df=df, path="file.parquet")  # Register data source
tracer.log("step", input_rows=100, output_rows=80) # Manual logging

tracer.to_csv("trace.csv")         # Export CSV
tracer.to_json("trace.json")       # Export JSON
tracer.to_dict()                   # Get as dict

Reporters

from pipeline_tracer import MarkdownReporter, MermaidReporter, ConsortReporter

# From tracer
data = tracer.to_dict()

# Or from JSON file
data = "trace.json"

MarkdownReporter(data).generate()      # Full Markdown report
MermaidReporter(data).generate()       # Mermaid flowchart code
ConsortReporter(data).to_text()        # Text-based CONSORT diagram
ConsortReporter(data).generate()       # Structured data

Example

See example/ folder for a complete working example.

cd example
python generate_test_data.py  # Generate test data
python run_pipeline.py        # Run pipeline with tracing

License

MIT