DataExcept

DataExcept is a production-ready Python library that provides structured, hierarchical exception classes specifically designed for data science, machine learning, and data engineering workflows. Stop debugging generic ValueErrors and RuntimeErrors -- get meaningful, actionable error messages that help you understand exactly what went wrong in your data pipeline.

🚀 Why DataExcept?

🎯 Key Features

• 🏗️ Hierarchical Structure: Catch one specific error, a whole domain, or every operational error via DataExceptError
• 📦 One Import: Every exception is available from dataexcept directly, or from its domain module — same objects either way
• 📊 Data Science Focused: 106 exception classes covering ML pipelines, feature engineering, model training
• 🧭 Language-neutral envelopes: exceptions export to strict JSON against a published, versioned schema — so a non-Python consumer can validate what it receives, or read it through the Pino projection
• 🔧 Production Ready: Logging helpers, error context, and exceptions that pickle — so they cross a process boundary with their message, attributes and cause intact
• 📚 Academic Quality: Proper documentation, type hints, and citation support
• 🐍 Python 3.10 – 3.14: Every supported version tested in CI, with full type safety
• 🧪 Well Tested: Full branch coverage of the package gated in CI, with contract tests over every exception class

📦 Quick Installation

pip install DataExcept

For development:

git clone https://github.com/DiogoRibeiro7/DataExcept.git
cd DataExcept
poetry install

🏃‍♂️ Quick Start

Basic Usage

from dataexcept import DataLoadingError, ModelTrainingError, ValidationError
import pandas as pd

# Data validation with context
def validate_dataframe(df: pd.DataFrame) -> None:
    if 'customer_id' not in df.columns:
        raise ValidationError(
            field='customer_id',
            value=list(df.columns),
            message="Customer ID column is required for processing"
        )

# Model training with specific error types
def train_model(model_type: str, epochs: int) -> None:
    try:
        # Your training code here
        if epochs > 1000:
            raise ModelTrainingError(
                model_type=model_type, 
                epoch=epochs,
                message=f"Training {model_type} exceeded reasonable epoch limit"
            )
    except Exception as e:
        # Wrap unknown errors with context
        raise ModelTrainingError(model_type, message=f"Unexpected error: {e}")

# File operations with detailed context
def load_dataset(file_path: str) -> pd.DataFrame:
    try:
        return pd.read_csv(file_path)
    except FileNotFoundError as e:
        raise DataLoadingError(source=file_path, original=e)

Exception Hierarchies

from dataexcept import ConvergenceError, DataExceptError, ModelTrainingError

try:
    # Your ML pipeline
    train_complex_model()
except ConvergenceError:
    # Handle specific convergence issues
    logger.warning("Model didn't converge, trying with different parameters")
    train_with_fallback_params()
except ModelTrainingError:
    # Handle any training-related error
    logger.error("Training failed, falling back to simpler model")
    train_simple_model()
except DataExceptError:
    # Handle anything else DataExcept raised
    logger.error("Job failed, notifying administrators")
    send_alert()

🏗️ Exception Categories

📊 Data Science & ML

from dataexcept.datascience_exceptions import *

# Data ingestion and validation
DataLoadingError("data.csv", FileNotFoundError())
DataValidationError("age", -5, "Age cannot be negative")
MissingDataError("income", "Required for credit scoring")

# Feature engineering and preprocessing
FeatureEngineeringError("log_transform", "Cannot take log of negative values")
DataNormalizationError("StandardScaler", "Division by zero in variance calculation")
DataImbalanceError(ratio=0.05, threshold=0.1)

# Model training and evaluation
ModelTrainingError("RandomForest", epoch=45)
ConvergenceError("GradientBoosting", iterations=1000)
OverfittingError(train_metric=0.98, val_metric=0.65)
BiasDetectionError("gender", bias_score=0.15, threshold=0.1)

# Model deployment and inference
ModelInferenceError("CNN", RuntimeError("CUDA out of memory"))
ModelCompatibilityError("2.1.0", "1.8.0")

🔧 Data Engineering & ETL

from dataexcept.dataengineering_exceptions import *

ETLJobError("daily_customer_pipeline")
SchemaEvolutionError("v2.1", reason="Incompatible column type change")
DataTransformationError("currency_conversion", "Invalid exchange rate")
BatchProcessingError("batch_2023_11_13", original=TimeoutError())

🐼 Pandas Operations

from dataexcept.pandas_exceptions import *

MissingColumnError("customer_id", dataframe="sales_df")
DtypeMismatchError("revenue", expected=["float64", "int64"], found="object")
MergeKeyError(["customer_id"], ["cust_id"])

🔗 Infrastructure & Networking

from dataexcept.network_exceptions import *
from dataexcept.database_exceptions import *

HostUnreachableError("api.example.com")
DatabaseConnectionError("postgresql://prod-db:5432/analytics")
QueryExecutionError("SELECT * FROM large_table", original=TimeoutError())

📨 Message Brokers

from dataexcept.broker_exceptions import *

BrokerConnectionError("kafka://broker.internal:9092", cause=OSError("refused"))
MessagePublishError("orders", partition=3, cause=OSError("leader not available"))
MessageConsumeError("orders", partition=3, offset=1042, consumer_group="billing")
MessageAcknowledgementError("orders", partition=3, offset=1042)

🔍 Advanced Features

Smart Logging Integration

from dataexcept.logging_helpers import log_and_raise, log_exception
import logging

logger = logging.getLogger(__name__)

# Context manager for automatic logging
with log_and_raise(logger=logger, context={"job_id": "ETL_001", "batch": "2023-11-13"}):
    process_daily_batch()

# Manual exception logging with context
try:
    risky_operation()
except Exception as exc:
    log_exception(
        exc, 
        logger=logger,
        context={"user_id": "12345", "operation": "feature_extraction"}
    )
    raise

Structured Error Envelopes

from dataexcept import ValidationError, exception_to_json

try:
    validate_row(row)
except ValidationError as exc:
    # Strict JSON: identity, message, public attributes, failure metadata and
    # the cause chain, with credential-bearing URLs already redacted.
    logger.error(exception_to_json(exc, sort_keys=True))

The payload shape is a versioned, language-neutral contract, published as a JSON Schema with reference fixtures, so a Node.js or Go consumer can be tested against the same document Python emits.

Command Line Interface

# List every exception class the package exports (106 of them, alphabetically)
$ dataexcept list
ApiError
AuthenticationError
AuthorizationError
BatchProcessingError
BiasDetectionError
...

# Check the installed version
$ dataexcept --version
dataexcept <installed version>

🎯 Use Cases

🏭 Production ML Pipelines

• Model Training: Distinguish between convergence issues, data problems, and infrastructure failures
• Feature Engineering: Track which transformation steps fail and why
• Model Serving: Provide actionable error messages for inference failures
• Data Drift: Alert when model assumptions are violated

📈 Data Engineering

• ETL Pipelines: Clear error categorization for debugging complex data flows
• Data Quality: Structured validation errors with field-level context
• Schema Evolution: Track migration failures and compatibility issues
• Batch Processing: Identify whether failures are data-related or system-related

🔬 Research & Academia

• Reproducible Experiments: Consistent error handling across research codebases
• Citation Support: Proper academic attribution with CITATION.cff
• Documentation: Auto-generated API docs with comprehensive examples

📚 Real-World Example

"""
Complete ML pipeline with DataExcept error handling
"""
import numpy as np
import pandas as pd
from sklearn.ensemble import RandomForestClassifier
from dataexcept import ValidationError
from dataexcept.datascience_exceptions import *
from dataexcept.pandas_exceptions import *
from dataexcept.logging_helpers import log_and_raise
import logging

def ml_pipeline(data_path: str, target_col: str):
    logger = logging.getLogger(__name__)

    with log_and_raise(logger=logger, context={"pipeline": "customer_churn"}):
        # 1\. Data Loading
        try:
            df = pd.read_csv(data_path)
        except FileNotFoundError as e:
            raise DataLoadingError(source=data_path, original=e)

        # 2\. Data Validation
        if target_col not in df.columns:
            raise MissingColumnError(target_col, dataframe="training_data")

        if df[target_col].dtype not in ['int64', 'bool']:
            raise DtypeMismatchError(
                target_col, 
                expected=['int64', 'bool'], 
                found=str(df[target_col].dtype)
            )

        # 3\. Data Quality Checks
        missing_ratio = df.isnull().sum().sum() / (df.shape[0] * df.shape[1])
        if missing_ratio > 0.3:
            raise DataValidationError(
                field="missing_data_ratio",
                value=missing_ratio,
                message=f"Dataset has {missing_ratio:.1%} missing values, exceeds 30% threshold"
            )

        # 4\. Class Imbalance Check
        class_ratio = df[target_col].value_counts().min() / df[target_col].value_counts().max()
        if class_ratio < 0.1:
            raise DataImbalanceError(ratio=class_ratio, threshold=0.1)

        # 5\. Feature Engineering
        try:
            df['log_revenue'] = np.log(df['revenue'] + 1)
        except Exception as e:
            raise FeatureEngineeringError("log_transform", cause=str(e))

        # 6\. Model Training
        try:
            model = RandomForestClassifier(n_estimators=100)
            X = df.drop(columns=[target_col])
            y = df[target_col]
            model.fit(X, y)
        except Exception as e:
            raise ModelTrainingError("RandomForest", message=f"Training failed: {e}")

        # 7\. Model Validation
        train_score = model.score(X, y)
        if train_score < 0.6:
            raise UnderfittingError(train_metric=train_score, threshold=0.6)

        return model

# Usage
if __name__ == "__main__":
    try:
        model = ml_pipeline("customer_data.csv", "churned")
        print("✅ Pipeline completed successfully!")
    except DataLoadingError as e:
        print(f"❌ Data loading failed: {e}")
    except MissingColumnError as e:
        print(f"❌ Schema validation failed: {e}")
    except DataImbalanceError as e:
        print(f"⚠️  Data quality issue: {e}")
    except ModelTrainingError as e:
        print(f"❌ Model training failed: {e}")
    except Exception as e:
        print(f"💥 Unexpected error: {e}")

🤝 Contributing

We welcome contributions! See our Contributing Guide for details.

# Development setup
git clone https://github.com/DiogoRibeiro7/DataExcept.git
cd DataExcept
make install          # poetry install --with dev,docs
pre-commit install

make check            # lint, formatting, mypy and tests - everything CI runs
make help             # list all targets

Please also read the Code of Conduct. Security issues go through SECURITY.md, not the public issue tracker.

📖 Documentation

• Full Documentation: diogoribeiro7.github.io/DataExcept
• API Reference: API Docs
• API Stability: What is public and what may change
• Upgrading from 0.x: Migration guide
• Advanced Usage: Advanced Guide
• Cause-aware Exceptions: Wrapping a third-party failure
• Failure Metadata: Transient, permanent, or unclassified
• Parsing Context: Reporting a failure without keeping the payload
• Message Brokers: Publish, consume and acknowledgement failures
• Envelope Schema: The published envelope contract
• Pino Interoperability: Consuming envelopes from Node.js
• CLI Reference: CLI Guide
• Changelog: CHANGELOG.md

🎓 Citation

If you use DataExcept in your research, please cite the exact release you used. The canonical release metadata is maintained in CITATION.cff, and the citation guide explains how to get the APA and BibTeX forms from it.

📄 License

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

🏆 About the Author

Diogo Ribeiro is a Lead Data Scientist at Mysense.ai and researcher/instructor at ESMAD (Instituto Politécnico do Porto). With expertise in machine learning, statistical analysis, and production ML systems, he created DataExcept to solve real-world error handling challenges in data science workflows.

• 🔗 ORCID: 0009-0001-2022-7072
• 🌐 Website: diogoribeiro7.github.io
• 🏢 Affiliation: ESMAD - Instituto Politécnico do Porto

⭐ Star this repo if DataExcept helps you build better data pipelines!