📊 Python Data Workflows – 📋 Logging & Monitoring 🐍

import logging
import time
from pathlib import Path

import pandas as pd


INPUT_FILE = Path("sales_data.csv")
OUTPUT_FILE = Path("processed_sales_data.csv")

LOG_DIR = Path("logs")
LOG_FILE = LOG_DIR / "data_workflow.log"


# -------------------------------
# Configure Logging
# -------------------------------
LOG_DIR.mkdir(exist_ok=True)

logging.basicConfig(
    level=logging.INFO,
    format=(
        "%(asctime)s | %(levelname)s | "
        "%(name)s | %(message)s"
    ),
    handlers=[
        logging.FileHandler(
            LOG_FILE,
            encoding="utf-8"
        ),
        logging.StreamHandler()
    ]
)

logger = logging.getLogger(
    "data_workflow"
)


# -------------------------------
# Load Data
# -------------------------------
def load_data(
    file_path: Path
) -> pd.DataFrame:
    """Load the source dataset."""
    logger.info(
        "Loading dataset | file=%s",
        file_path
    )

    if not file_path.exists():
        raise FileNotFoundError(
            f"Input file not found: {file_path}"
        )

    df = pd.read_csv(file_path)

    logger.info(
        "Dataset loaded | "
        "rows=%d | columns=%d",
        len(df),
        len(df.columns)
    )

    return df


# -------------------------------
# Validate Data
# -------------------------------
def validate_data(
    df: pd.DataFrame
) -> None:
    """Validate required columns."""
    required_columns = {
        "order_id",
        "customer_name",
        "category",
        "region",
        "sales",
        "quantity",
        "order_date",
    }

    missing_columns = (
        required_columns -
        set(df.columns)
    )

    if missing_columns:
        logger.error(
            "Validation failed | "
            "missing_columns=%s",
            sorted(missing_columns)
        )

        raise ValueError(
            "Missing required columns: "
            f"{sorted(missing_columns)}"
        )

    duplicate_ids = (
        df["order_id"]
        .duplicated()
        .sum()
    )

    missing_values = int(
        df.isnull()
        .sum()
        .sum()
    )

    logger.info(
        "Validation passed | "
        "duplicate_order_ids=%d | "
        "missing_values=%d",
        duplicate_ids,
        missing_values
    )


# -------------------------------
# Clean Data
# -------------------------------
def clean_data(
    df: pd.DataFrame
) -> pd.DataFrame:
    """Clean and prepare the dataset."""
    cleaned_df = df.copy()

    input_rows = len(cleaned_df)

    cleaned_df = (
        cleaned_df
        .drop_duplicates(
            subset=["order_id"],
            keep="first"
        )
    )

    cleaned_df["sales"] = (
        pd.to_numeric(
            cleaned_df["sales"],
            errors="coerce"
        )
    )

    cleaned_df["quantity"] = (
        pd.to_numeric(
            cleaned_df["quantity"],
            errors="coerce"
        )
    )

    cleaned_df["order_date"] = (
        pd.to_datetime(
            cleaned_df["order_date"],
            errors="coerce"
        )
    )

    cleaned_df = (
        cleaned_df
        .dropna(
            subset=[
                "sales",
                "quantity",
                "order_date"
            ]
        )
    )

    cleaned_df = cleaned_df[
        (cleaned_df["sales"] >= 0) &
        (cleaned_df["quantity"] > 0)
    ].copy()

    output_rows = len(cleaned_df)
    removed_rows = (
        input_rows - output_rows
    )

    logger.info(
        "Cleaning completed | "
        "input_rows=%d | "
        "output_rows=%d | "
        "removed_rows=%d",
        input_rows,
        output_rows,
        removed_rows
    )

    return cleaned_df


# -------------------------------
# Generate Metrics
# -------------------------------
def generate_metrics(
    df: pd.DataFrame
) -> dict:
    """Calculate workflow metrics."""
    metrics = {
        "processed_records": len(df),
        "total_sales": (
            float(df["sales"].sum())
        ),
        "average_sales": (
            float(df["sales"].mean())
        ),
        "total_quantity": (
            float(df["quantity"].sum())
        ),
    }

    logger.info(
        "Metrics generated | "
        "processed_records=%d | "
        "total_sales=%.2f | "
        "average_sales=%.2f | "
        "total_quantity=%.2f",
        metrics["processed_records"],
        metrics["total_sales"],
        metrics["average_sales"],
        metrics["total_quantity"]
    )

    return metrics


# -------------------------------
# Save Data
# -------------------------------
def save_data(
    df: pd.DataFrame,
    output_file: Path
) -> None:
    """Save processed workflow output."""
    df.to_csv(
        output_file,
        index=False
    )

    logger.info(
        "Output saved | "
        "file=%s | rows=%d",
        output_file,
        len(df)
    )


# -------------------------------
# Run Workflow
# -------------------------------
def run_workflow() -> None:
    """Run the monitored data workflow."""
    start_time = time.perf_counter()

    logger.info(
        "Workflow started"
    )

    try:
        raw_df = load_data(
            INPUT_FILE
        )

        validate_data(
            raw_df
        )

        cleaned_df = clean_data(
            raw_df
        )

        if cleaned_df.empty:
            raise ValueError(
                "No valid records remain "
                "after cleaning."
            )

        metrics = generate_metrics(
            cleaned_df
        )

        save_data(
            cleaned_df,
            OUTPUT_FILE
        )

        logger.info(
            "Workflow completed | "
            "status=SUCCESS | "
            "processed_records=%d",
            metrics[
                "processed_records"
            ]
        )

    except (
        FileNotFoundError,
        ValueError,
        KeyError,
        pd.errors.ParserError,
    ):
        logger.exception(
            "Workflow failed | "
            "status=FAILED"
        )

    finally:
        execution_time = (
            time.perf_counter() -
            start_time
        )

        logger.info(
            "Execution finished | "
            "duration_seconds=%.3f",
            execution_time
        )


if __name__ == "__main__":
    run_workflow()