Pipeline Tracker SDK

A Python SDK for monitoring and tracking data pipeline executions. Easily log pipeline starts, successes, failures, and steps with built-in status updates and logging.

Features

• Pipeline Monitoring: Log the start, success, and failure of your runs.
• Step Tracking: Track individual steps within a pipeline.
• Robust Exception Logging: Automatically log errors on step failure and re-raise.
• Clean Logging: No more print() statements; uses Python's standard logging library for clean and configurable output.

Installation

pip install pipeline-tracker-sdk

Quick Start

1. Basic Tracking

from pipeline_tracker import Tracker

# Initialize the tracker with your monitoring base URL and API key
tracker = Tracker(
    base_url="https://your-monitoring-domain.com",
    api_key="your-secure-api-key"
)

# Start a run
run_id = tracker.start(
    pipeline_name="Weather Data Pipeline",
    expected_frequency_minutes=1440,
    criticality="HIGH"
    owner_name="Data Team",
    owner_email="datateam@example.com"
)

try:
    # Your pipeline logic goes here
    # ...
    
    # Log success once done
    tracker.success(run_id, records_processed=42)
except Exception as e:
    # Log failure with the error message
    tracker.fail(run_id, error_message=str(e))
    raise e

2. Tracking Individual Steps

You can track individual steps within a run manually:

tracker.step(run_id, step_name="Fetch API Data", status="SUCCESS", message="Fetched 100 rows.")

Or you can use the run_step helper function, which handles step success and failure automatically:

from pipeline_tracker import run_step

def fetch_data(url):
    # Some work
    return "data"

# This will call fetch_data and automatically log success or failure for "Fetch API Data" step.
data = run_step(tracker, run_id, "Fetch API Data", fetch_data, "https://api.example.com")

Configuring Logging

The SDK uses Python's standard logging library under the logger name pipeline_tracker. You can configure logging in your application to control the verbosity and destination of logs:

import logging

logging.basicConfig(level=logging.WARNING)
# Now SDK warnings (e.g. API connection issues) will be printed according to your config.

License

This project is licensed under the MIT License - see the LICENSE file for details.