pipeline-tracer 0.1.0
Lightweight decorator-based pipeline tracing for ML/data projects
License: MIT
Requires Python: >=3.9
Keywords: data-lineage, ml, pipeline, reporting, tracing

# Pipeline Tracer

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

## Installation

```bash
pip install -e /path/to/pipeline_tracer
```

## Quick Start

**In your script (init tracer):**
```python
from pipeline_tracer import Tracer, set_tracer

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

**In your src (add decorators):**
```python
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):**
```python
# 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:**
```mermaid
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

```python
@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

```python
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

```python
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.

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

## License

MIT