API Reference

Everything on this page is generated directly from the docstrings and type annotations in the source, so it always matches the installed version.

Top-level package

The names below are re-exported from dataexcept itself, so from dataexcept import ValidationError works without reaching into a submodule.

dataexcept

Top-level package for DataExcept.

Every exception the package defines is importable straight from here::

from dataexcept import ValidationError, ModelTrainingError

They all derive from :class:DataExceptError, so one clause catches every operational exception this package raises::

except DataExceptError:
    ...

The domain modules (datascience_exceptions, pipeline_exceptions and so on) remain importable and export the same objects, so both spellings work and refer to the same classes.

DataExceptError

Bases: Exception

Base class for every operational exception DataExcept raises.

Source code in dataexcept/base.py

class DataExceptError(Exception):
    """Base class for every operational exception DataExcept raises."""

    #: Passed to redact_urls_in_text when scrubbing this class's message.
    #: WebhookError sets it False, because a webhook URL's path *is* the
    #: credential.
    _keep_url_path = True

    def __init__(self, *args: Any) -> None:
        # One boundary for the whole hierarchy. Whatever built the message -- a
        # constructor, a caller-supplied `message`, or the text of a wrapped
        # exception quoting the original URL -- it is scrubbed here, because
        # redacting only the structured argument leaves all three routes open.
        keep_path = type(self)._keep_url_path
        if args and isinstance(args[0], str):
            args = (redact_urls_in_text(args[0], keep_path=keep_path),) + args[1:]

        # Many classes store the message on self.message and render *that* in
        # __str__, and 18 interpolate some other attribute -- a field, a
        # column, a resource -- any of which a caller can fill with a URL. So
        # every stored string is swept, not just the message.
        #
        # redact_urls_in_text rather than redact_if_url: a message has the URL
        # embedded in prose, and redact_if_url only handles a value that is
        # wholly a URL. It is a no-op on anything without "://" in it, so
        # ordinary names and file paths are untouched.
        for name, value in list(self.__dict__.items()):
            if isinstance(value, str) and "://" in value:
                self.__dict__[name] = redact_urls_in_text(value, keep_path=keep_path)

        super().__init__(*args)
        # Constructors that wrap another exception record it on an attribute.
        # Mirroring it into __cause__ is what makes a traceback print the
        # underlying failure, exactly as `raise ... from exc` would; assigning
        # __cause__ also sets __suppress_context__, as `raise from` does.
        for attribute in _CAUSE_ATTRIBUTES:
            candidate = getattr(self, attribute, None)
            if isinstance(candidate, BaseException):
                self.__cause__ = candidate
                break

    def __reduce__(self) -> Tuple[Any, Tuple[Any, ...]]:
        args = tuple(_safe(arg) for arg in self.args)
        state = {key: _safe(value) for key, value in self.__dict__.items()}
        return (
            _rebuild,
            (
                type(self),
                args,
                state,
                _safe_exception(self.__cause__),
                _safe_exception(self.__context__),
                self.__suppress_context__,
            ),
        )

UnpicklableCause

Bases: DataExceptError

Stands in for a cause that could not be serialized.

__cause__ and __context__ must be exceptions, so the placeholder used for ordinary attributes will not do here. Dropping the chain instead would silently lose the reason for the failure.

Source code in dataexcept/base.py

class UnpicklableCause(DataExceptError):
    """Stands in for a cause that could not be serialized.

    ``__cause__`` and ``__context__`` must be exceptions, so the placeholder
    used for ordinary attributes will not do here. Dropping the chain instead
    would silently lose the reason for the failure.
    """

UnpicklableValue

Stands in for state that could not survive serialization.

An exception carrying a lambda, an open file or a lock would otherwise be unraisable across a process boundary. Keeping a description preserves what the value was for debugging, which is the reason it was attached.

Source code in dataexcept/base.py

class UnpicklableValue:
    """Stands in for state that could not survive serialization.

    An exception carrying a lambda, an open file or a lock would otherwise be
    unraisable across a process boundary. Keeping a description preserves what
    the value was for debugging, which is the reason it was attached.
    """

    __slots__ = ("description",)

    def __init__(self, description: str) -> None:
        self.description = description

    def __repr__(self) -> str:
        return f"<unpicklable: {self.description}>"

    def __str__(self) -> str:
        return self.__repr__()

    def __eq__(self, other: object) -> bool:
        return (
            isinstance(other, UnpicklableValue)
            and other.description == self.description
        )

    def __hash__(self) -> int:
        return hash(self.description)

DatabaseConnectionError

Bases: DatabaseError

Raised when connecting to the database fails.

Source code in dataexcept/database_exceptions.py

class DatabaseConnectionError(DatabaseError):
    """Raised when connecting to the database fails."""

    def __init__(self, db_url: str, message: str | None = None) -> None:
        """Initialize DatabaseConnectionError.

        Args:
            db_url: Database connection URL.
            message: Optional custom error message.
        """
        # A connection URL routinely carries a username and password.
        self.db_url = redact_url(db_url)
        default = f"Failed to connect to database at '{self.db_url}'"
        super().__init__(message or default)

DatabaseError

Bases: DataExceptError

Base exception for database-related errors.

Source code in dataexcept/database_exceptions.py

class DatabaseError(DataExceptError):
    """Base exception for database-related errors."""

    pass

QueryExecutionError

Bases: DatabaseError

Raised when a database query execution fails.

Source code in dataexcept/database_exceptions.py

class QueryExecutionError(DatabaseError):
    """Raised when a database query execution fails."""

    def __init__(self, query: str, original: Exception | None = None) -> None:
        """Initialize QueryExecutionError.

        Args:
            query: SQL query string.
            original: Optional underlying exception.
        """
        self.query = query
        self.original = original
        msg = f"Query failed: {query}"
        if original:
            msg += f" ({original})"
        super().__init__(msg)

TransactionError

Bases: DatabaseError

Raised when a database transaction fails.

Source code in dataexcept/database_exceptions.py

class TransactionError(DatabaseError):
    """Raised when a database transaction fails."""

    def __init__(
        self,
        transaction_id: str | None = None,
        message: str | None = None,
    ) -> None:
        """Initialize TransactionError.

        Args:
            transaction_id: Identifier for the transaction.
            message: Optional custom error message.
        """
        self.transaction_id = transaction_id
        default = "Database transaction failed"
        if transaction_id:
            default += f" (id={transaction_id})"
        super().__init__(message or default)

BatchProcessingError

Bases: DataEngineeringError

Raised when processing a data batch fails.

Source code in dataexcept/dataengineering_exceptions.py

class BatchProcessingError(DataEngineeringError):
    """Raised when processing a data batch fails."""

    def __init__(self, batch_id: str, original: Optional[Exception] = None) -> None:
        """Initialize BatchProcessingError.

        Args:
            batch_id: Identifier of the batch being processed.
            original: Optional underlying exception.
        """
        self.batch_id = batch_id
        self.original = original
        msg = f"Batch '{batch_id}' processing failed"
        if original:
            msg += f": {original}"
        super().__init__(msg)

DataEngineeringError

Bases: DataExceptError

Base exception for data engineering errors.

Source code in dataexcept/dataengineering_exceptions.py

class DataEngineeringError(DataExceptError):
    """Base exception for data engineering errors."""

    pass

DataTransformationError

Bases: DataEngineeringError

Raised when a data transformation step fails.

Source code in dataexcept/dataengineering_exceptions.py

class DataTransformationError(DataEngineeringError):
    """Raised when a data transformation step fails."""

    def __init__(self, step: str, details: Optional[str] = None) -> None:
        """Initialize DataTransformationError.

        Args:
            step: Name of the transformation step.
            details: Optional details about the failure.
        """
        self.step = step
        self.details = details
        msg = f"Data transformation '{step}' failed"
        if details:
            msg += f": {details}"
        super().__init__(msg)

DataWarehouseConnectionError

Bases: DataEngineeringError

Raised when a connection to a data warehouse cannot be established.

Source code in dataexcept/dataengineering_exceptions.py

class DataWarehouseConnectionError(DataEngineeringError):
    """Raised when a connection to a data warehouse cannot be established."""

    def __init__(self, warehouse: str, message: Optional[str] = None) -> None:
        """Initialize DataWarehouseConnectionError.

        Args:
            warehouse: Identifier of the data warehouse.
            message: Optional custom error message.
        """
        self.warehouse = warehouse
        default = f"Failed to connect to warehouse '{warehouse}'"
        super().__init__(message or default)

ETLJobError

Bases: DataEngineeringError

Raised when an ETL job fails to complete successfully.

Source code in dataexcept/dataengineering_exceptions.py

class ETLJobError(DataEngineeringError):
    """Raised when an ETL job fails to complete successfully."""

    def __init__(self, job_name: str, message: Optional[str] = None) -> None:
        """Initialize ETLJobError.

        Args:
            job_name: Name of the ETL job.
            message: Optional custom error message.
        """
        self.job_name = job_name
        default = f"ETL job '{job_name}' failed"
        super().__init__(message or default)

MissingPartitionError

Bases: DataEngineeringError

Raised when a required data partition is missing.

Source code in dataexcept/dataengineering_exceptions.py

class MissingPartitionError(DataEngineeringError):
    """Raised when a required data partition is missing."""

    def __init__(
        self, partition: str, location: str, message: Optional[str] = None
    ) -> None:
        """Initialize MissingPartitionError.

        Args:
            partition: Name of the missing partition.
            location: Data location checked for the partition.
            message: Optional custom error message.
        """
        self.partition = partition
        self.location = redact_if_url(location)
        default = f"Partition '{partition}' not found at {location}"
        super().__init__(message or default)

SchemaEvolutionError

Bases: DataEngineeringError

Raised when database schema evolution fails.

Source code in dataexcept/dataengineering_exceptions.py

class SchemaEvolutionError(DataEngineeringError):
    """Raised when database schema evolution fails."""

    def __init__(self, schema_version: str, reason: Optional[str] = None) -> None:
        """Initialize SchemaEvolutionError.

        Args:
            schema_version: Version of the schema being applied.
            reason: Optional explanation of the failure.
        """
        self.schema_version = schema_version
        self.reason = reason
        msg = f"Schema evolution to {schema_version} failed"
        if reason:
            msg += f": {reason}"
        super().__init__(msg)

BiasDetectionError

Bases: DataScienceError

Raised when algorithmic bias exceeds an acceptable threshold.

Parameters:

Name	Type	Description	Default
feature	str	Feature or group where bias was detected.	required
bias_score	float	Calculated bias metric.	required
threshold	float	Maximum acceptable bias metric.	required
message	Optional[str]	Optional custom message.	None

Source code in dataexcept/datascience_exceptions/training.py

class BiasDetectionError(DataScienceError):
    """Raised when algorithmic bias exceeds an acceptable threshold.

    Args:
        feature: Feature or group where bias was detected.
        bias_score: Calculated bias metric.
        threshold: Maximum acceptable bias metric.
        message: Optional custom message.
    """

    def __init__(
        self,
        feature: str,
        bias_score: float,
        threshold: float,
        message: Optional[str] = None,
    ) -> None:
        if not isinstance(feature, str):
            raise TypeError(f"feature must be str, got {type(feature).__name__}")
        if not is_number(bias_score):
            raise TypeError(
                f"bias_score must be numeric, got {type(bias_score).__name__}"
            )
        if not is_number(threshold):
            raise TypeError(
                f"threshold must be numeric, got {type(threshold).__name__}"
            )
        if message is not None and not isinstance(message, str):
            raise TypeError(
                f"message must be str or None, got {type(message).__name__}"
            )

        if message is None:
            msg = (
                f"Bias detected in '{feature}': score={bias_score:.3f} > "
                f"threshold={threshold:.3f}"
            )
        else:
            msg = message

        self.feature = feature
        self.bias_score = float(bias_score)
        self.threshold = float(threshold)
        super().__init__(msg)

    def __str__(self) -> str:
        return f"[BiasDetectionError:{self.feature}] {self.message}"

ConvergenceError

Bases: ModelTrainingError

Raised when optimization fails to converge.

Attributes:

Name	Type	Description
iterations		number of iterations run.

Source code in dataexcept/datascience_exceptions/training.py

class ConvergenceError(ModelTrainingError):
    """
    Raised when optimization fails to converge.

    Attributes:
        iterations: number of iterations run.
    """

    def __init__(
        self, model_type: str, iterations: int, message: Optional[str] = None
    ) -> None:
        if not isinstance(iterations, int):
            raise TypeError(f"iterations must be int, got {type(iterations).__name__}")

        if message is None:
            message = (
                f"Model '{model_type}' failed to converge after "
                f"{iterations} iterations"
            )
        # Assigned before super(): DataExceptError.__init__ sweeps the stored
        # strings for URLs, and anything set afterwards escapes that.
        self.iterations = iterations
        super().__init__(model_type=model_type, epoch=None, message=message)

    def __str__(self) -> str:
        return f"[ConvergenceError] {self.message}"

CrossValidationError

Bases: DataScienceError

Failure during cross-validation procedure.

Source code in dataexcept/datascience_exceptions/training.py

class CrossValidationError(DataScienceError):
    """Failure during cross-validation procedure."""

    def __init__(self, folds: int, cause: Optional[str] = None) -> None:
        if not isinstance(folds, int):
            raise TypeError(f"folds must be int, got {type(folds).__name__}")
        msg = f"Cross-validation failed on {folds} folds" + (
            f": {cause}" if cause else ""
        )
        self.folds = folds
        super().__init__(msg)

DataAugmentationError

Bases: DataScienceError

Raised when a data augmentation technique fails.

Parameters:

Name	Type	Description	Default
technique	str	Name of the augmentation technique.	required
details	Optional[str]	Optional explanation of the failure.	None

Source code in dataexcept/datascience_exceptions/ingestion.py

class DataAugmentationError(DataScienceError):
    """Raised when a data augmentation technique fails.

    Args:
        technique: Name of the augmentation technique.
        details: Optional explanation of the failure.
    """

    def __init__(self, technique: str, details: Optional[str] = None) -> None:
        if not isinstance(technique, str):
            raise TypeError(f"technique must be str, got {type(technique).__name__}")
        if details is not None and not isinstance(details, str):
            raise TypeError(
                f"details must be str or None, got {type(details).__name__}"
            )

        msg = f"Data augmentation '{technique}' failed"
        if details:
            msg += f": {details}"

        self.technique = technique
        self.details = details
        super().__init__(msg)

    def __str__(self) -> str:
        return f"[DataAugmentationError:{self.technique}] {self.message}"

DataDriftError

Bases: DataScienceError

Raised when data drift is detected beyond threshold.

Attributes:

Name	Type	Description
feature		feature name.
drift_score		computed drift metric.

Source code in dataexcept/datascience_exceptions/operations.py

class DataDriftError(DataScienceError):
    """
    Raised when data drift is detected beyond threshold.

    Attributes:
        feature: feature name.
        drift_score: computed drift metric.
    """

    def __init__(
        self, feature: str, drift_score: float, message: Optional[str] = None
    ) -> None:
        if not isinstance(feature, str):
            raise TypeError(f"feature must be str, got {type(feature).__name__}")
        if not is_number(drift_score):
            raise TypeError(
                f"drift_score must be number, got {type(drift_score).__name__}"
            )

        self.feature = feature
        self.drift_score = float(drift_score)
        if message is None:
            message = f"Data drift detected on '{feature}', score={drift_score:.4f}"

        super().__init__(message)

    def __str__(self) -> str:
        return f"[DataDriftError:{self.feature}] {self.message}"

DataExportError

Bases: DataScienceError

Failed to export or write data to destination.

Source code in dataexcept/datascience_exceptions/operations.py

class DataExportError(DataScienceError):
    """Failed to export or write data to destination."""

    def __init__(self, destination: str, original: Exception) -> None:
        if not isinstance(destination, str):
            raise TypeError(
                f"destination must be str, got {type(destination).__name__}"
            )
        if not isinstance(original, Exception):
            raise TypeError(
                f"original must be Exception, got {type(original).__name__}"
            )
        msg = f"Unable to export data to {destination}: {original}"
        self.destination = destination
        self.original = original
        super().__init__(msg)

DataFormatError

Bases: DataScienceError

Raised when input data is not in the expected format.

Source code in dataexcept/datascience_exceptions/ingestion.py

class DataFormatError(DataScienceError):
    """Raised when input data is not in the expected format."""

    def __init__(self, expected_formats: Sequence[str], found_format: str) -> None:
        if not isinstance(found_format, str):
            raise TypeError(
                f"found_format must be str, got {type(found_format).__name__}"
            )
        if not isinstance(expected_formats, Sequence) or isinstance(
            expected_formats, str
        ):
            raise TypeError("expected_formats must be a sequence of strings")
        if not all(isinstance(fmt, str) for fmt in expected_formats):
            raise TypeError("expected_formats must contain strings")

        self.expected_formats = list(expected_formats)
        self.found_format = found_format
        fmt_list = ", ".join(self.expected_formats)
        message = f"Expected data format {fmt_list}; got {found_format}"
        super().__init__(message)

    def __str__(self) -> str:
        return f"[DataFormatError] {self.message}"

DataImbalanceError

Bases: DataScienceError

Raised when class distribution is too imbalanced.

Parameters:

Name	Type	Description	Default
ratio	float	Observed minority-to-majority ratio.	required
threshold	float	Minimum acceptable ratio.	required
message	Optional[str]	Optional custom error message.	None

Source code in dataexcept/datascience_exceptions/ingestion.py

class DataImbalanceError(DataScienceError):
    """Raised when class distribution is too imbalanced.

    Args:
        ratio: Observed minority-to-majority ratio.
        threshold: Minimum acceptable ratio.
        message: Optional custom error message.
    """

    def __init__(
        self, ratio: float, threshold: float, message: Optional[str] = None
    ) -> None:
        if not is_number(ratio):
            raise TypeError(f"ratio must be numeric, got {type(ratio).__name__}")
        if not is_number(threshold):
            raise TypeError(
                f"threshold must be numeric, got {type(threshold).__name__}"
            )
        if message is not None and not isinstance(message, str):
            raise TypeError(
                f"message must be str or None, got {type(message).__name__}"
            )
        self.ratio = float(ratio)
        self.threshold = float(threshold)
        if message is None:
            msg = (
                f"Data imbalance detected: ratio={self.ratio:.3f} < "
                f"threshold={self.threshold:.3f}"
            )
        else:
            msg = message
        super().__init__(msg)

    def __str__(self) -> str:
        return f"[DataImbalanceError] {self.message}"

DataLeakageError

Bases: DataScienceError

Raised when data leakage is detected between train and test sets.

Parameters:

Name	Type	Description	Default
feature	str	Name of the leaked feature.	required
stage	str	Stage where the leakage occurred.	required
message	Optional[str]	Optional custom message.	None

Source code in dataexcept/datascience_exceptions/ingestion.py

class DataLeakageError(DataScienceError):
    """Raised when data leakage is detected between train and test sets.

    Args:
        feature: Name of the leaked feature.
        stage: Stage where the leakage occurred.
        message: Optional custom message.
    """

    def __init__(self, feature: str, stage: str, message: Optional[str] = None) -> None:
        if not isinstance(feature, str):
            raise TypeError(f"feature must be str, got {type(feature).__name__}")
        if not isinstance(stage, str):
            raise TypeError(f"stage must be str, got {type(stage).__name__}")
        if message is not None and not isinstance(message, str):
            raise TypeError(
                f"message must be str or None, got {type(message).__name__}"
            )

        if message is None:
            msg = f"Data leakage detected for '{feature}' during {stage}"
        else:
            msg = message

        self.feature = feature
        self.stage = stage
        super().__init__(msg)

    def __str__(self) -> str:
        return f"[DataLeakageError:{self.feature}] {self.message}"

DataLoadingError

Bases: DataScienceError

Raised when loading data fails.

Attributes:

Name	Type	Description
source		data source description (file path, URL).
original		underlying exception.

Source code in dataexcept/datascience_exceptions/ingestion.py

class DataLoadingError(DataScienceError):
    """
    Raised when loading data fails.

    Attributes:
        source: data source description (file path, URL).
        original: underlying exception.
    """

    def __init__(self, source: str, original: Exception) -> None:
        if not isinstance(source, str):
            raise TypeError(f"source must be str, got {type(source).__name__}")
        if not isinstance(original, Exception):
            raise TypeError(
                f"original must be Exception, got {type(original).__name__}"
            )

        message = f"Failed to load data from {source!r}: {original}"
        self.source = redact_if_url(source)
        self.original = original
        super().__init__(message)

    def __str__(self) -> str:
        return f"[DataLoadingError:{self.source}] {self.message}"

DataNormalizationError

Bases: DataScienceError

Raised when data normalization fails.

Parameters:

Name	Type	Description	Default
method	str	Normalization technique identifier.	required
details	Optional[str]	Optional explanation of the failure.	None

Source code in dataexcept/datascience_exceptions/ingestion.py

class DataNormalizationError(DataScienceError):
    """Raised when data normalization fails.

    Args:
        method: Normalization technique identifier.
        details: Optional explanation of the failure.
    """

    def __init__(self, method: str, details: Optional[str] = None) -> None:
        if not isinstance(method, str):
            raise TypeError(f"method must be str, got {type(method).__name__}")
        if details is not None and not isinstance(details, str):
            raise TypeError(
                f"details must be str or None, got {type(details).__name__}"
            )
        # Build a helpful error message
        msg = f"Normalization using '{method}' failed"
        if details:
            msg += f": {details}"
        self.method = method
        self.details = details
        super().__init__(msg)

    def __str__(self) -> str:
        return f"[DataNormalizationError:{self.method}] {self.message}"

DataScienceError

Bases: DataExceptError

Base exception for data science errors.

Source code in dataexcept/datascience_exceptions/base.py

class DataScienceError(DataExceptError):
    """Base exception for data science errors."""

    def __init__(self, message: str) -> None:
        # Ensure message is a string
        if not isinstance(message, str):
            raise TypeError(f"message must be str, got {type(message).__name__}")
        self.message = message
        super().__init__(message)

    def __str__(self) -> str:
        return f"[DataScienceError] {self.message}"

DataValidationError

Bases: DataScienceError

Raised when data fails validation rules.

Attributes:

Name	Type	Description
field		name of invalid field.
value		the invalid value.

Source code in dataexcept/datascience_exceptions/ingestion.py

class DataValidationError(DataScienceError):
    """
    Raised when data fails validation rules.

    Attributes:
        field: name of invalid field.
        value: the invalid value.
    """

    def __init__(self, field: str, value: Any, message: Optional[str] = None) -> None:
        if not isinstance(field, str):
            raise TypeError(f"field must be str, got {type(field).__name__}")

        if message is None:
            message = f"Invalid value for '{field}': {value!r}"
        elif not isinstance(message, str):
            raise TypeError(f"message must be str, got {type(message).__name__}")

        self.field = field
        self.value = value
        super().__init__(message)

    def __str__(self) -> str:
        return f"[DataValidationError:{self.field}] {self.message}"

DeploymentError

Bases: DataScienceError

Raised when deploying a model or pipeline fails.

Attributes:

Name	Type	Description
target		deployment target identifier.
cause		optional detail.

Source code in dataexcept/datascience_exceptions/operations.py

class DeploymentError(DataScienceError):
    """
    Raised when deploying a model or pipeline fails.

    Attributes:
        target: deployment target identifier.
        cause: optional detail.
    """

    def __init__(self, target: str, cause: Optional[str] = None) -> None:
        if not isinstance(target, str):
            raise TypeError(f"target must be str, got {type(target).__name__}")
        if cause is not None and not isinstance(cause, str):
            raise TypeError(f"cause must be str or None, got {type(cause).__name__}")

        msg = f"Deployment failed to '{target}'"
        if cause:
            msg += f": {cause}"

        self.target = target
        self.cause = cause
        super().__init__(msg)

    def __str__(self) -> str:
        return f"[DeploymentError:{self.target}] {self.message}"

DimensionalityReductionError

Bases: DataScienceError

Error applying dimensionality reduction method.

Source code in dataexcept/datascience_exceptions/training.py

class DimensionalityReductionError(DataScienceError):
    """Error applying dimensionality reduction method."""

    def __init__(self, method: str, components: Optional[int] = None) -> None:
        if not isinstance(method, str):
            raise TypeError(f"method must be str, got {type(method).__name__}")
        if components is not None and not isinstance(components, int):
            raise TypeError(
                ("components must be int or None, " f"got {type(components).__name__}")
            )
        msg = f"Dimensionality reduction '{method}' failed" + (
            f" for {components} components" if components else ""
        )
        self.method = method
        self.components = components
        super().__init__(msg)

EarlyStoppingError

Bases: DataScienceError

Raised when training stops early based on a stopping criterion.

Parameters:

Name	Type	Description	Default
epoch	int	Epoch index where training stopped.	required
reason	Optional[str]	Optional reason for stopping.	None

Source code in dataexcept/datascience_exceptions/training.py

class EarlyStoppingError(DataScienceError):
    """Raised when training stops early based on a stopping criterion.

    Args:
        epoch: Epoch index where training stopped.
        reason: Optional reason for stopping.
    """

    def __init__(self, epoch: int, reason: Optional[str] = None) -> None:
        if not isinstance(epoch, int):
            raise TypeError(f"epoch must be int, got {type(epoch).__name__}")
        if reason is not None and not isinstance(reason, str):
            raise TypeError(f"reason must be str or None, got {type(reason).__name__}")

        msg = f"Training stopped early at epoch {epoch}"
        if reason:
            msg += f": {reason}"

        self.epoch = epoch
        self.reason = reason
        super().__init__(msg)

    def __str__(self) -> str:
        return f"[EarlyStoppingError:{self.epoch}] {self.message}"

ExperimentTrackingError

Bases: DataScienceError

Issues logging or retrieving experiment metadata.

Source code in dataexcept/datascience_exceptions/training.py

class ExperimentTrackingError(DataScienceError):
    """Issues logging or retrieving experiment metadata."""

    def __init__(self, run_id: str, cause: Optional[str] = None) -> None:
        if not isinstance(run_id, str):
            raise TypeError(f"run_id must be str, got {type(run_id).__name__}")
        msg = f"Experiment tracking failed for run '{run_id}'" + (
            f": {cause}" if cause else ""
        )
        self.run_id = run_id
        super().__init__(msg)

ExplainabilityError

Bases: DataScienceError

Raised when generating model explanations fails.

Parameters:

Name	Type	Description	Default
method	str	Explanation technique identifier.	required
details	Optional[str]	Optional description of the failure.	None

Source code in dataexcept/datascience_exceptions/training.py

class ExplainabilityError(DataScienceError):
    """Raised when generating model explanations fails.

    Args:
        method: Explanation technique identifier.
        details: Optional description of the failure.
    """

    def __init__(self, method: str, details: Optional[str] = None) -> None:
        if not isinstance(method, str):
            raise TypeError(f"method must be str, got {type(method).__name__}")
        if details is not None and not isinstance(details, str):
            raise TypeError(
                f"details must be str or None, got {type(details).__name__}"
            )
        msg = f"Explainability using '{method}' failed"
        if details:
            msg += f": {details}"
        self.method = method
        self.details = details
        super().__init__(msg)

    def __str__(self) -> str:
        return f"[ExplainabilityError:{self.method}] {self.message}"

FeatureEngineeringError

Bases: DataScienceError

Raised during feature engineering steps.

Attributes:

Name	Type	Description
step		description of the step that failed.
cause		optional underlying reason.

Source code in dataexcept/datascience_exceptions/ingestion.py

class FeatureEngineeringError(DataScienceError):
    """
    Raised during feature engineering steps.

    Attributes:
        step: description of the step that failed.
        cause: optional underlying reason.
    """

    def __init__(self, step: str, cause: Optional[str] = None) -> None:
        if not isinstance(step, str):
            raise TypeError(f"step must be str, got {type(step).__name__}")
        if cause is not None and not isinstance(cause, str):
            raise TypeError(f"cause must be str or None, got {type(cause).__name__}")

        msg = f"Feature engineering failed at step '{step}'"
        if cause:
            msg += f": {cause}"

        self.step = step
        self.cause = cause
        super().__init__(msg)

    def __str__(self) -> str:
        return f"[FeatureEngineeringError] {self.message}"

FeatureScalingError

Bases: DataScienceError

Raised when scaling or standardization of features fails.

Parameters:

Name	Type	Description	Default
scaler	str	Name of the scaler or transformation used.	required
details	Optional[str]	Optional explanation of the failure.	None

Source code in dataexcept/datascience_exceptions/training.py

class FeatureScalingError(DataScienceError):
    """Raised when scaling or standardization of features fails.

    Args:
        scaler: Name of the scaler or transformation used.
        details: Optional explanation of the failure.
    """

    def __init__(self, scaler: str, details: Optional[str] = None) -> None:
        if not isinstance(scaler, str):
            raise TypeError(f"scaler must be str, got {type(scaler).__name__}")
        if details is not None and not isinstance(details, str):
            raise TypeError(
                f"details must be str or None, got {type(details).__name__}"
            )
        msg = f"Feature scaling with '{scaler}' failed"
        if details:
            msg += f": {details}"
        self.scaler = scaler
        self.details = details
        super().__init__(msg)

    def __str__(self) -> str:
        return f"[FeatureScalingError:{self.scaler}] {self.message}"

FeatureSelectionError

Bases: DataScienceError

Failure in feature selection procedure.

Source code in dataexcept/datascience_exceptions/training.py

class FeatureSelectionError(DataScienceError):
    """Failure in feature selection procedure."""

    def __init__(self, technique: str, details: Optional[str] = None) -> None:
        if not isinstance(technique, str):
            raise TypeError(f"technique must be str, got {type(technique).__name__}")
        msg = f"Feature selection failed using {technique}" + (
            f": {details}" if details else ""
        )
        self.technique = technique
        super().__init__(msg)

GPUOutOfMemoryError

Bases: DataScienceError

Model or tensor exceeds GPU memory capacity.

Source code in dataexcept/datascience_exceptions/training.py

class GPUOutOfMemoryError(DataScienceError):
    """Model or tensor exceeds GPU memory capacity."""

    def __init__(self, device: str, required: str, available: str) -> None:
        if not all(isinstance(v, str) for v in (device, required, available)):
            raise TypeError("device, required, available must be str")
        msg = f"GPU OOM on {device}: required={required}, available={available}"
        self.device = device
        self.required = required
        self.available = available
        super().__init__(msg)

HyperparameterError

Bases: DataScienceError

Raised for invalid hyperparameter settings.

Attributes:

Name	Type	Description
param		name of hyperparameter.
value		invalid value.

Source code in dataexcept/datascience_exceptions/training.py

class HyperparameterError(DataScienceError):
    """
    Raised for invalid hyperparameter settings.

    Attributes:
        param: name of hyperparameter.
        value: invalid value.
    """

    def __init__(self, param: str, value: Any, message: Optional[str] = None) -> None:
        if not isinstance(param, str):
            raise TypeError(f"param must be str, got {type(param).__name__}")

        if message is None:
            message = f"Invalid hyperparameter '{param}': {value!r}"

        self.param = param
        self.value = value
        super().__init__(message)

    def __str__(self) -> str:
        return f"[HyperparameterError:{self.param}] {self.message}"

HyperparameterTuningError

Bases: DataScienceError

Error during hyperparameter search or tuning.

Source code in dataexcept/datascience_exceptions/training.py

class HyperparameterTuningError(DataScienceError):
    """Error during hyperparameter search or tuning."""

    def __init__(self, method: str, details: Optional[str] = None) -> None:
        if not isinstance(method, str):
            raise TypeError(f"method must be str, got {type(method).__name__}")
        msg = f"Hyperparameter tuning ({method}) failed" + (
            f": {details}" if details else ""
        )
        self.method = method
        super().__init__(msg)

MissingDataError

Bases: DataScienceError

Raised when required data is missing.

Attributes:

Name	Type	Description
feature		name of missing feature.

Source code in dataexcept/datascience_exceptions/ingestion.py

class MissingDataError(DataScienceError):
    """
    Raised when required data is missing.

    Attributes:
        feature: name of missing feature.
    """

    def __init__(self, feature: str, message: Optional[str] = None) -> None:
        if not isinstance(feature, str):
            raise TypeError(f"feature must be str, got {type(feature).__name__}")

        if message is None:
            message = f"Missing required feature: {feature!r}"
        elif not isinstance(message, str):
            raise TypeError(f"message must be str, got {type(message).__name__}")

        self.feature = feature
        super().__init__(message)

    def __str__(self) -> str:
        return f"[MissingDataError:{self.feature}] {self.message}"

ModelCompatibilityError

Bases: DataScienceError

Raised when a model is incompatible with the runtime environment.

Parameters:

Name	Type	Description	Default
expected_version	str	Required model version.	required
found_version	str	Detected model version.	required
message	Optional[str]	Optional custom message.	None

Source code in dataexcept/datascience_exceptions/training.py

class ModelCompatibilityError(DataScienceError):
    """Raised when a model is incompatible with the runtime environment.

    Args:
        expected_version: Required model version.
        found_version: Detected model version.
        message: Optional custom message.
    """

    def __init__(
        self,
        expected_version: str,
        found_version: str,
        message: Optional[str] = None,
    ) -> None:
        if not isinstance(expected_version, str):
            raise TypeError(
                "expected_version must be str, got "
                f"{type(expected_version).__name__}"
            )
        if not isinstance(found_version, str):
            raise TypeError(
                "found_version must be str, got " f"{type(found_version).__name__}"
            )
        if message is not None and not isinstance(message, str):
            raise TypeError(
                f"message must be str or None, got {type(message).__name__}"
            )

        if message is None:
            msg = (
                f"Model requires version {expected_version}, "
                f"but found {found_version}"
            )
        else:
            msg = message

        self.expected_version = expected_version
        self.found_version = found_version
        super().__init__(msg)

    def __str__(self) -> str:
        return f"[ModelCompatibilityError] {self.message}"

ModelEvaluationError

Bases: DataScienceError

Raised during evaluation metrics computation.

Attributes:

Name	Type	Description
metric		name of the metric.
value		computed value.

Source code in dataexcept/datascience_exceptions/training.py

class ModelEvaluationError(DataScienceError):
    """
    Raised during evaluation metrics computation.

    Attributes:
        metric: name of the metric.
        value: computed value.
    """

    def __init__(
        self, metric: str, value: float, message: Optional[str] = None
    ) -> None:
        if not isinstance(metric, str):
            raise TypeError(f"metric must be str, got {type(metric).__name__}")
        if not is_number(value):
            raise TypeError(f"value must be number, got {type(value).__name__}")

        if message is None:
            message = f"Failed to compute metric '{metric}', got {value}"

        self.metric = metric
        self.value = float(value)
        super().__init__(message)

    def __str__(self) -> str:
        return f"[ModelEvaluationError:{self.metric}] {self.message}"

ModelInferenceError

Bases: DataScienceError

Raised when model inference fails.

Parameters:

Name	Type	Description	Default
model_type	str	Identifier of the model used for inference.	required
original	Exception	Underlying exception raised by the model.	required

Source code in dataexcept/datascience_exceptions/training.py

class ModelInferenceError(DataScienceError):
    """Raised when model inference fails.

    Args:
        model_type: Identifier of the model used for inference.
        original: Underlying exception raised by the model.
    """

    def __init__(self, model_type: str, original: Exception) -> None:
        if not isinstance(model_type, str):
            raise TypeError(f"model_type must be str, got {type(model_type).__name__}")
        if not isinstance(original, Exception):
            raise TypeError(
                f"original must be Exception, got {type(original).__name__}"
            )
        msg = f"Inference failed for model '{model_type}': {original}"
        self.model_type = model_type
        self.original = original
        super().__init__(msg)

    def __str__(self) -> str:
        return f"[ModelInferenceError:{self.model_type}] {self.message}"

ModelSerializationError

Bases: DataScienceError

Raised when saving or loading a model fails.

Attributes:

Name	Type	Description
path		file path involved.
original		underlying exception.

Source code in dataexcept/datascience_exceptions/operations.py

class ModelSerializationError(DataScienceError):
    """
    Raised when saving or loading a model fails.

    Attributes:
        path: file path involved.
        original: underlying exception.
    """

    def __init__(self, path: str, original: Exception) -> None:
        if not isinstance(path, str):
            raise TypeError(f"path must be str, got {type(path).__name__}")
        if not isinstance(original, Exception):
            raise TypeError(
                f"original must be Exception, got {type(original).__name__}"
            )

        message = f"Failed to serialize to {path!r}: {original}"
        self.path = redact_if_url(path)
        self.original = original
        super().__init__(message)

    def __str__(self) -> str:
        return f"[ModelSerializationError:{self.path}] {self.message}"

ModelTrainingError

Bases: DataScienceError

Raised when model training fails.

Attributes:

Name	Type	Description
model_type		model class or name.
epoch		optional epoch index.

Source code in dataexcept/datascience_exceptions/training.py

class ModelTrainingError(DataScienceError):
    """
    Raised when model training fails.

    Attributes:
        model_type: model class or name.
        epoch: optional epoch index.
    """

    def __init__(
        self,
        model_type: str,
        epoch: Optional[int] = None,
        message: Optional[str] = None,
    ) -> None:
        if not isinstance(model_type, str):
            raise TypeError(f"model_type must be str, got {type(model_type).__name__}")
        if epoch is not None and not isinstance(epoch, int):
            raise TypeError(f"epoch must be int or None, got {type(epoch).__name__}")
        if message is not None and not isinstance(message, str):
            raise TypeError(
                f"message must be str or None, got {type(message).__name__}"
            )

        if message is None:
            msg = f"Training failed for model '{model_type}'"
            if epoch is not None:
                msg += f" at epoch {epoch}"  # include epoch
        else:
            msg = message

        self.model_type = model_type
        self.epoch = epoch
        super().__init__(msg)

    def __str__(self) -> str:
        base = f"{self.model_type}"
        if self.epoch is not None:
            base += f"@{self.epoch}"
        return f"[ModelTrainingError:{base}] {self.message}"

OutlierDetectionError

Bases: DataScienceError

Raised when outlier detection fails.

Attributes:

Name	Type	Description
method		detection method name.
details		optional extra info.

Source code in dataexcept/datascience_exceptions/ingestion.py

class OutlierDetectionError(DataScienceError):
    """
    Raised when outlier detection fails.

    Attributes:
        method: detection method name.
        details: optional extra info.
    """

    def __init__(self, method: str, details: Optional[str] = None) -> None:
        if not isinstance(method, str):
            raise TypeError(f"method must be str, got {type(method).__name__}")
        if details is not None and not isinstance(details, str):
            raise TypeError(
                f"details must be str or None, got {type(details).__name__}"
            )

        msg = f"Outlier detection failed using method '{method}'"
        if details:
            msg += f": {details}"

        self.method = method
        self.details = details
        super().__init__(msg)

    def __str__(self) -> str:
        return f"[OutlierDetectionError:{self.method}] {self.message}"

OverfittingError

Bases: DataScienceError

Raised when a model is overfitting the training data.

Parameters:

Name	Type	Description	Default
train_metric	float	Metric value on the training set.	required
val_metric	float	Metric value on the validation set.	required

Source code in dataexcept/datascience_exceptions/training.py

class OverfittingError(DataScienceError):
    """Raised when a model is overfitting the training data.

    Args:
        train_metric: Metric value on the training set.
        val_metric: Metric value on the validation set.
    """

    def __init__(self, train_metric: float, val_metric: float) -> None:
        if not is_number(train_metric):
            raise TypeError(
                ("train_metric must be numeric, got " f"{type(train_metric).__name__}")
            )
        if not is_number(val_metric):
            raise TypeError(
                f"val_metric must be numeric, got {type(val_metric).__name__}"
            )

        self.train_metric = float(train_metric)
        self.val_metric = float(val_metric)
        msg = (
            f"Overfitting detected: train={self.train_metric}, "
            f"val={self.val_metric}"
        )
        super().__init__(msg)

    def __str__(self) -> str:
        return f"[OverfittingError] {self.message}"

PredictionError

Bases: DataScienceError

Raised when making predictions fails.

Attributes:

Name	Type	Description
model_type		model used.
inputs		input data snapshot.

Source code in dataexcept/datascience_exceptions/training.py

class PredictionError(DataScienceError):
    """
    Raised when making predictions fails.

    Attributes:
        model_type: model used.
        inputs: input data snapshot.
    """

    def __init__(
        self, model_type: str, inputs: Any, message: Optional[str] = None
    ) -> None:
        if not isinstance(model_type, str):
            raise TypeError(f"model_type must be str, got {type(model_type).__name__}")

        if message is None:
            message = (
                f"Prediction failed for model '{model_type}' " f"with inputs {inputs!r}"
            )

        self.model_type = model_type
        self.inputs = inputs
        super().__init__(message)

    def __str__(self) -> str:
        return f"[PredictionError:{self.model_type}] {self.message}"

ResourceLimitError

Bases: DataScienceError

Raised when computation exceeds resources (memory, CPU).

Attributes:

Name	Type	Description
resource		'memory', 'cpu', etc.
limit		threshold exceeded.

Source code in dataexcept/datascience_exceptions/operations.py

class ResourceLimitError(DataScienceError):
    """
    Raised when computation exceeds resources (memory, CPU).

    Attributes:
        resource: 'memory', 'cpu', etc.
        limit: threshold exceeded.
    """

    def __init__(self, resource: str, limit: Any) -> None:
        if not isinstance(resource, str):
            raise TypeError(f"resource must be str, got {type(resource).__name__}")

        message = f"Resource limit exceeded: {resource} at {limit!r}"
        self.resource = resource
        self.limit = limit
        super().__init__(message)

    def __str__(self) -> str:
        return f"[ResourceLimitError:{self.resource}] {self.message}"

SchemaMismatchError

Bases: DataScienceError

Raised when data schema does not match expected.

Attributes:

Name	Type	Description
expected		expected schema description.
found		actual schema description.

Source code in dataexcept/datascience_exceptions/ingestion.py

class SchemaMismatchError(DataScienceError):
    """
    Raised when data schema does not match expected.

    Attributes:
        expected: expected schema description.
        found: actual schema description.
    """

    def __init__(self, expected: str, found: str) -> None:
        if not isinstance(expected, str):
            raise TypeError(f"expected must be str, got {type(expected).__name__}")
        if not isinstance(found, str):
            raise TypeError(f"found must be str, got {type(found).__name__}")

        message = f"Schema mismatch. Expected: {expected}, Found: {found}"
        self.expected = expected
        self.found = found
        super().__init__(message)

    def __str__(self) -> str:
        return f"[SchemaMismatchError] {self.message}"

TrainingTimeoutError

Bases: ModelTrainingError

Raised when model training exceeds a time limit.

Source code in dataexcept/datascience_exceptions/training.py

class TrainingTimeoutError(ModelTrainingError):
    """Raised when model training exceeds a time limit."""

    def __init__(self, model_type: str, timeout: float) -> None:
        if not is_number(timeout):
            raise TypeError(f"timeout must be a number, got {type(timeout).__name__}")
        message = f"Training '{model_type}' exceeded timeout of {timeout} seconds"
        self.timeout = float(timeout)
        super().__init__(model_type=model_type, epoch=None, message=message)

    def __str__(self) -> str:
        return f"[TrainingTimeoutError] {self.message}"

UnderfittingError

Bases: DataScienceError

Raised when a model fails to capture patterns in the data.

Parameters:

Name	Type	Description	Default
train_metric	float	Metric value on the training set.	required
threshold	float	Minimum acceptable metric value.	required

Source code in dataexcept/datascience_exceptions/training.py

class UnderfittingError(DataScienceError):
    """Raised when a model fails to capture patterns in the data.

    Args:
        train_metric: Metric value on the training set.
        threshold: Minimum acceptable metric value.
    """

    def __init__(self, train_metric: float, threshold: float) -> None:
        if not is_number(train_metric):
            raise TypeError(
                ("train_metric must be numeric, got " f"{type(train_metric).__name__}")
            )
        if not is_number(threshold):
            raise TypeError(
                f"threshold must be numeric, got {type(threshold).__name__}"
            )

        self.train_metric = float(train_metric)
        self.threshold = float(threshold)
        msg = (
            f"Underfitting detected: training metric {self.train_metric} "
            f"< threshold {self.threshold}"
        )
        super().__init__(msg)

    def __str__(self) -> str:
        return f"[UnderfittingError] {self.message}"

AuthenticationError

Bases: JobError

Raised when user authentication fails.

Source code in dataexcept/exceptions/authentication.py

class AuthenticationError(JobError):
    """Raised when user authentication fails."""

    def __init__(self, user: str, message: str | None = None):
        self.user = user
        self.message = message or f"Authentication failed for user '{user}'"
        super().__init__(self.message)

AuthorizationError

Bases: JobError

Raised when user lacks permission for an action.

Source code in dataexcept/exceptions/authentication.py

class AuthorizationError(JobError):
    """Raised when user lacks permission for an action."""

    def __init__(self, user: str, permission: str):
        self.user = user
        self.permission = permission
        msg = f"User '{user}' lacks permission '{permission}'"
        super().__init__(msg)

ConfigurationError

Bases: JobError

Raised when there is a problem with configuration or settings.

Source code in dataexcept/exceptions/configuration.py

class ConfigurationError(JobError):
    """Raised when there is a problem with configuration or settings."""

    def __init__(self, option: str, message: str | None = None):
        self.option = option
        self.message = message or f"Invalid configuration for '{option}'"
        super().__init__(self.message)

CronExpressionError

Bases: JobError

Raised when a cron expression is invalid.

Source code in dataexcept/exceptions/scheduling.py

class CronExpressionError(JobError):
    """Raised when a cron expression is invalid."""

    def __init__(self, expression: str, message: str | None = None):
        self.expression = expression
        self.message = message or f"Invalid cron expression: '{expression}'"
        super().__init__(self.message)

DependencyError

Bases: JobError

Raised when a job dependency is missing or fails.

Source code in dataexcept/exceptions/external.py

class DependencyError(JobError):
    """Raised when a job dependency is missing or fails."""

    def __init__(self, dependency_name: str, message: str | None = None):
        self.dependency_name = dependency_name
        self.message = message or f"Dependency '{dependency_name}' error"
        super().__init__(self.message)

DeserializationError

Bases: JobError

Raised when deserialization of data fails.

Source code in dataexcept/exceptions/parsing.py

class DeserializationError(JobError):
    """Raised when deserialization of data fails."""

    def __init__(self, data: bytes, format: str, message: str | None = None):
        self.data = data
        self.format = format
        self.message = message or f"Failed to deserialize data from {format}"
        super().__init__(self.message)

EmailError

Bases: NotificationError

Raised when sending an email fails.

Source code in dataexcept/exceptions/notification.py

class EmailError(NotificationError):
    """Raised when sending an email fails."""

    def __init__(
        self,
        recipient: str,
        subject: str,
        original_exception: Exception | None = None,
    ):
        self.recipient = recipient
        self.subject = subject
        # original_exception is set by NotificationError.__init__ below.
        msg = f"Email to '{recipient}' with subject '{subject}' failed"
        if original_exception:
            msg += f": {original_exception}"
        super().__init__("email", original_exception, message=msg)

JobCancellationError

Bases: JobError

Raised when a job is cancelled before completion.

Source code in dataexcept/exceptions/lifecycle.py

class JobCancellationError(JobError):
    """Raised when a job is cancelled before completion."""

    def __init__(self, job_id: str, reason: str | None = None):
        self.job_id = job_id
        self.reason = reason
        msg = f"Job '{job_id}' was cancelled"
        if reason:
            msg += f": {reason}"
        super().__init__(msg)

JobError

Bases: DataExceptError

Base exception for all job-related errors.

Source code in dataexcept/exceptions/base.py

class JobError(DataExceptError):
    """Base exception for all job-related errors."""

    pass

NotificationError

Bases: JobError

Base exception for notification failures.

Source code in dataexcept/exceptions/notification.py

class NotificationError(JobError):
    """Base exception for notification failures."""

    def __init__(
        self,
        channel: str,
        original_exception: Exception | None = None,
        message: str | None = None,
    ):
        self.channel = channel
        self.original_exception = original_exception
        msg = message or f"Notification via '{channel}' failed"
        if message is None and original_exception:
            msg += f": {original_exception}"
        super().__init__(msg)

OperationTimeoutError

Bases: JobError

Raised when an operation exceeds its time limit.

Source code in dataexcept/exceptions/external.py

class OperationTimeoutError(JobError):
    """Raised when an operation exceeds its time limit."""

    def __init__(self, operation: str, timeout: float):
        self.operation = operation
        self.timeout = timeout
        msg = f"Operation '{operation}' timed out after {timeout} seconds"
        super().__init__(msg)

ParsingError

Bases: JobError

Raised when parsing of input data fails.

Source code in dataexcept/exceptions/parsing.py

class ParsingError(JobError):
    """Raised when parsing of input data fails."""

    def __init__(self, text: str, message: str | None = None):
        self.text = text
        self.message = message or f"Failed to parse text: {text!r}"
        super().__init__(self.message)

ResourceNotFoundError

Bases: JobError

Raised when a required resource cannot be found.

Source code in dataexcept/exceptions/external.py

class ResourceNotFoundError(JobError):
    """Raised when a required resource cannot be found."""

    def __init__(self, resource_type: str, identifier: str):
        self.resource_type = resource_type
        self.identifier = identifier
        msg = f"{resource_type} with identifier '{identifier}' not found"
        super().__init__(msg)

ScheduleConflictError

Bases: JobError

Raised when two jobs have conflicting schedules.

Source code in dataexcept/exceptions/scheduling.py

class ScheduleConflictError(JobError):
    """Raised when two jobs have conflicting schedules."""

    def __init__(self, job_name: str, schedule: str):
        self.job_name = job_name
        self.schedule = schedule
        msg = f"Schedule conflict for job '{job_name}' on schedule '{schedule}'"
        super().__init__(msg)

SerializationError

Bases: JobError

Raised when serialization of an object fails.

Source code in dataexcept/exceptions/parsing.py

class SerializationError(JobError):
    """Raised when serialization of an object fails."""

    def __init__(self, obj, format: str, message: str | None = None):
        self.obj = obj
        self.format = format
        self.message = message or f"Failed to serialize object to {format}"
        super().__init__(self.message)

ServiceConnectionError

Bases: JobError

Raised when a connection to an external service fails.

Source code in dataexcept/exceptions/external.py

class ServiceConnectionError(JobError):
    """Raised when a connection to an external service fails."""

    def __init__(self, service_name: str, original_exception: Exception | None = None):
        self.service_name = service_name
        self.original_exception = original_exception
        msg = f"Failed to connect to service '{service_name}'"
        if original_exception:
            msg += f": {original_exception}"
        super().__init__(msg)

ValidationError

Bases: JobError

Raised when input data fails validation.

Source code in dataexcept/exceptions/validation.py

class ValidationError(JobError):
    """Raised when input data fails validation."""

    def __init__(self, field: str, value, message: str | None = None):
        self.field = field
        self.value = value
        self.message = message or f"Validation failed for field '{field}': {value!r}"
        super().__init__(self.message)

WebhookError

Bases: NotificationError

Raised when a webhook POST fails.

Source code in dataexcept/exceptions/notification.py

class WebhookError(NotificationError):
    """Raised when a webhook POST fails."""

    # Slack, Discord and others put the secret in the webhook path, so keeping
    # the path would defeat the redaction. This also applies to the scrubbing
    # of the whole message, which is how a wrapped HTTP exception quoting the
    # original URL gets its path dropped too.
    _keep_url_path = False

    def __init__(self, url: str, original_exception: Exception | None = None):
        self.url = redact_url(url, keep_path=False)
        # original_exception is set by NotificationError.__init__ below.
        msg = f"Webhook to URL '{self.url}' failed"
        if original_exception:
            msg += f": {original_exception}"
        super().__init__("webhook", original_exception, message=msg)

CustomIOError

Bases: DataExceptError

Base exception for I/O errors.

Source code in dataexcept/io_exceptions.py

class CustomIOError(DataExceptError):
    """Base exception for I/O errors."""

    pass

FileLockError

Bases: CustomIOError

Raised when a file lock cannot be acquired.

Source code in dataexcept/io_exceptions.py

class FileLockError(CustomIOError):
    """Raised when a file lock cannot be acquired."""

    def __init__(self, path: str) -> None:
        """Initialize FileLockError.

        Args:
            path: Path of the lock file.
        """
        self.path = redact_if_url(path)
        super().__init__(f"Unable to obtain lock for '{path}'")

FileReadError

Bases: CustomIOError

Raised when reading a file fails.

Source code in dataexcept/io_exceptions.py

class FileReadError(CustomIOError):
    """Raised when reading a file fails."""

    def __init__(self, path: str, original: Exception | None = None) -> None:
        """Initialize FileReadError.

        Args:
            path: File path that could not be read.
            original: Optional underlying exception.
        """
        self.path = redact_if_url(path)
        self.original = original
        msg = f"Failed to read file '{path}'"
        if original:
            msg += f": {original}"
        super().__init__(msg)

FileWriteError

Bases: CustomIOError

Raised when writing to a file fails.

Source code in dataexcept/io_exceptions.py

class FileWriteError(CustomIOError):
    """Raised when writing to a file fails."""

    def __init__(self, path: str, original: Exception | None = None) -> None:
        """Initialize FileWriteError.

        Args:
            path: File path that could not be written to.
            original: Optional underlying exception.
        """
        self.path = redact_if_url(path)
        self.original = original
        msg = f"Failed to write file '{path}'"
        if original:
            msg += f": {original}"
        super().__init__(msg)

ConnectionTimeoutError

Bases: NetworkError

Raised when a network connection attempt times out.

Example

from dataexcept.network_exceptions import ConnectionTimeoutError try: ... raise ConnectionTimeoutError("api.example.com", 30) ... except ConnectionTimeoutError as exc: ... print(exc) Connection to 'api.example.com' timed out after 30 seconds

Source code in dataexcept/network_exceptions.py

class ConnectionTimeoutError(NetworkError):
    """Raised when a network connection attempt times out.

    Example:
        >>> from dataexcept.network_exceptions import ConnectionTimeoutError
        >>> try:
        ...     raise ConnectionTimeoutError("api.example.com", 30)
        ... except ConnectionTimeoutError as exc:
        ...     print(exc)
        Connection to 'api.example.com' timed out after 30 seconds
    """

    def __init__(self, host: str, timeout: float) -> None:
        """Initialize ConnectionTimeoutError.

        Args:
            host: Host address.
            timeout: Timeout in seconds.
        """
        self.host = host
        self.timeout = timeout
        msg = f"Connection to '{host}' timed out after {timeout} seconds"
        super().__init__(msg)

HostUnreachableError

Bases: NetworkError

Raised when a remote host cannot be reached.

Example

from dataexcept.network_exceptions import HostUnreachableError try: ... raise HostUnreachableError("api.example.com") ... except HostUnreachableError as exc: ... print(exc) Host 'api.example.com' is unreachable

Source code in dataexcept/network_exceptions.py

class HostUnreachableError(NetworkError):
    """Raised when a remote host cannot be reached.

    Example:
        >>> from dataexcept.network_exceptions import HostUnreachableError
        >>> try:
        ...     raise HostUnreachableError("api.example.com")
        ... except HostUnreachableError as exc:
        ...     print(exc)
        Host 'api.example.com' is unreachable
    """

    def __init__(self, host: str, message: str | None = None) -> None:
        """Initialize HostUnreachableError.

        Args:
            host: Host address that could not be reached.
            message: Optional custom error message.
        """
        self.host = host
        default = f"Host '{host}' is unreachable"
        super().__init__(message or default)

NetworkError

Bases: DataExceptError

Base exception for network-related errors.

Example

from dataexcept.network_exceptions import NetworkError try: ... raise NetworkError("Something went wrong") ... except NetworkError: ... print("Caught network error") Caught network error

Source code in dataexcept/network_exceptions.py

class NetworkError(DataExceptError):
    """Base exception for network-related errors.

    Example:
        >>> from dataexcept.network_exceptions import NetworkError
        >>> try:
        ...     raise NetworkError("Something went wrong")
        ... except NetworkError:
        ...     print("Caught network error")
        Caught network error
    """

    pass

ProtocolError

Bases: NetworkError

Raised when an unexpected protocol error occurs.

Example

from dataexcept.network_exceptions import ProtocolError try: ... raise ProtocolError("HTTP", "Invalid status line") ... except ProtocolError as exc: ... print(exc) Protocol error in HTTP: Invalid status line

Source code in dataexcept/network_exceptions.py

class ProtocolError(NetworkError):
    """Raised when an unexpected protocol error occurs.

    Example:
        >>> from dataexcept.network_exceptions import ProtocolError
        >>> try:
        ...     raise ProtocolError("HTTP", "Invalid status line")
        ... except ProtocolError as exc:
        ...     print(exc)
        Protocol error in HTTP: Invalid status line
    """

    def __init__(self, protocol: str, details: str | None = None) -> None:
        """Initialize ProtocolError.

        Args:
            protocol: Protocol name (e.g., HTTP).
            details: Optional additional details about the failure.
        """
        self.protocol = protocol
        self.details = details
        msg = f"Protocol error in {protocol}"
        if details:
            msg += f": {details}"
        super().__init__(msg)

DtypeMismatchError

Bases: PandasError

Raised when a column has an unexpected dtype.

Parameters:

Name	Type	Description	Default
column	str	Name of the column.	required
expected	Sequence[str]	Sequence of allowed dtypes.	required
found	str	Detected dtype for the column.	required

Source code in dataexcept/pandas_exceptions.py

class DtypeMismatchError(PandasError):
    """Raised when a column has an unexpected dtype.

    Args:
        column: Name of the column.
        expected: Sequence of allowed dtypes.
        found: Detected dtype for the column.
    """

    def __init__(self, column: str, expected: Sequence[str], found: str) -> None:
        if not isinstance(column, str):
            raise TypeError(f"column must be str, got {type(column).__name__}")
        if not isinstance(found, str):
            raise TypeError(f"found must be str, got {type(found).__name__}")
        if not isinstance(expected, Sequence) or isinstance(expected, str):
            raise TypeError("expected must be a sequence of strings")
        if not all(isinstance(dt, str) for dt in expected):
            raise TypeError("expected must contain strings")

        self.column = column
        self.expected = list(expected)
        self.found = found
        expected_fmt = ", ".join(self.expected)
        msg = f"Column '{column}' has dtype {found}; expected {expected_fmt}"
        super().__init__(msg)

    def __str__(self) -> str:
        return f"[DtypeMismatchError:{self.column}] {self.args[0]}"

IndexAlignmentError

Bases: PandasError

Raised when DataFrame indices are misaligned for an operation.

Parameters:

Name	Type	Description	Default
details	Optional[str]	Optional details about the misalignment.	None

Source code in dataexcept/pandas_exceptions.py

class IndexAlignmentError(PandasError):
    """Raised when DataFrame indices are misaligned for an operation.

    Args:
        details: Optional details about the misalignment.
    """

    def __init__(self, details: Optional[str] = None) -> None:
        if details is not None and not isinstance(details, str):
            raise TypeError(
                f"details must be str or None, got {type(details).__name__}"
            )
        msg = "DataFrame indices are misaligned"
        if details:
            msg += f": {details}"
        self.details = details
        super().__init__(msg)

    def __str__(self) -> str:
        return f"[IndexAlignmentError] {self.args[0]}"

MergeKeyError

Bases: PandasError

Raised when merging DataFrames fails due to key issues.

Parameters:

Name	Type	Description	Default
left_keys	Sequence[str]	Keys from the left DataFrame.	required
right_keys	Sequence[str]	Keys from the right DataFrame.	required

Source code in dataexcept/pandas_exceptions.py

class MergeKeyError(PandasError):
    """Raised when merging DataFrames fails due to key issues.

    Args:
        left_keys: Keys from the left DataFrame.
        right_keys: Keys from the right DataFrame.
    """

    def __init__(self, left_keys: Sequence[str], right_keys: Sequence[str]) -> None:
        # A bare string is a sequence of strings, so "id" would silently become
        # ['i', 'd']. Reject it, as DtypeMismatchError already does.
        for name, keys in (("left_keys", left_keys), ("right_keys", right_keys)):
            if isinstance(keys, str) or not all(isinstance(k, str) for k in keys):
                raise TypeError(f"{name} must be a sequence of strings, not a string")
        self.left_keys = list(left_keys)
        self.right_keys = list(right_keys)
        msg = f"Failed to merge on keys {self.left_keys} and {self.right_keys}"
        super().__init__(msg)

    def __str__(self) -> str:
        return f"[MergeKeyError] {self.args[0]}"

MissingColumnError

Bases: PandasError

Raised when a required DataFrame column is missing.

Parameters:

Name	Type	Description	Default
column	str	Name of the missing column.	required
dataframe	Optional[str]	Optional name of the DataFrame being inspected.	None

Source code in dataexcept/pandas_exceptions.py

class MissingColumnError(PandasError):
    """Raised when a required DataFrame column is missing.

    Args:
        column: Name of the missing column.
        dataframe: Optional name of the DataFrame being inspected.
    """

    def __init__(self, column: str, dataframe: Optional[str] = None) -> None:
        if not isinstance(column, str):
            raise TypeError(f"column must be str, got {type(column).__name__}")
        if dataframe is not None and not isinstance(dataframe, str):
            raise TypeError(
                "dataframe must be str or None, " f"got {type(dataframe).__name__}"
            )

        self.column = column
        self.dataframe = dataframe
        name = f" in DataFrame '{dataframe}'" if dataframe else ""
        msg = f"Missing required column '{column}'{name}"
        super().__init__(msg)

    def __str__(self) -> str:
        return f"[MissingColumnError] {self.args[0]}"

PandasError

Bases: DataExceptError

Base exception for pandas-related errors.

Source code in dataexcept/pandas_exceptions.py

class PandasError(DataExceptError):
    """Base exception for pandas-related errors."""

PandasIOError

Bases: PandasError

Raised when reading from or writing to disk with pandas fails.

Parameters:

Name	Type	Description	Default
path	str	File path involved in the operation.	required
original	Exception	The underlying exception that was raised.	required

Source code in dataexcept/pandas_exceptions.py

class PandasIOError(PandasError):
    """Raised when reading from or writing to disk with pandas fails.

    Args:
        path: File path involved in the operation.
        original: The underlying exception that was raised.
    """

    def __init__(self, path: str, original: Exception) -> None:
        if not isinstance(path, str):
            raise TypeError(f"path must be str, got {type(path).__name__}")
        if not isinstance(original, Exception):
            raise TypeError(
                f"original must be Exception, got {type(original).__name__}"
            )
        self.path = redact_if_url(path)
        self.original = original
        msg = f"Pandas I/O operation failed on {path!r}: {original}"
        super().__init__(msg)

    def __str__(self) -> str:
        return f"[PandasIOError] {self.args[0]}"

ApiError

Bases: PipelineError

Failure calling a REST API endpoint.

Source code in dataexcept/pipeline_exceptions.py

class ApiError(PipelineError):
    """Failure calling a REST API endpoint."""

    def __init__(
        self,
        endpoint: str,
        status_code: Optional[int] = None,
        message: Optional[str] = None,
    ) -> None:
        # An endpoint URL may authenticate through a query parameter.
        self.endpoint = redact_url(endpoint)
        default = f"API call failed: {self.endpoint}"
        if status_code is not None:
            default += f" (status {status_code})"
        self.status_code = status_code
        super().__init__(message or default)

DataFetchError

Bases: PipelineError

Failed to fetch data from a storage backend.

Source code in dataexcept/pipeline_exceptions.py

class DataFetchError(PipelineError):
    """Failed to fetch data from a storage backend."""

    def __init__(
        self,
        source: str,
        cid: str,
        message: Optional[str] = None,
    ) -> None:
        default = f"Failed to fetch '{source}' data for cid={cid}"
        self.source = redact_if_url(source)
        self.cid = cid
        super().__init__(message or default)

ExternalServiceError

Bases: PipelineError

General failure when calling an external service.

Source code in dataexcept/pipeline_exceptions.py

class ExternalServiceError(PipelineError):
    """General failure when calling an external service."""

    def __init__(
        self,
        service_name: str,
        status_code: Optional[int] = None,
        response: Optional[Any] = None,
        message: Optional[str] = None,
    ) -> None:
        default = f"Call to external service '{service_name}' failed."
        self.service_name = service_name
        self.status_code = status_code
        self.response = response
        super().__init__(message or default)

FeaturePreprocessingError

Bases: PreprocessingError

Raised when feature engineering fails.

Source code in dataexcept/pipeline_exceptions.py

class FeaturePreprocessingError(PreprocessingError):
    """Raised when feature engineering fails."""

    def __init__(self, feature: str, reason: Optional[str] = None) -> None:
        # Assigned before super(): DataExceptError.__init__ sweeps the stored
        # strings for URLs, and anything set afterwards escapes that.
        self.feature = feature
        self.reason = reason
        super().__init__(step_name=f"feature_{feature}", details=reason)

PipelineError

Bases: DataExceptError

Base exception for pipeline errors.

Source code in dataexcept/pipeline_exceptions.py

class PipelineError(DataExceptError):
    """Base exception for pipeline errors."""

    pass

PipelineNotificationError

Bases: PipelineError

Raised when sending a notification fails.

Source code in dataexcept/pipeline_exceptions.py

class PipelineNotificationError(PipelineError):
    """Raised when sending a notification fails."""

    def __init__(
        self,
        channel: str,
        payload: Any,
        message: Optional[str] = None,
    ) -> None:
        default = f"Notification via '{channel}' failed."
        self.channel = channel
        self.payload = payload
        super().__init__(message or default)

PreprocessingError

Bases: PipelineError

Raised when a preprocessing step fails.

Source code in dataexcept/pipeline_exceptions.py

class PreprocessingError(PipelineError):
    """Raised when a preprocessing step fails."""

    def __init__(self, step_name: str, details: Optional[str] = None) -> None:
        default = f"Preprocessing failed at step: '{step_name}'."
        message = f"{default} Details: {details}" if details else default
        self.step_name = step_name
        self.details = details
        super().__init__(message)

RetryLimitExceededError

Bases: PipelineError

Raised when an operation is retried too many times.

Source code in dataexcept/pipeline_exceptions.py

class RetryLimitExceededError(PipelineError):
    """Raised when an operation is retried too many times."""

    def __init__(
        self,
        operation: str,
        retries: int,
        message: Optional[str] = None,
    ) -> None:
        default = (
            "Retry limit exceeded for operation "
            f"'{operation}' after {retries} attempts."
        )
        self.operation = operation
        self.retries = retries
        super().__init__(message or default)

ServiceAuthenticationError

Bases: ExternalServiceError

Authentication to an external service failed.

Source code in dataexcept/pipeline_exceptions.py

class ServiceAuthenticationError(ExternalServiceError):
    """Authentication to an external service failed."""

    def __init__(
        self,
        service_name: str,
        message: Optional[str] = None,
    ) -> None:
        default = f"Authentication failed for service '{service_name}'."
        super().__init__(service_name=service_name, message=message or default)

ServiceAuthorizationError

Bases: ExternalServiceError

Authorization was denied by an external service.

Source code in dataexcept/pipeline_exceptions.py

class ServiceAuthorizationError(ExternalServiceError):
    """Authorization was denied by an external service."""

    def __init__(
        self,
        service_name: str,
        message: Optional[str] = None,
    ) -> None:
        default = f"Authorization denied for service '{service_name}'."
        super().__init__(service_name=service_name, message=message or default)

ServiceTimeoutError

Bases: ExternalServiceError

A call to an external service exceeded the allotted time.

Source code in dataexcept/pipeline_exceptions.py

class ServiceTimeoutError(ExternalServiceError):
    """A call to an external service exceeded the allotted time."""

    def __init__(
        self,
        service_name: str,
        timeout_seconds: Optional[float] = None,
    ) -> None:
        default = (
            "Operation timed out after "
            f"{timeout_seconds}s on service '{service_name}'."
        )
        self.timeout_seconds = timeout_seconds
        super().__init__(service_name=service_name, message=default)

StorageError

Bases: PipelineError

Raised when reading from or writing to storage fails.

Source code in dataexcept/pipeline_exceptions.py

class StorageError(PipelineError):
    """Raised when reading from or writing to storage fails."""

    def __init__(
        self,
        location: str,
        operation: str,
        message: Optional[str] = None,
    ) -> None:
        default = f"Storage {operation} failed at location: '{location}'."
        self.location = redact_if_url(location)
        self.operation = operation
        super().__init__(message or default)

TimeDeltaTooLargeError

Bases: PipelineError

The time span between records exceeded a threshold.

Source code in dataexcept/pipeline_exceptions.py

class TimeDeltaTooLargeError(PipelineError):
    """The time span between records exceeded a threshold."""

    def __init__(
        self,
        user: str,
        delta_minutes: float,
        message: Optional[str] = None,
    ) -> None:
        default = f"Time delta {delta_minutes}m too large for user {user}"
        self.user = user
        self.delta_minutes = delta_minutes
        super().__init__(message or default)

TypeCheckError

Bases: PipelineError

Invalid type detected during recursive type inspection.

Source code in dataexcept/pipeline_exceptions.py

class TypeCheckError(PipelineError):
    """Invalid type detected during recursive type inspection."""

DecryptionError

Bases: SecurityError

Raised when data decryption fails.

Source code in dataexcept/security_exceptions.py

class DecryptionError(SecurityError):
    """Raised when data decryption fails."""

    def __init__(self, algorithm: str, message: str | None = None) -> None:
        """Initialize DecryptionError.

        Args:
            algorithm: Name of the decryption algorithm.
            message: Optional custom error message.
        """
        self.algorithm = algorithm
        default = f"Decryption failed using {algorithm}"
        super().__init__(message or default)

EncryptionError

Bases: SecurityError

Raised when data encryption fails.

Source code in dataexcept/security_exceptions.py

class EncryptionError(SecurityError):
    """Raised when data encryption fails."""

    def __init__(self, algorithm: str, message: str | None = None) -> None:
        """Initialize EncryptionError.

        Args:
            algorithm: Name of the encryption algorithm.
            message: Optional custom error message.
        """
        self.algorithm = algorithm
        default = f"Encryption failed using {algorithm}"
        super().__init__(message or default)

InvalidTokenError

Bases: SecurityError

Raised when an authentication token is invalid or expired.

Source code in dataexcept/security_exceptions.py

class InvalidTokenError(SecurityError):
    """Raised when an authentication token is invalid or expired."""

    def __init__(
        self,
        token: str | None = None,
        message: str | None = None,
    ) -> None:
        """Initialize InvalidTokenError.

        Args:
            token: The problematic token.
            message: Optional custom error message.
        """
        # The raw token is never stored or rendered: this exception is often
        # logged, and the caller already holds the value it passed in.
        self.token = redact_secret(token)
        default = "Invalid authentication token"
        if token:
            default += f": {self.token}"
        # The library was handed the secret, so it can be removed even from a
        # message the caller wrote themselves.
        super().__init__(remove_secret(message or default, token))

SecurityError

Bases: DataExceptError

Base exception for security errors.

Source code in dataexcept/security_exceptions.py

class SecurityError(DataExceptError):
    """Base exception for security errors."""

    pass

log_and_raise

log_and_raise(logger: Optional[Logger] = None, level: int = logging.ERROR, context: Context | None = None) -> Iterator[None]

Context manager that logs and re-raises exceptions preserving traceback.

Source code in dataexcept/logging_helpers.py

@contextlib.contextmanager
def log_and_raise(
    logger: Optional[logging.Logger] = None,
    level: int = logging.ERROR,
    context: Context | None = None,
) -> Iterator[None]:
    """Context manager that logs and re-raises exceptions preserving traceback."""
    try:
        yield
    except Exception as exc:
        log_exception(exc, logger=logger, level=level, context=context)
        raise

log_exception

log_exception(exc: Exception, logger: Optional[Logger] = None, level: int = logging.ERROR, context: Context | None = None) -> None

Log exc at the given log level using logger.

If logger is None a module level logger is used.

DataExcept redacts what it renders, but a wrapped third-party exception renders itself: an HTTP client's error may quote the credential-bearing URL it was called with, and exc_info makes logging print that whole chain. When the chain contains a URL the traceback is formatted and scrubbed here; otherwise the structured exc_info path is used unchanged, so ordinary exceptions keep the shape log aggregators expect.

Source code in dataexcept/logging_helpers.py

def log_exception(
    exc: Exception,
    logger: Optional[logging.Logger] = None,
    level: int = logging.ERROR,
    context: Context | None = None,
) -> None:
    """Log *exc* at the given log *level* using *logger*.

    If *logger* is ``None`` a module level logger is used.

    DataExcept redacts what it renders, but a wrapped third-party exception
    renders itself: an HTTP client's error may quote the credential-bearing URL
    it was called with, and ``exc_info`` makes logging print that whole chain.
    When the chain contains a URL the traceback is formatted and scrubbed here;
    otherwise the structured ``exc_info`` path is used unchanged, so ordinary
    exceptions keep the shape log aggregators expect.
    """
    if logger is None:
        logger = logging.getLogger(__name__)
    extra = _build_extra(context)

    if _chain_mentions_a_url(exc):
        formatted = "".join(
            traceback.format_exception(type(exc), exc, exc.__traceback__)
        )
        keep_path = getattr(type(exc), "_keep_url_path", True)
        scrubbed = redact_urls_in_text(formatted, keep_path=keep_path).rstrip()
        logger.log(level, "%s\n%s", exc, scrubbed, extra=extra)
        return

    exc_info = (type(exc), exc, exc.__traceback__)
    logger.log(level, "%s", exc, exc_info=exc_info, extra=extra)

log_then_raise

log_then_raise(exc: Exception, logger: Optional[Logger] = None, level: int = logging.ERROR, context: Context | None = None) -> None

Log exc and immediately raise it.

This helper mirrors the pre-context-manager API for scenarios where adding a with block would be too intrusive. Prefer :func:log_and_raise whenever possible so tracebacks remain untouched.

Source code in dataexcept/logging_helpers.py

def log_then_raise(
    exc: Exception,
    logger: Optional[logging.Logger] = None,
    level: int = logging.ERROR,
    context: Context | None = None,
) -> None:
    """Log *exc* and immediately raise it.

    This helper mirrors the pre-context-manager API for scenarios where adding a
    ``with`` block would be too intrusive. Prefer :func:`log_and_raise` whenever
    possible so tracebacks remain untouched.
    """
    log_exception(exc, logger=logger, level=level, context=context)
    raise exc

exception_to_dict

exception_to_dict(exc: BaseException, *, include_attributes: bool = True, max_depth: int = 8) -> dict[str, Any]

Return a strict JSON-safe structured representation of exc.

The representation contains the exception type, module and rendered message, optionally public instance attributes, bounded cause/context chains, and on Python 3.11+ the member tree of exception groups. Traceback frames and private attributes are deliberately excluded.

Source code in dataexcept/serialization.py

def exception_to_dict(
    exc: BaseException,
    *,
    include_attributes: bool = True,
    max_depth: int = 8,
) -> dict[str, Any]:
    """Return a strict JSON-safe structured representation of *exc*.

    The representation contains the exception type, module and rendered
    message, optionally public instance attributes, bounded cause/context
    chains, and on Python 3.11+ the member tree of exception groups. Traceback
    frames and private attributes are deliberately excluded.
    """
    if not isinstance(exc, BaseException):
        raise TypeError("exc must be an exception instance")
    if not isinstance(max_depth, int) or isinstance(max_depth, bool):
        raise TypeError("max_depth must be an integer")
    if max_depth < 0:
        raise ValueError("max_depth must be non-negative")
    return _exception_record(
        exc,
        include_attributes=include_attributes,
        max_depth=max_depth,
        depth=0,
        seen=set(),
    )

exception_to_json

exception_to_json(exc: BaseException, *, include_attributes: bool = True, max_depth: int = 8, **json_kwargs: Any) -> str

Return :func:exception_to_dict encoded as strict JSON.

Source code in dataexcept/serialization.py

def exception_to_json(
    exc: BaseException,
    *,
    include_attributes: bool = True,
    max_depth: int = 8,
    **json_kwargs: Any,
) -> str:
    """Return :func:`exception_to_dict` encoded as strict JSON."""
    json_kwargs["allow_nan"] = False
    return json.dumps(
        exception_to_dict(
            exc,
            include_attributes=include_attributes,
            max_depth=max_depth,
        ),
        **json_kwargs,
    )

wrap

wrap(original: BaseException, target: Type[DataExceptError], /, **kwargs: Any) -> DataExceptError

Build target from original, recording it as the cause.

Extra keyword arguments go to the constructor::

raise wrap(exc, DataLoadingError, source=path) from exc

If target accepts a cause parameter, original is passed to it. Either way __cause__ is set, so a traceback shows the underlying failure even for a class that records nothing.

An explicit original/cause keyword wins, so a caller can still say exactly what they mean.

Source code in dataexcept/wrapping.py

def wrap(
    original: BaseException,
    target: Type[DataExceptError],
    /,
    **kwargs: Any,
) -> DataExceptError:
    """Build *target* from *original*, recording it as the cause.

    Extra keyword arguments go to the constructor::

        raise wrap(exc, DataLoadingError, source=path) from exc

    If *target* accepts a cause parameter, *original* is passed to it. Either
    way ``__cause__`` is set, so a traceback shows the underlying failure even
    for a class that records nothing.

    An explicit ``original``/``cause`` keyword wins, so a caller can still say
    exactly what they mean.
    """
    parameter = _cause_parameter(target)
    if parameter is not None and parameter not in kwargs:
        kwargs[parameter] = original

    exception = target(**kwargs)
    # Set unconditionally: the target may record nothing, and the point is that
    # the traceback shows what actually failed.
    exception.__cause__ = original
    return exception

Core job exceptions

exceptions

AuthenticationError

Bases: JobError

Raised when user authentication fails.

Source code in dataexcept/exceptions/authentication.py

class AuthenticationError(JobError):
    """Raised when user authentication fails."""

    def __init__(self, user: str, message: str | None = None):
        self.user = user
        self.message = message or f"Authentication failed for user '{user}'"
        super().__init__(self.message)

AuthorizationError

Bases: JobError

Raised when user lacks permission for an action.

Source code in dataexcept/exceptions/authentication.py

class AuthorizationError(JobError):
    """Raised when user lacks permission for an action."""

    def __init__(self, user: str, permission: str):
        self.user = user
        self.permission = permission
        msg = f"User '{user}' lacks permission '{permission}'"
        super().__init__(msg)

JobError

Bases: DataExceptError

Base exception for all job-related errors.

Source code in dataexcept/exceptions/base.py

class JobError(DataExceptError):
    """Base exception for all job-related errors."""

    pass

ConfigurationError

Bases: JobError

Raised when there is a problem with configuration or settings.

Source code in dataexcept/exceptions/configuration.py

class ConfigurationError(JobError):
    """Raised when there is a problem with configuration or settings."""

    def __init__(self, option: str, message: str | None = None):
        self.option = option
        self.message = message or f"Invalid configuration for '{option}'"
        super().__init__(self.message)

DependencyError

Bases: JobError

Raised when a job dependency is missing or fails.

Source code in dataexcept/exceptions/external.py

class DependencyError(JobError):
    """Raised when a job dependency is missing or fails."""

    def __init__(self, dependency_name: str, message: str | None = None):
        self.dependency_name = dependency_name
        self.message = message or f"Dependency '{dependency_name}' error"
        super().__init__(self.message)

OperationTimeoutError

Bases: JobError

Raised when an operation exceeds its time limit.

Source code in dataexcept/exceptions/external.py

class OperationTimeoutError(JobError):
    """Raised when an operation exceeds its time limit."""

    def __init__(self, operation: str, timeout: float):
        self.operation = operation
        self.timeout = timeout
        msg = f"Operation '{operation}' timed out after {timeout} seconds"
        super().__init__(msg)

ResourceNotFoundError

Bases: JobError

Raised when a required resource cannot be found.

Source code in dataexcept/exceptions/external.py

class ResourceNotFoundError(JobError):
    """Raised when a required resource cannot be found."""

    def __init__(self, resource_type: str, identifier: str):
        self.resource_type = resource_type
        self.identifier = identifier
        msg = f"{resource_type} with identifier '{identifier}' not found"
        super().__init__(msg)

ServiceConnectionError

Bases: JobError

Raised when a connection to an external service fails.

Source code in dataexcept/exceptions/external.py

class ServiceConnectionError(JobError):
    """Raised when a connection to an external service fails."""

    def __init__(self, service_name: str, original_exception: Exception | None = None):
        self.service_name = service_name
        self.original_exception = original_exception
        msg = f"Failed to connect to service '{service_name}'"
        if original_exception:
            msg += f": {original_exception}"
        super().__init__(msg)

JobCancellationError

Bases: JobError

Raised when a job is cancelled before completion.

Source code in dataexcept/exceptions/lifecycle.py

class JobCancellationError(JobError):
    """Raised when a job is cancelled before completion."""

    def __init__(self, job_id: str, reason: str | None = None):
        self.job_id = job_id
        self.reason = reason
        msg = f"Job '{job_id}' was cancelled"
        if reason:
            msg += f": {reason}"
        super().__init__(msg)

EmailError

Bases: NotificationError

Raised when sending an email fails.

Source code in dataexcept/exceptions/notification.py

class EmailError(NotificationError):
    """Raised when sending an email fails."""

    def __init__(
        self,
        recipient: str,
        subject: str,
        original_exception: Exception | None = None,
    ):
        self.recipient = recipient
        self.subject = subject
        # original_exception is set by NotificationError.__init__ below.
        msg = f"Email to '{recipient}' with subject '{subject}' failed"
        if original_exception:
            msg += f": {original_exception}"
        super().__init__("email", original_exception, message=msg)

NotificationError

Bases: JobError

Base exception for notification failures.

Source code in dataexcept/exceptions/notification.py

class NotificationError(JobError):
    """Base exception for notification failures."""

    def __init__(
        self,
        channel: str,
        original_exception: Exception | None = None,
        message: str | None = None,
    ):
        self.channel = channel
        self.original_exception = original_exception
        msg = message or f"Notification via '{channel}' failed"
        if message is None and original_exception:
            msg += f": {original_exception}"
        super().__init__(msg)

WebhookError

Bases: NotificationError

Raised when a webhook POST fails.

Source code in dataexcept/exceptions/notification.py

class WebhookError(NotificationError):
    """Raised when a webhook POST fails."""

    # Slack, Discord and others put the secret in the webhook path, so keeping
    # the path would defeat the redaction. This also applies to the scrubbing
    # of the whole message, which is how a wrapped HTTP exception quoting the
    # original URL gets its path dropped too.
    _keep_url_path = False

    def __init__(self, url: str, original_exception: Exception | None = None):
        self.url = redact_url(url, keep_path=False)
        # original_exception is set by NotificationError.__init__ below.
        msg = f"Webhook to URL '{self.url}' failed"
        if original_exception:
            msg += f": {original_exception}"
        super().__init__("webhook", original_exception, message=msg)

DeserializationError

Bases: JobError

Raised when deserialization of data fails.

Source code in dataexcept/exceptions/parsing.py

class DeserializationError(JobError):
    """Raised when deserialization of data fails."""

    def __init__(self, data: bytes, format: str, message: str | None = None):
        self.data = data
        self.format = format
        self.message = message or f"Failed to deserialize data from {format}"
        super().__init__(self.message)

ParsingError

Bases: JobError

Raised when parsing of input data fails.

Source code in dataexcept/exceptions/parsing.py

class ParsingError(JobError):
    """Raised when parsing of input data fails."""

    def __init__(self, text: str, message: str | None = None):
        self.text = text
        self.message = message or f"Failed to parse text: {text!r}"
        super().__init__(self.message)

SerializationError

Bases: JobError

Raised when serialization of an object fails.

Source code in dataexcept/exceptions/parsing.py

class SerializationError(JobError):
    """Raised when serialization of an object fails."""

    def __init__(self, obj, format: str, message: str | None = None):
        self.obj = obj
        self.format = format
        self.message = message or f"Failed to serialize object to {format}"
        super().__init__(self.message)

CronExpressionError

Bases: JobError

Raised when a cron expression is invalid.

Source code in dataexcept/exceptions/scheduling.py

class CronExpressionError(JobError):
    """Raised when a cron expression is invalid."""

    def __init__(self, expression: str, message: str | None = None):
        self.expression = expression
        self.message = message or f"Invalid cron expression: '{expression}'"
        super().__init__(self.message)

ScheduleConflictError

Bases: JobError

Raised when two jobs have conflicting schedules.

Source code in dataexcept/exceptions/scheduling.py

class ScheduleConflictError(JobError):
    """Raised when two jobs have conflicting schedules."""

    def __init__(self, job_name: str, schedule: str):
        self.job_name = job_name
        self.schedule = schedule
        msg = f"Schedule conflict for job '{job_name}' on schedule '{schedule}'"
        super().__init__(msg)

ValidationError

Bases: JobError

Raised when input data fails validation.

Source code in dataexcept/exceptions/validation.py

class ValidationError(JobError):
    """Raised when input data fails validation."""

    def __init__(self, field: str, value, message: str | None = None):
        self.field = field
        self.value = value
        self.message = message or f"Validation failed for field '{field}': {value!r}"
        super().__init__(self.message)

Data science exceptions

datascience_exceptions

Custom exceptions for data science workflows.

DataScienceError

Bases: DataExceptError

Base exception for data science errors.

Source code in dataexcept/datascience_exceptions/base.py

class DataScienceError(DataExceptError):
    """Base exception for data science errors."""

    def __init__(self, message: str) -> None:
        # Ensure message is a string
        if not isinstance(message, str):
            raise TypeError(f"message must be str, got {type(message).__name__}")
        self.message = message
        super().__init__(message)

    def __str__(self) -> str:
        return f"[DataScienceError] {self.message}"

DataAugmentationError

Bases: DataScienceError

Raised when a data augmentation technique fails.

Parameters:

Name	Type	Description	Default
technique	str	Name of the augmentation technique.	required
details	Optional[str]	Optional explanation of the failure.	None

Source code in dataexcept/datascience_exceptions/ingestion.py

class DataAugmentationError(DataScienceError):
    """Raised when a data augmentation technique fails.

    Args:
        technique: Name of the augmentation technique.
        details: Optional explanation of the failure.
    """

    def __init__(self, technique: str, details: Optional[str] = None) -> None:
        if not isinstance(technique, str):
            raise TypeError(f"technique must be str, got {type(technique).__name__}")
        if details is not None and not isinstance(details, str):
            raise TypeError(
                f"details must be str or None, got {type(details).__name__}"
            )

        msg = f"Data augmentation '{technique}' failed"
        if details:
            msg += f": {details}"

        self.technique = technique
        self.details = details
        super().__init__(msg)

    def __str__(self) -> str:
        return f"[DataAugmentationError:{self.technique}] {self.message}"

DataFormatError

Bases: DataScienceError

Raised when input data is not in the expected format.

Source code in dataexcept/datascience_exceptions/ingestion.py

class DataFormatError(DataScienceError):
    """Raised when input data is not in the expected format."""

    def __init__(self, expected_formats: Sequence[str], found_format: str) -> None:
        if not isinstance(found_format, str):
            raise TypeError(
                f"found_format must be str, got {type(found_format).__name__}"
            )
        if not isinstance(expected_formats, Sequence) or isinstance(
            expected_formats, str
        ):
            raise TypeError("expected_formats must be a sequence of strings")
        if not all(isinstance(fmt, str) for fmt in expected_formats):
            raise TypeError("expected_formats must contain strings")

        self.expected_formats = list(expected_formats)
        self.found_format = found_format
        fmt_list = ", ".join(self.expected_formats)
        message = f"Expected data format {fmt_list}; got {found_format}"
        super().__init__(message)

    def __str__(self) -> str:
        return f"[DataFormatError] {self.message}"

DataImbalanceError

Bases: DataScienceError

Raised when class distribution is too imbalanced.

Parameters:

Name	Type	Description	Default
ratio	float	Observed minority-to-majority ratio.	required
threshold	float	Minimum acceptable ratio.	required
message	Optional[str]	Optional custom error message.	None

Source code in dataexcept/datascience_exceptions/ingestion.py

class DataImbalanceError(DataScienceError):
    """Raised when class distribution is too imbalanced.

    Args:
        ratio: Observed minority-to-majority ratio.
        threshold: Minimum acceptable ratio.
        message: Optional custom error message.
    """

    def __init__(
        self, ratio: float, threshold: float, message: Optional[str] = None
    ) -> None:
        if not is_number(ratio):
            raise TypeError(f"ratio must be numeric, got {type(ratio).__name__}")
        if not is_number(threshold):
            raise TypeError(
                f"threshold must be numeric, got {type(threshold).__name__}"
            )
        if message is not None and not isinstance(message, str):
            raise TypeError(
                f"message must be str or None, got {type(message).__name__}"
            )
        self.ratio = float(ratio)
        self.threshold = float(threshold)
        if message is None:
            msg = (
                f"Data imbalance detected: ratio={self.ratio:.3f} < "
                f"threshold={self.threshold:.3f}"
            )
        else:
            msg = message
        super().__init__(msg)

    def __str__(self) -> str:
        return f"[DataImbalanceError] {self.message}"

DataLeakageError

Bases: DataScienceError

Raised when data leakage is detected between train and test sets.

Parameters:

Name	Type	Description	Default
feature	str	Name of the leaked feature.	required
stage	str	Stage where the leakage occurred.	required
message	Optional[str]	Optional custom message.	None

Source code in dataexcept/datascience_exceptions/ingestion.py

class DataLeakageError(DataScienceError):
    """Raised when data leakage is detected between train and test sets.

    Args:
        feature: Name of the leaked feature.
        stage: Stage where the leakage occurred.
        message: Optional custom message.
    """

    def __init__(self, feature: str, stage: str, message: Optional[str] = None) -> None:
        if not isinstance(feature, str):
            raise TypeError(f"feature must be str, got {type(feature).__name__}")
        if not isinstance(stage, str):
            raise TypeError(f"stage must be str, got {type(stage).__name__}")
        if message is not None and not isinstance(message, str):
            raise TypeError(
                f"message must be str or None, got {type(message).__name__}"
            )

        if message is None:
            msg = f"Data leakage detected for '{feature}' during {stage}"
        else:
            msg = message

        self.feature = feature
        self.stage = stage
        super().__init__(msg)

    def __str__(self) -> str:
        return f"[DataLeakageError:{self.feature}] {self.message}"

DataLoadingError

Bases: DataScienceError

Raised when loading data fails.

Attributes:

Name	Type	Description
source		data source description (file path, URL).
original		underlying exception.

Source code in dataexcept/datascience_exceptions/ingestion.py

class DataLoadingError(DataScienceError):
    """
    Raised when loading data fails.

    Attributes:
        source: data source description (file path, URL).
        original: underlying exception.
    """

    def __init__(self, source: str, original: Exception) -> None:
        if not isinstance(source, str):
            raise TypeError(f"source must be str, got {type(source).__name__}")
        if not isinstance(original, Exception):
            raise TypeError(
                f"original must be Exception, got {type(original).__name__}"
            )

        message = f"Failed to load data from {source!r}: {original}"
        self.source = redact_if_url(source)
        self.original = original
        super().__init__(message)

    def __str__(self) -> str:
        return f"[DataLoadingError:{self.source}] {self.message}"

DataNormalizationError

Bases: DataScienceError

Raised when data normalization fails.

Parameters:

Name	Type	Description	Default
method	str	Normalization technique identifier.	required
details	Optional[str]	Optional explanation of the failure.	None

Source code in dataexcept/datascience_exceptions/ingestion.py

class DataNormalizationError(DataScienceError):
    """Raised when data normalization fails.

    Args:
        method: Normalization technique identifier.
        details: Optional explanation of the failure.
    """

    def __init__(self, method: str, details: Optional[str] = None) -> None:
        if not isinstance(method, str):
            raise TypeError(f"method must be str, got {type(method).__name__}")
        if details is not None and not isinstance(details, str):
            raise TypeError(
                f"details must be str or None, got {type(details).__name__}"
            )
        # Build a helpful error message
        msg = f"Normalization using '{method}' failed"
        if details:
            msg += f": {details}"
        self.method = method
        self.details = details
        super().__init__(msg)

    def __str__(self) -> str:
        return f"[DataNormalizationError:{self.method}] {self.message}"

DataValidationError

Bases: DataScienceError

Raised when data fails validation rules.

Attributes:

Name	Type	Description
field		name of invalid field.
value		the invalid value.

Source code in dataexcept/datascience_exceptions/ingestion.py

class DataValidationError(DataScienceError):
    """
    Raised when data fails validation rules.

    Attributes:
        field: name of invalid field.
        value: the invalid value.
    """

    def __init__(self, field: str, value: Any, message: Optional[str] = None) -> None:
        if not isinstance(field, str):
            raise TypeError(f"field must be str, got {type(field).__name__}")

        if message is None:
            message = f"Invalid value for '{field}': {value!r}"
        elif not isinstance(message, str):
            raise TypeError(f"message must be str, got {type(message).__name__}")

        self.field = field
        self.value = value
        super().__init__(message)

    def __str__(self) -> str:
        return f"[DataValidationError:{self.field}] {self.message}"

FeatureEngineeringError

Bases: DataScienceError

Raised during feature engineering steps.

Attributes:

Name	Type	Description
step		description of the step that failed.
cause		optional underlying reason.

Source code in dataexcept/datascience_exceptions/ingestion.py

class FeatureEngineeringError(DataScienceError):
    """
    Raised during feature engineering steps.

    Attributes:
        step: description of the step that failed.
        cause: optional underlying reason.
    """

    def __init__(self, step: str, cause: Optional[str] = None) -> None:
        if not isinstance(step, str):
            raise TypeError(f"step must be str, got {type(step).__name__}")
        if cause is not None and not isinstance(cause, str):
            raise TypeError(f"cause must be str or None, got {type(cause).__name__}")

        msg = f"Feature engineering failed at step '{step}'"
        if cause:
            msg += f": {cause}"

        self.step = step
        self.cause = cause
        super().__init__(msg)

    def __str__(self) -> str:
        return f"[FeatureEngineeringError] {self.message}"

MissingDataError

Bases: DataScienceError

Raised when required data is missing.

Attributes:

Name	Type	Description
feature		name of missing feature.

Source code in dataexcept/datascience_exceptions/ingestion.py

class MissingDataError(DataScienceError):
    """
    Raised when required data is missing.

    Attributes:
        feature: name of missing feature.
    """

    def __init__(self, feature: str, message: Optional[str] = None) -> None:
        if not isinstance(feature, str):
            raise TypeError(f"feature must be str, got {type(feature).__name__}")

        if message is None:
            message = f"Missing required feature: {feature!r}"
        elif not isinstance(message, str):
            raise TypeError(f"message must be str, got {type(message).__name__}")

        self.feature = feature
        super().__init__(message)

    def __str__(self) -> str:
        return f"[MissingDataError:{self.feature}] {self.message}"

OutlierDetectionError

Bases: DataScienceError

Raised when outlier detection fails.

Attributes:

Name	Type	Description
method		detection method name.
details		optional extra info.

Source code in dataexcept/datascience_exceptions/ingestion.py

class OutlierDetectionError(DataScienceError):
    """
    Raised when outlier detection fails.

    Attributes:
        method: detection method name.
        details: optional extra info.
    """

    def __init__(self, method: str, details: Optional[str] = None) -> None:
        if not isinstance(method, str):
            raise TypeError(f"method must be str, got {type(method).__name__}")
        if details is not None and not isinstance(details, str):
            raise TypeError(
                f"details must be str or None, got {type(details).__name__}"
            )

        msg = f"Outlier detection failed using method '{method}'"
        if details:
            msg += f": {details}"

        self.method = method
        self.details = details
        super().__init__(msg)

    def __str__(self) -> str:
        return f"[OutlierDetectionError:{self.method}] {self.message}"

SchemaMismatchError

Bases: DataScienceError

Raised when data schema does not match expected.

Attributes:

Name	Type	Description
expected		expected schema description.
found		actual schema description.

Source code in dataexcept/datascience_exceptions/ingestion.py

class SchemaMismatchError(DataScienceError):
    """
    Raised when data schema does not match expected.

    Attributes:
        expected: expected schema description.
        found: actual schema description.
    """

    def __init__(self, expected: str, found: str) -> None:
        if not isinstance(expected, str):
            raise TypeError(f"expected must be str, got {type(expected).__name__}")
        if not isinstance(found, str):
            raise TypeError(f"found must be str, got {type(found).__name__}")

        message = f"Schema mismatch. Expected: {expected}, Found: {found}"
        self.expected = expected
        self.found = found
        super().__init__(message)

    def __str__(self) -> str:
        return f"[SchemaMismatchError] {self.message}"

DataDriftError

Bases: DataScienceError

Raised when data drift is detected beyond threshold.

Attributes:

Name	Type	Description
feature		feature name.
drift_score		computed drift metric.

Source code in dataexcept/datascience_exceptions/operations.py

class DataDriftError(DataScienceError):
    """
    Raised when data drift is detected beyond threshold.

    Attributes:
        feature: feature name.
        drift_score: computed drift metric.
    """

    def __init__(
        self, feature: str, drift_score: float, message: Optional[str] = None
    ) -> None:
        if not isinstance(feature, str):
            raise TypeError(f"feature must be str, got {type(feature).__name__}")
        if not is_number(drift_score):
            raise TypeError(
                f"drift_score must be number, got {type(drift_score).__name__}"
            )

        self.feature = feature
        self.drift_score = float(drift_score)
        if message is None:
            message = f"Data drift detected on '{feature}', score={drift_score:.4f}"

        super().__init__(message)

    def __str__(self) -> str:
        return f"[DataDriftError:{self.feature}] {self.message}"

DataExportError

Bases: DataScienceError

Failed to export or write data to destination.

Source code in dataexcept/datascience_exceptions/operations.py

class DataExportError(DataScienceError):
    """Failed to export or write data to destination."""

    def __init__(self, destination: str, original: Exception) -> None:
        if not isinstance(destination, str):
            raise TypeError(
                f"destination must be str, got {type(destination).__name__}"
            )
        if not isinstance(original, Exception):
            raise TypeError(
                f"original must be Exception, got {type(original).__name__}"
            )
        msg = f"Unable to export data to {destination}: {original}"
        self.destination = destination
        self.original = original
        super().__init__(msg)

DeploymentError

Bases: DataScienceError

Raised when deploying a model or pipeline fails.

Attributes:

Name	Type	Description
target		deployment target identifier.
cause		optional detail.

Source code in dataexcept/datascience_exceptions/operations.py

class DeploymentError(DataScienceError):
    """
    Raised when deploying a model or pipeline fails.

    Attributes:
        target: deployment target identifier.
        cause: optional detail.
    """

    def __init__(self, target: str, cause: Optional[str] = None) -> None:
        if not isinstance(target, str):
            raise TypeError(f"target must be str, got {type(target).__name__}")
        if cause is not None and not isinstance(cause, str):
            raise TypeError(f"cause must be str or None, got {type(cause).__name__}")

        msg = f"Deployment failed to '{target}'"
        if cause:
            msg += f": {cause}"

        self.target = target
        self.cause = cause
        super().__init__(msg)

    def __str__(self) -> str:
        return f"[DeploymentError:{self.target}] {self.message}"

ModelSerializationError

Bases: DataScienceError

Raised when saving or loading a model fails.

Attributes:

Name	Type	Description
path		file path involved.
original		underlying exception.

Source code in dataexcept/datascience_exceptions/operations.py

class ModelSerializationError(DataScienceError):
    """
    Raised when saving or loading a model fails.

    Attributes:
        path: file path involved.
        original: underlying exception.
    """

    def __init__(self, path: str, original: Exception) -> None:
        if not isinstance(path, str):
            raise TypeError(f"path must be str, got {type(path).__name__}")
        if not isinstance(original, Exception):
            raise TypeError(
                f"original must be Exception, got {type(original).__name__}"
            )

        message = f"Failed to serialize to {path!r}: {original}"
        self.path = redact_if_url(path)
        self.original = original
        super().__init__(message)

    def __str__(self) -> str:
        return f"[ModelSerializationError:{self.path}] {self.message}"

ResourceLimitError

Bases: DataScienceError

Raised when computation exceeds resources (memory, CPU).

Attributes:

Name	Type	Description
resource		'memory', 'cpu', etc.
limit		threshold exceeded.

Source code in dataexcept/datascience_exceptions/operations.py

class ResourceLimitError(DataScienceError):
    """
    Raised when computation exceeds resources (memory, CPU).

    Attributes:
        resource: 'memory', 'cpu', etc.
        limit: threshold exceeded.
    """

    def __init__(self, resource: str, limit: Any) -> None:
        if not isinstance(resource, str):
            raise TypeError(f"resource must be str, got {type(resource).__name__}")

        message = f"Resource limit exceeded: {resource} at {limit!r}"
        self.resource = resource
        self.limit = limit
        super().__init__(message)

    def __str__(self) -> str:
        return f"[ResourceLimitError:{self.resource}] {self.message}"

BiasDetectionError

Bases: DataScienceError

Raised when algorithmic bias exceeds an acceptable threshold.

Parameters:

Name	Type	Description	Default
feature	str	Feature or group where bias was detected.	required
bias_score	float	Calculated bias metric.	required
threshold	float	Maximum acceptable bias metric.	required
message	Optional[str]	Optional custom message.	None

Source code in dataexcept/datascience_exceptions/training.py

class BiasDetectionError(DataScienceError):
    """Raised when algorithmic bias exceeds an acceptable threshold.

    Args:
        feature: Feature or group where bias was detected.
        bias_score: Calculated bias metric.
        threshold: Maximum acceptable bias metric.
        message: Optional custom message.
    """

    def __init__(
        self,
        feature: str,
        bias_score: float,
        threshold: float,
        message: Optional[str] = None,
    ) -> None:
        if not isinstance(feature, str):
            raise TypeError(f"feature must be str, got {type(feature).__name__}")
        if not is_number(bias_score):
            raise TypeError(
                f"bias_score must be numeric, got {type(bias_score).__name__}"
            )
        if not is_number(threshold):
            raise TypeError(
                f"threshold must be numeric, got {type(threshold).__name__}"
            )
        if message is not None and not isinstance(message, str):
            raise TypeError(
                f"message must be str or None, got {type(message).__name__}"
            )

        if message is None:
            msg = (
                f"Bias detected in '{feature}': score={bias_score:.3f} > "
                f"threshold={threshold:.3f}"
            )
        else:
            msg = message

        self.feature = feature
        self.bias_score = float(bias_score)
        self.threshold = float(threshold)
        super().__init__(msg)

    def __str__(self) -> str:
        return f"[BiasDetectionError:{self.feature}] {self.message}"

ConvergenceError

Bases: ModelTrainingError

Raised when optimization fails to converge.

Attributes:

Name	Type	Description
iterations		number of iterations run.

Source code in dataexcept/datascience_exceptions/training.py

class ConvergenceError(ModelTrainingError):
    """
    Raised when optimization fails to converge.

    Attributes:
        iterations: number of iterations run.
    """

    def __init__(
        self, model_type: str, iterations: int, message: Optional[str] = None
    ) -> None:
        if not isinstance(iterations, int):
            raise TypeError(f"iterations must be int, got {type(iterations).__name__}")

        if message is None:
            message = (
                f"Model '{model_type}' failed to converge after "
                f"{iterations} iterations"
            )
        # Assigned before super(): DataExceptError.__init__ sweeps the stored
        # strings for URLs, and anything set afterwards escapes that.
        self.iterations = iterations
        super().__init__(model_type=model_type, epoch=None, message=message)

    def __str__(self) -> str:
        return f"[ConvergenceError] {self.message}"

CrossValidationError

Bases: DataScienceError

Failure during cross-validation procedure.

Source code in dataexcept/datascience_exceptions/training.py

class CrossValidationError(DataScienceError):
    """Failure during cross-validation procedure."""

    def __init__(self, folds: int, cause: Optional[str] = None) -> None:
        if not isinstance(folds, int):
            raise TypeError(f"folds must be int, got {type(folds).__name__}")
        msg = f"Cross-validation failed on {folds} folds" + (
            f": {cause}" if cause else ""
        )
        self.folds = folds
        super().__init__(msg)

DimensionalityReductionError

Bases: DataScienceError

Error applying dimensionality reduction method.

Source code in dataexcept/datascience_exceptions/training.py

class DimensionalityReductionError(DataScienceError):
    """Error applying dimensionality reduction method."""

    def __init__(self, method: str, components: Optional[int] = None) -> None:
        if not isinstance(method, str):
            raise TypeError(f"method must be str, got {type(method).__name__}")
        if components is not None and not isinstance(components, int):
            raise TypeError(
                ("components must be int or None, " f"got {type(components).__name__}")
            )
        msg = f"Dimensionality reduction '{method}' failed" + (
            f" for {components} components" if components else ""
        )
        self.method = method
        self.components = components
        super().__init__(msg)

EarlyStoppingError

Bases: DataScienceError

Raised when training stops early based on a stopping criterion.

Parameters:

Name	Type	Description	Default
epoch	int	Epoch index where training stopped.	required
reason	Optional[str]	Optional reason for stopping.	None

Source code in dataexcept/datascience_exceptions/training.py

class EarlyStoppingError(DataScienceError):
    """Raised when training stops early based on a stopping criterion.

    Args:
        epoch: Epoch index where training stopped.
        reason: Optional reason for stopping.
    """

    def __init__(self, epoch: int, reason: Optional[str] = None) -> None:
        if not isinstance(epoch, int):
            raise TypeError(f"epoch must be int, got {type(epoch).__name__}")
        if reason is not None and not isinstance(reason, str):
            raise TypeError(f"reason must be str or None, got {type(reason).__name__}")

        msg = f"Training stopped early at epoch {epoch}"
        if reason:
            msg += f": {reason}"

        self.epoch = epoch
        self.reason = reason
        super().__init__(msg)

    def __str__(self) -> str:
        return f"[EarlyStoppingError:{self.epoch}] {self.message}"

ExperimentTrackingError

Bases: DataScienceError

Issues logging or retrieving experiment metadata.

Source code in dataexcept/datascience_exceptions/training.py

class ExperimentTrackingError(DataScienceError):
    """Issues logging or retrieving experiment metadata."""

    def __init__(self, run_id: str, cause: Optional[str] = None) -> None:
        if not isinstance(run_id, str):
            raise TypeError(f"run_id must be str, got {type(run_id).__name__}")
        msg = f"Experiment tracking failed for run '{run_id}'" + (
            f": {cause}" if cause else ""
        )
        self.run_id = run_id
        super().__init__(msg)

ExplainabilityError

Bases: DataScienceError

Raised when generating model explanations fails.

Parameters:

Name	Type	Description	Default
method	str	Explanation technique identifier.	required
details	Optional[str]	Optional description of the failure.	None

Source code in dataexcept/datascience_exceptions/training.py

class ExplainabilityError(DataScienceError):
    """Raised when generating model explanations fails.

    Args:
        method: Explanation technique identifier.
        details: Optional description of the failure.
    """

    def __init__(self, method: str, details: Optional[str] = None) -> None:
        if not isinstance(method, str):
            raise TypeError(f"method must be str, got {type(method).__name__}")
        if details is not None and not isinstance(details, str):
            raise TypeError(
                f"details must be str or None, got {type(details).__name__}"
            )
        msg = f"Explainability using '{method}' failed"
        if details:
            msg += f": {details}"
        self.method = method
        self.details = details
        super().__init__(msg)

    def __str__(self) -> str:
        return f"[ExplainabilityError:{self.method}] {self.message}"

FeatureScalingError

Bases: DataScienceError

Raised when scaling or standardization of features fails.

Parameters:

Name	Type	Description	Default
scaler	str	Name of the scaler or transformation used.	required
details	Optional[str]	Optional explanation of the failure.	None

Source code in dataexcept/datascience_exceptions/training.py

class FeatureScalingError(DataScienceError):
    """Raised when scaling or standardization of features fails.

    Args:
        scaler: Name of the scaler or transformation used.
        details: Optional explanation of the failure.
    """

    def __init__(self, scaler: str, details: Optional[str] = None) -> None:
        if not isinstance(scaler, str):
            raise TypeError(f"scaler must be str, got {type(scaler).__name__}")
        if details is not None and not isinstance(details, str):
            raise TypeError(
                f"details must be str or None, got {type(details).__name__}"
            )
        msg = f"Feature scaling with '{scaler}' failed"
        if details:
            msg += f": {details}"
        self.scaler = scaler
        self.details = details
        super().__init__(msg)

    def __str__(self) -> str:
        return f"[FeatureScalingError:{self.scaler}] {self.message}"

FeatureSelectionError

Bases: DataScienceError

Failure in feature selection procedure.

Source code in dataexcept/datascience_exceptions/training.py

class FeatureSelectionError(DataScienceError):
    """Failure in feature selection procedure."""

    def __init__(self, technique: str, details: Optional[str] = None) -> None:
        if not isinstance(technique, str):
            raise TypeError(f"technique must be str, got {type(technique).__name__}")
        msg = f"Feature selection failed using {technique}" + (
            f": {details}" if details else ""
        )
        self.technique = technique
        super().__init__(msg)

GPUOutOfMemoryError

Bases: DataScienceError

Model or tensor exceeds GPU memory capacity.

Source code in dataexcept/datascience_exceptions/training.py

class GPUOutOfMemoryError(DataScienceError):
    """Model or tensor exceeds GPU memory capacity."""

    def __init__(self, device: str, required: str, available: str) -> None:
        if not all(isinstance(v, str) for v in (device, required, available)):
            raise TypeError("device, required, available must be str")
        msg = f"GPU OOM on {device}: required={required}, available={available}"
        self.device = device
        self.required = required
        self.available = available
        super().__init__(msg)

HyperparameterError

Bases: DataScienceError

Raised for invalid hyperparameter settings.

Attributes:

Name	Type	Description
param		name of hyperparameter.
value		invalid value.

Source code in dataexcept/datascience_exceptions/training.py

class HyperparameterError(DataScienceError):
    """
    Raised for invalid hyperparameter settings.

    Attributes:
        param: name of hyperparameter.
        value: invalid value.
    """

    def __init__(self, param: str, value: Any, message: Optional[str] = None) -> None:
        if not isinstance(param, str):
            raise TypeError(f"param must be str, got {type(param).__name__}")

        if message is None:
            message = f"Invalid hyperparameter '{param}': {value!r}"

        self.param = param
        self.value = value
        super().__init__(message)

    def __str__(self) -> str:
        return f"[HyperparameterError:{self.param}] {self.message}"

HyperparameterTuningError

Bases: DataScienceError

Error during hyperparameter search or tuning.

Source code in dataexcept/datascience_exceptions/training.py

class HyperparameterTuningError(DataScienceError):
    """Error during hyperparameter search or tuning."""

    def __init__(self, method: str, details: Optional[str] = None) -> None:
        if not isinstance(method, str):
            raise TypeError(f"method must be str, got {type(method).__name__}")
        msg = f"Hyperparameter tuning ({method}) failed" + (
            f": {details}" if details else ""
        )
        self.method = method
        super().__init__(msg)

ModelCompatibilityError

Bases: DataScienceError

Raised when a model is incompatible with the runtime environment.

Parameters:

Name	Type	Description	Default
expected_version	str	Required model version.	required
found_version	str	Detected model version.	required
message	Optional[str]	Optional custom message.	None

Source code in dataexcept/datascience_exceptions/training.py

class ModelCompatibilityError(DataScienceError):
    """Raised when a model is incompatible with the runtime environment.

    Args:
        expected_version: Required model version.
        found_version: Detected model version.
        message: Optional custom message.
    """

    def __init__(
        self,
        expected_version: str,
        found_version: str,
        message: Optional[str] = None,
    ) -> None:
        if not isinstance(expected_version, str):
            raise TypeError(
                "expected_version must be str, got "
                f"{type(expected_version).__name__}"
            )
        if not isinstance(found_version, str):
            raise TypeError(
                "found_version must be str, got " f"{type(found_version).__name__}"
            )
        if message is not None and not isinstance(message, str):
            raise TypeError(
                f"message must be str or None, got {type(message).__name__}"
            )

        if message is None:
            msg = (
                f"Model requires version {expected_version}, "
                f"but found {found_version}"
            )
        else:
            msg = message

        self.expected_version = expected_version
        self.found_version = found_version
        super().__init__(msg)

    def __str__(self) -> str:
        return f"[ModelCompatibilityError] {self.message}"

ModelEvaluationError

Bases: DataScienceError

Raised during evaluation metrics computation.

Attributes:

Name	Type	Description
metric		name of the metric.
value		computed value.

Source code in dataexcept/datascience_exceptions/training.py

class ModelEvaluationError(DataScienceError):
    """
    Raised during evaluation metrics computation.

    Attributes:
        metric: name of the metric.
        value: computed value.
    """

    def __init__(
        self, metric: str, value: float, message: Optional[str] = None
    ) -> None:
        if not isinstance(metric, str):
            raise TypeError(f"metric must be str, got {type(metric).__name__}")
        if not is_number(value):
            raise TypeError(f"value must be number, got {type(value).__name__}")

        if message is None:
            message = f"Failed to compute metric '{metric}', got {value}"

        self.metric = metric
        self.value = float(value)
        super().__init__(message)

    def __str__(self) -> str:
        return f"[ModelEvaluationError:{self.metric}] {self.message}"

ModelInferenceError

Bases: DataScienceError

Raised when model inference fails.

Parameters:

Name	Type	Description	Default
model_type	str	Identifier of the model used for inference.	required
original	Exception	Underlying exception raised by the model.	required

Source code in dataexcept/datascience_exceptions/training.py

class ModelInferenceError(DataScienceError):
    """Raised when model inference fails.

    Args:
        model_type: Identifier of the model used for inference.
        original: Underlying exception raised by the model.
    """

    def __init__(self, model_type: str, original: Exception) -> None:
        if not isinstance(model_type, str):
            raise TypeError(f"model_type must be str, got {type(model_type).__name__}")
        if not isinstance(original, Exception):
            raise TypeError(
                f"original must be Exception, got {type(original).__name__}"
            )
        msg = f"Inference failed for model '{model_type}': {original}"
        self.model_type = model_type
        self.original = original
        super().__init__(msg)

    def __str__(self) -> str:
        return f"[ModelInferenceError:{self.model_type}] {self.message}"

ModelTrainingError

Bases: DataScienceError

Raised when model training fails.

Attributes:

Name	Type	Description
model_type		model class or name.
epoch		optional epoch index.

Source code in dataexcept/datascience_exceptions/training.py

class ModelTrainingError(DataScienceError):
    """
    Raised when model training fails.

    Attributes:
        model_type: model class or name.
        epoch: optional epoch index.
    """

    def __init__(
        self,
        model_type: str,
        epoch: Optional[int] = None,
        message: Optional[str] = None,
    ) -> None:
        if not isinstance(model_type, str):
            raise TypeError(f"model_type must be str, got {type(model_type).__name__}")
        if epoch is not None and not isinstance(epoch, int):
            raise TypeError(f"epoch must be int or None, got {type(epoch).__name__}")
        if message is not None and not isinstance(message, str):
            raise TypeError(
                f"message must be str or None, got {type(message).__name__}"
            )

        if message is None:
            msg = f"Training failed for model '{model_type}'"
            if epoch is not None:
                msg += f" at epoch {epoch}"  # include epoch
        else:
            msg = message

        self.model_type = model_type
        self.epoch = epoch
        super().__init__(msg)

    def __str__(self) -> str:
        base = f"{self.model_type}"
        if self.epoch is not None:
            base += f"@{self.epoch}"
        return f"[ModelTrainingError:{base}] {self.message}"

OverfittingError

Bases: DataScienceError

Raised when a model is overfitting the training data.

Parameters:

Name	Type	Description	Default
train_metric	float	Metric value on the training set.	required
val_metric	float	Metric value on the validation set.	required

Source code in dataexcept/datascience_exceptions/training.py

class OverfittingError(DataScienceError):
    """Raised when a model is overfitting the training data.

    Args:
        train_metric: Metric value on the training set.
        val_metric: Metric value on the validation set.
    """

    def __init__(self, train_metric: float, val_metric: float) -> None:
        if not is_number(train_metric):
            raise TypeError(
                ("train_metric must be numeric, got " f"{type(train_metric).__name__}")
            )
        if not is_number(val_metric):
            raise TypeError(
                f"val_metric must be numeric, got {type(val_metric).__name__}"
            )

        self.train_metric = float(train_metric)
        self.val_metric = float(val_metric)
        msg = (
            f"Overfitting detected: train={self.train_metric}, "
            f"val={self.val_metric}"
        )
        super().__init__(msg)

    def __str__(self) -> str:
        return f"[OverfittingError] {self.message}"

PredictionError

Bases: DataScienceError

Raised when making predictions fails.

Attributes:

Name	Type	Description
model_type		model used.
inputs		input data snapshot.

Source code in dataexcept/datascience_exceptions/training.py

class PredictionError(DataScienceError):
    """
    Raised when making predictions fails.

    Attributes:
        model_type: model used.
        inputs: input data snapshot.
    """

    def __init__(
        self, model_type: str, inputs: Any, message: Optional[str] = None
    ) -> None:
        if not isinstance(model_type, str):
            raise TypeError(f"model_type must be str, got {type(model_type).__name__}")

        if message is None:
            message = (
                f"Prediction failed for model '{model_type}' " f"with inputs {inputs!r}"
            )

        self.model_type = model_type
        self.inputs = inputs
        super().__init__(message)

    def __str__(self) -> str:
        return f"[PredictionError:{self.model_type}] {self.message}"

TrainingTimeoutError

Bases: ModelTrainingError

Raised when model training exceeds a time limit.

Source code in dataexcept/datascience_exceptions/training.py

class TrainingTimeoutError(ModelTrainingError):
    """Raised when model training exceeds a time limit."""

    def __init__(self, model_type: str, timeout: float) -> None:
        if not is_number(timeout):
            raise TypeError(f"timeout must be a number, got {type(timeout).__name__}")
        message = f"Training '{model_type}' exceeded timeout of {timeout} seconds"
        self.timeout = float(timeout)
        super().__init__(model_type=model_type, epoch=None, message=message)

    def __str__(self) -> str:
        return f"[TrainingTimeoutError] {self.message}"

UnderfittingError

Bases: DataScienceError

Raised when a model fails to capture patterns in the data.

Parameters:

Name	Type	Description	Default
train_metric	float	Metric value on the training set.	required
threshold	float	Minimum acceptable metric value.	required

Source code in dataexcept/datascience_exceptions/training.py

class UnderfittingError(DataScienceError):
    """Raised when a model fails to capture patterns in the data.

    Args:
        train_metric: Metric value on the training set.
        threshold: Minimum acceptable metric value.
    """

    def __init__(self, train_metric: float, threshold: float) -> None:
        if not is_number(train_metric):
            raise TypeError(
                ("train_metric must be numeric, got " f"{type(train_metric).__name__}")
            )
        if not is_number(threshold):
            raise TypeError(
                f"threshold must be numeric, got {type(threshold).__name__}"
            )

        self.train_metric = float(train_metric)
        self.threshold = float(threshold)
        msg = (
            f"Underfitting detected: training metric {self.train_metric} "
            f"< threshold {self.threshold}"
        )
        super().__init__(msg)

    def __str__(self) -> str:
        return f"[UnderfittingError] {self.message}"

Data engineering exceptions

dataengineering_exceptions

Custom exceptions for data engineering workflows.

DataEngineeringError

Bases: DataExceptError

Base exception for data engineering errors.

Source code in dataexcept/dataengineering_exceptions.py

class DataEngineeringError(DataExceptError):
    """Base exception for data engineering errors."""

    pass

ETLJobError

Bases: DataEngineeringError

Raised when an ETL job fails to complete successfully.

Source code in dataexcept/dataengineering_exceptions.py

class ETLJobError(DataEngineeringError):
    """Raised when an ETL job fails to complete successfully."""

    def __init__(self, job_name: str, message: Optional[str] = None) -> None:
        """Initialize ETLJobError.

        Args:
            job_name: Name of the ETL job.
            message: Optional custom error message.
        """
        self.job_name = job_name
        default = f"ETL job '{job_name}' failed"
        super().__init__(message or default)

SchemaEvolutionError

Bases: DataEngineeringError

Raised when database schema evolution fails.

Source code in dataexcept/dataengineering_exceptions.py

class SchemaEvolutionError(DataEngineeringError):
    """Raised when database schema evolution fails."""

    def __init__(self, schema_version: str, reason: Optional[str] = None) -> None:
        """Initialize SchemaEvolutionError.

        Args:
            schema_version: Version of the schema being applied.
            reason: Optional explanation of the failure.
        """
        self.schema_version = schema_version
        self.reason = reason
        msg = f"Schema evolution to {schema_version} failed"
        if reason:
            msg += f": {reason}"
        super().__init__(msg)

DataTransformationError

Bases: DataEngineeringError

Raised when a data transformation step fails.

Source code in dataexcept/dataengineering_exceptions.py

class DataTransformationError(DataEngineeringError):
    """Raised when a data transformation step fails."""

    def __init__(self, step: str, details: Optional[str] = None) -> None:
        """Initialize DataTransformationError.

        Args:
            step: Name of the transformation step.
            details: Optional details about the failure.
        """
        self.step = step
        self.details = details
        msg = f"Data transformation '{step}' failed"
        if details:
            msg += f": {details}"
        super().__init__(msg)

BatchProcessingError

Bases: DataEngineeringError

Raised when processing a data batch fails.

Source code in dataexcept/dataengineering_exceptions.py

class BatchProcessingError(DataEngineeringError):
    """Raised when processing a data batch fails."""

    def __init__(self, batch_id: str, original: Optional[Exception] = None) -> None:
        """Initialize BatchProcessingError.

        Args:
            batch_id: Identifier of the batch being processed.
            original: Optional underlying exception.
        """
        self.batch_id = batch_id
        self.original = original
        msg = f"Batch '{batch_id}' processing failed"
        if original:
            msg += f": {original}"
        super().__init__(msg)

DataWarehouseConnectionError

Bases: DataEngineeringError

Raised when a connection to a data warehouse cannot be established.

Source code in dataexcept/dataengineering_exceptions.py

class DataWarehouseConnectionError(DataEngineeringError):
    """Raised when a connection to a data warehouse cannot be established."""

    def __init__(self, warehouse: str, message: Optional[str] = None) -> None:
        """Initialize DataWarehouseConnectionError.

        Args:
            warehouse: Identifier of the data warehouse.
            message: Optional custom error message.
        """
        self.warehouse = warehouse
        default = f"Failed to connect to warehouse '{warehouse}'"
        super().__init__(message or default)

MissingPartitionError

Bases: DataEngineeringError

Raised when a required data partition is missing.

Source code in dataexcept/dataengineering_exceptions.py

class MissingPartitionError(DataEngineeringError):
    """Raised when a required data partition is missing."""

    def __init__(
        self, partition: str, location: str, message: Optional[str] = None
    ) -> None:
        """Initialize MissingPartitionError.

        Args:
            partition: Name of the missing partition.
            location: Data location checked for the partition.
            message: Optional custom error message.
        """
        self.partition = partition
        self.location = redact_if_url(location)
        default = f"Partition '{partition}' not found at {location}"
        super().__init__(message or default)

Pipeline exceptions

pipeline_exceptions

Additional exception classes for data pipeline workflows.

PipelineError

Bases: DataExceptError

Base exception for pipeline errors.

Source code in dataexcept/pipeline_exceptions.py

class PipelineError(DataExceptError):
    """Base exception for pipeline errors."""

    pass

PreprocessingError

Bases: PipelineError

Raised when a preprocessing step fails.

Source code in dataexcept/pipeline_exceptions.py

class PreprocessingError(PipelineError):
    """Raised when a preprocessing step fails."""

    def __init__(self, step_name: str, details: Optional[str] = None) -> None:
        default = f"Preprocessing failed at step: '{step_name}'."
        message = f"{default} Details: {details}" if details else default
        self.step_name = step_name
        self.details = details
        super().__init__(message)

FeaturePreprocessingError

Bases: PreprocessingError

Raised when feature engineering fails.

Source code in dataexcept/pipeline_exceptions.py

class FeaturePreprocessingError(PreprocessingError):
    """Raised when feature engineering fails."""

    def __init__(self, feature: str, reason: Optional[str] = None) -> None:
        # Assigned before super(): DataExceptError.__init__ sweeps the stored
        # strings for URLs, and anything set afterwards escapes that.
        self.feature = feature
        self.reason = reason
        super().__init__(step_name=f"feature_{feature}", details=reason)

StorageError

Bases: PipelineError

Raised when reading from or writing to storage fails.

Source code in dataexcept/pipeline_exceptions.py

class StorageError(PipelineError):
    """Raised when reading from or writing to storage fails."""

    def __init__(
        self,
        location: str,
        operation: str,
        message: Optional[str] = None,
    ) -> None:
        default = f"Storage {operation} failed at location: '{location}'."
        self.location = redact_if_url(location)
        self.operation = operation
        super().__init__(message or default)

PipelineNotificationError

Bases: PipelineError

Raised when sending a notification fails.

Source code in dataexcept/pipeline_exceptions.py

class PipelineNotificationError(PipelineError):
    """Raised when sending a notification fails."""

    def __init__(
        self,
        channel: str,
        payload: Any,
        message: Optional[str] = None,
    ) -> None:
        default = f"Notification via '{channel}' failed."
        self.channel = channel
        self.payload = payload
        super().__init__(message or default)

RetryLimitExceededError

Bases: PipelineError

Raised when an operation is retried too many times.

Source code in dataexcept/pipeline_exceptions.py

class RetryLimitExceededError(PipelineError):
    """Raised when an operation is retried too many times."""

    def __init__(
        self,
        operation: str,
        retries: int,
        message: Optional[str] = None,
    ) -> None:
        default = (
            "Retry limit exceeded for operation "
            f"'{operation}' after {retries} attempts."
        )
        self.operation = operation
        self.retries = retries
        super().__init__(message or default)

ExternalServiceError

Bases: PipelineError

General failure when calling an external service.

Source code in dataexcept/pipeline_exceptions.py

class ExternalServiceError(PipelineError):
    """General failure when calling an external service."""

    def __init__(
        self,
        service_name: str,
        status_code: Optional[int] = None,
        response: Optional[Any] = None,
        message: Optional[str] = None,
    ) -> None:
        default = f"Call to external service '{service_name}' failed."
        self.service_name = service_name
        self.status_code = status_code
        self.response = response
        super().__init__(message or default)

ServiceAuthenticationError

Bases: ExternalServiceError

Authentication to an external service failed.

Source code in dataexcept/pipeline_exceptions.py

class ServiceAuthenticationError(ExternalServiceError):
    """Authentication to an external service failed."""

    def __init__(
        self,
        service_name: str,
        message: Optional[str] = None,
    ) -> None:
        default = f"Authentication failed for service '{service_name}'."
        super().__init__(service_name=service_name, message=message or default)

ServiceAuthorizationError

Bases: ExternalServiceError

Authorization was denied by an external service.

Source code in dataexcept/pipeline_exceptions.py

class ServiceAuthorizationError(ExternalServiceError):
    """Authorization was denied by an external service."""

    def __init__(
        self,
        service_name: str,
        message: Optional[str] = None,
    ) -> None:
        default = f"Authorization denied for service '{service_name}'."
        super().__init__(service_name=service_name, message=message or default)

ServiceTimeoutError

Bases: ExternalServiceError

A call to an external service exceeded the allotted time.

Source code in dataexcept/pipeline_exceptions.py

class ServiceTimeoutError(ExternalServiceError):
    """A call to an external service exceeded the allotted time."""

    def __init__(
        self,
        service_name: str,
        timeout_seconds: Optional[float] = None,
    ) -> None:
        default = (
            "Operation timed out after "
            f"{timeout_seconds}s on service '{service_name}'."
        )
        self.timeout_seconds = timeout_seconds
        super().__init__(service_name=service_name, message=default)

ApiError

Bases: PipelineError

Failure calling a REST API endpoint.

Source code in dataexcept/pipeline_exceptions.py

class ApiError(PipelineError):
    """Failure calling a REST API endpoint."""

    def __init__(
        self,
        endpoint: str,
        status_code: Optional[int] = None,
        message: Optional[str] = None,
    ) -> None:
        # An endpoint URL may authenticate through a query parameter.
        self.endpoint = redact_url(endpoint)
        default = f"API call failed: {self.endpoint}"
        if status_code is not None:
            default += f" (status {status_code})"
        self.status_code = status_code
        super().__init__(message or default)

TimeDeltaTooLargeError

Bases: PipelineError

The time span between records exceeded a threshold.

Source code in dataexcept/pipeline_exceptions.py

class TimeDeltaTooLargeError(PipelineError):
    """The time span between records exceeded a threshold."""

    def __init__(
        self,
        user: str,
        delta_minutes: float,
        message: Optional[str] = None,
    ) -> None:
        default = f"Time delta {delta_minutes}m too large for user {user}"
        self.user = user
        self.delta_minutes = delta_minutes
        super().__init__(message or default)

TypeCheckError

Bases: PipelineError

Invalid type detected during recursive type inspection.

Source code in dataexcept/pipeline_exceptions.py

class TypeCheckError(PipelineError):
    """Invalid type detected during recursive type inspection."""

DataFetchError

Bases: PipelineError

Failed to fetch data from a storage backend.

Source code in dataexcept/pipeline_exceptions.py

class DataFetchError(PipelineError):
    """Failed to fetch data from a storage backend."""

    def __init__(
        self,
        source: str,
        cid: str,
        message: Optional[str] = None,
    ) -> None:
        default = f"Failed to fetch '{source}' data for cid={cid}"
        self.source = redact_if_url(source)
        self.cid = cid
        super().__init__(message or default)

Database exceptions

database_exceptions

Custom exceptions for database operations.

DatabaseError

Bases: DataExceptError

Base exception for database-related errors.

Source code in dataexcept/database_exceptions.py

class DatabaseError(DataExceptError):
    """Base exception for database-related errors."""

    pass

DatabaseConnectionError

Bases: DatabaseError

Raised when connecting to the database fails.

Source code in dataexcept/database_exceptions.py

class DatabaseConnectionError(DatabaseError):
    """Raised when connecting to the database fails."""

    def __init__(self, db_url: str, message: str | None = None) -> None:
        """Initialize DatabaseConnectionError.

        Args:
            db_url: Database connection URL.
            message: Optional custom error message.
        """
        # A connection URL routinely carries a username and password.
        self.db_url = redact_url(db_url)
        default = f"Failed to connect to database at '{self.db_url}'"
        super().__init__(message or default)

QueryExecutionError

Bases: DatabaseError

Raised when a database query execution fails.

Source code in dataexcept/database_exceptions.py

class QueryExecutionError(DatabaseError):
    """Raised when a database query execution fails."""

    def __init__(self, query: str, original: Exception | None = None) -> None:
        """Initialize QueryExecutionError.

        Args:
            query: SQL query string.
            original: Optional underlying exception.
        """
        self.query = query
        self.original = original
        msg = f"Query failed: {query}"
        if original:
            msg += f" ({original})"
        super().__init__(msg)

TransactionError

Bases: DatabaseError

Raised when a database transaction fails.

Source code in dataexcept/database_exceptions.py

class TransactionError(DatabaseError):
    """Raised when a database transaction fails."""

    def __init__(
        self,
        transaction_id: str | None = None,
        message: str | None = None,
    ) -> None:
        """Initialize TransactionError.

        Args:
            transaction_id: Identifier for the transaction.
            message: Optional custom error message.
        """
        self.transaction_id = transaction_id
        default = "Database transaction failed"
        if transaction_id:
            default += f" (id={transaction_id})"
        super().__init__(message or default)

I/O exceptions

io_exceptions

Custom exceptions for file and I/O operations.

CustomIOError

Bases: DataExceptError

Base exception for I/O errors.

Source code in dataexcept/io_exceptions.py

class CustomIOError(DataExceptError):
    """Base exception for I/O errors."""

    pass

FileReadError

Bases: CustomIOError

Raised when reading a file fails.

Source code in dataexcept/io_exceptions.py

class FileReadError(CustomIOError):
    """Raised when reading a file fails."""

    def __init__(self, path: str, original: Exception | None = None) -> None:
        """Initialize FileReadError.

        Args:
            path: File path that could not be read.
            original: Optional underlying exception.
        """
        self.path = redact_if_url(path)
        self.original = original
        msg = f"Failed to read file '{path}'"
        if original:
            msg += f": {original}"
        super().__init__(msg)

FileWriteError

Bases: CustomIOError

Raised when writing to a file fails.

Source code in dataexcept/io_exceptions.py

class FileWriteError(CustomIOError):
    """Raised when writing to a file fails."""

    def __init__(self, path: str, original: Exception | None = None) -> None:
        """Initialize FileWriteError.

        Args:
            path: File path that could not be written to.
            original: Optional underlying exception.
        """
        self.path = redact_if_url(path)
        self.original = original
        msg = f"Failed to write file '{path}'"
        if original:
            msg += f": {original}"
        super().__init__(msg)

FileLockError

Bases: CustomIOError

Raised when a file lock cannot be acquired.

Source code in dataexcept/io_exceptions.py

class FileLockError(CustomIOError):
    """Raised when a file lock cannot be acquired."""

    def __init__(self, path: str) -> None:
        """Initialize FileLockError.

        Args:
            path: Path of the lock file.
        """
        self.path = redact_if_url(path)
        super().__init__(f"Unable to obtain lock for '{path}'")

Network exceptions

network_exceptions

Custom exceptions for network operations.

NetworkError

Bases: DataExceptError

Base exception for network-related errors.

Example

from dataexcept.network_exceptions import NetworkError try: ... raise NetworkError("Something went wrong") ... except NetworkError: ... print("Caught network error") Caught network error

Source code in dataexcept/network_exceptions.py

class NetworkError(DataExceptError):
    """Base exception for network-related errors.

    Example:
        >>> from dataexcept.network_exceptions import NetworkError
        >>> try:
        ...     raise NetworkError("Something went wrong")
        ... except NetworkError:
        ...     print("Caught network error")
        Caught network error
    """

    pass

HostUnreachableError

Bases: NetworkError

Raised when a remote host cannot be reached.

Example

from dataexcept.network_exceptions import HostUnreachableError try: ... raise HostUnreachableError("api.example.com") ... except HostUnreachableError as exc: ... print(exc) Host 'api.example.com' is unreachable

Source code in dataexcept/network_exceptions.py

class HostUnreachableError(NetworkError):
    """Raised when a remote host cannot be reached.

    Example:
        >>> from dataexcept.network_exceptions import HostUnreachableError
        >>> try:
        ...     raise HostUnreachableError("api.example.com")
        ... except HostUnreachableError as exc:
        ...     print(exc)
        Host 'api.example.com' is unreachable
    """

    def __init__(self, host: str, message: str | None = None) -> None:
        """Initialize HostUnreachableError.

        Args:
            host: Host address that could not be reached.
            message: Optional custom error message.
        """
        self.host = host
        default = f"Host '{host}' is unreachable"
        super().__init__(message or default)

ConnectionTimeoutError

Bases: NetworkError

Raised when a network connection attempt times out.

Example

from dataexcept.network_exceptions import ConnectionTimeoutError try: ... raise ConnectionTimeoutError("api.example.com", 30) ... except ConnectionTimeoutError as exc: ... print(exc) Connection to 'api.example.com' timed out after 30 seconds

Source code in dataexcept/network_exceptions.py

class ConnectionTimeoutError(NetworkError):
    """Raised when a network connection attempt times out.

    Example:
        >>> from dataexcept.network_exceptions import ConnectionTimeoutError
        >>> try:
        ...     raise ConnectionTimeoutError("api.example.com", 30)
        ... except ConnectionTimeoutError as exc:
        ...     print(exc)
        Connection to 'api.example.com' timed out after 30 seconds
    """

    def __init__(self, host: str, timeout: float) -> None:
        """Initialize ConnectionTimeoutError.

        Args:
            host: Host address.
            timeout: Timeout in seconds.
        """
        self.host = host
        self.timeout = timeout
        msg = f"Connection to '{host}' timed out after {timeout} seconds"
        super().__init__(msg)

ProtocolError

Bases: NetworkError

Raised when an unexpected protocol error occurs.

Example

from dataexcept.network_exceptions import ProtocolError try: ... raise ProtocolError("HTTP", "Invalid status line") ... except ProtocolError as exc: ... print(exc) Protocol error in HTTP: Invalid status line

Source code in dataexcept/network_exceptions.py

class ProtocolError(NetworkError):
    """Raised when an unexpected protocol error occurs.

    Example:
        >>> from dataexcept.network_exceptions import ProtocolError
        >>> try:
        ...     raise ProtocolError("HTTP", "Invalid status line")
        ... except ProtocolError as exc:
        ...     print(exc)
        Protocol error in HTTP: Invalid status line
    """

    def __init__(self, protocol: str, details: str | None = None) -> None:
        """Initialize ProtocolError.

        Args:
            protocol: Protocol name (e.g., HTTP).
            details: Optional additional details about the failure.
        """
        self.protocol = protocol
        self.details = details
        msg = f"Protocol error in {protocol}"
        if details:
            msg += f": {details}"
        super().__init__(msg)

pandas exceptions

pandas_exceptions

Custom exceptions for pandas DataFrame operations.

PandasError

Bases: DataExceptError

Base exception for pandas-related errors.

Source code in dataexcept/pandas_exceptions.py

class PandasError(DataExceptError):
    """Base exception for pandas-related errors."""

MissingColumnError

Bases: PandasError

Raised when a required DataFrame column is missing.

Parameters:

Name	Type	Description	Default
column	str	Name of the missing column.	required
dataframe	Optional[str]	Optional name of the DataFrame being inspected.	None

Source code in dataexcept/pandas_exceptions.py

class MissingColumnError(PandasError):
    """Raised when a required DataFrame column is missing.

    Args:
        column: Name of the missing column.
        dataframe: Optional name of the DataFrame being inspected.
    """

    def __init__(self, column: str, dataframe: Optional[str] = None) -> None:
        if not isinstance(column, str):
            raise TypeError(f"column must be str, got {type(column).__name__}")
        if dataframe is not None and not isinstance(dataframe, str):
            raise TypeError(
                "dataframe must be str or None, " f"got {type(dataframe).__name__}"
            )

        self.column = column
        self.dataframe = dataframe
        name = f" in DataFrame '{dataframe}'" if dataframe else ""
        msg = f"Missing required column '{column}'{name}"
        super().__init__(msg)

    def __str__(self) -> str:
        return f"[MissingColumnError] {self.args[0]}"

DtypeMismatchError

Bases: PandasError

Raised when a column has an unexpected dtype.

Parameters:

Name	Type	Description	Default
column	str	Name of the column.	required
expected	Sequence[str]	Sequence of allowed dtypes.	required
found	str	Detected dtype for the column.	required

Source code in dataexcept/pandas_exceptions.py

class DtypeMismatchError(PandasError):
    """Raised when a column has an unexpected dtype.

    Args:
        column: Name of the column.
        expected: Sequence of allowed dtypes.
        found: Detected dtype for the column.
    """

    def __init__(self, column: str, expected: Sequence[str], found: str) -> None:
        if not isinstance(column, str):
            raise TypeError(f"column must be str, got {type(column).__name__}")
        if not isinstance(found, str):
            raise TypeError(f"found must be str, got {type(found).__name__}")
        if not isinstance(expected, Sequence) or isinstance(expected, str):
            raise TypeError("expected must be a sequence of strings")
        if not all(isinstance(dt, str) for dt in expected):
            raise TypeError("expected must contain strings")

        self.column = column
        self.expected = list(expected)
        self.found = found
        expected_fmt = ", ".join(self.expected)
        msg = f"Column '{column}' has dtype {found}; expected {expected_fmt}"
        super().__init__(msg)

    def __str__(self) -> str:
        return f"[DtypeMismatchError:{self.column}] {self.args[0]}"

IndexAlignmentError

Bases: PandasError

Raised when DataFrame indices are misaligned for an operation.

Parameters:

Name	Type	Description	Default
details	Optional[str]	Optional details about the misalignment.	None

Source code in dataexcept/pandas_exceptions.py

class IndexAlignmentError(PandasError):
    """Raised when DataFrame indices are misaligned for an operation.

    Args:
        details: Optional details about the misalignment.
    """

    def __init__(self, details: Optional[str] = None) -> None:
        if details is not None and not isinstance(details, str):
            raise TypeError(
                f"details must be str or None, got {type(details).__name__}"
            )
        msg = "DataFrame indices are misaligned"
        if details:
            msg += f": {details}"
        self.details = details
        super().__init__(msg)

    def __str__(self) -> str:
        return f"[IndexAlignmentError] {self.args[0]}"

MergeKeyError

Bases: PandasError

Raised when merging DataFrames fails due to key issues.

Parameters:

Name	Type	Description	Default
left_keys	Sequence[str]	Keys from the left DataFrame.	required
right_keys	Sequence[str]	Keys from the right DataFrame.	required

Source code in dataexcept/pandas_exceptions.py

class MergeKeyError(PandasError):
    """Raised when merging DataFrames fails due to key issues.

    Args:
        left_keys: Keys from the left DataFrame.
        right_keys: Keys from the right DataFrame.
    """

    def __init__(self, left_keys: Sequence[str], right_keys: Sequence[str]) -> None:
        # A bare string is a sequence of strings, so "id" would silently become
        # ['i', 'd']. Reject it, as DtypeMismatchError already does.
        for name, keys in (("left_keys", left_keys), ("right_keys", right_keys)):
            if isinstance(keys, str) or not all(isinstance(k, str) for k in keys):
                raise TypeError(f"{name} must be a sequence of strings, not a string")
        self.left_keys = list(left_keys)
        self.right_keys = list(right_keys)
        msg = f"Failed to merge on keys {self.left_keys} and {self.right_keys}"
        super().__init__(msg)

    def __str__(self) -> str:
        return f"[MergeKeyError] {self.args[0]}"

PandasIOError

Bases: PandasError

Raised when reading from or writing to disk with pandas fails.

Parameters:

Name	Type	Description	Default
path	str	File path involved in the operation.	required
original	Exception	The underlying exception that was raised.	required

Source code in dataexcept/pandas_exceptions.py

class PandasIOError(PandasError):
    """Raised when reading from or writing to disk with pandas fails.

    Args:
        path: File path involved in the operation.
        original: The underlying exception that was raised.
    """

    def __init__(self, path: str, original: Exception) -> None:
        if not isinstance(path, str):
            raise TypeError(f"path must be str, got {type(path).__name__}")
        if not isinstance(original, Exception):
            raise TypeError(
                f"original must be Exception, got {type(original).__name__}"
            )
        self.path = redact_if_url(path)
        self.original = original
        msg = f"Pandas I/O operation failed on {path!r}: {original}"
        super().__init__(msg)

    def __str__(self) -> str:
        return f"[PandasIOError] {self.args[0]}"

Security exceptions

security_exceptions

Custom exceptions for security-related operations.

SecurityError

Bases: DataExceptError

Base exception for security errors.

Source code in dataexcept/security_exceptions.py

class SecurityError(DataExceptError):
    """Base exception for security errors."""

    pass

EncryptionError

Bases: SecurityError

Raised when data encryption fails.

Source code in dataexcept/security_exceptions.py

class EncryptionError(SecurityError):
    """Raised when data encryption fails."""

    def __init__(self, algorithm: str, message: str | None = None) -> None:
        """Initialize EncryptionError.

        Args:
            algorithm: Name of the encryption algorithm.
            message: Optional custom error message.
        """
        self.algorithm = algorithm
        default = f"Encryption failed using {algorithm}"
        super().__init__(message or default)

DecryptionError

Bases: SecurityError

Raised when data decryption fails.

Source code in dataexcept/security_exceptions.py

class DecryptionError(SecurityError):
    """Raised when data decryption fails."""

    def __init__(self, algorithm: str, message: str | None = None) -> None:
        """Initialize DecryptionError.

        Args:
            algorithm: Name of the decryption algorithm.
            message: Optional custom error message.
        """
        self.algorithm = algorithm
        default = f"Decryption failed using {algorithm}"
        super().__init__(message or default)

InvalidTokenError

Bases: SecurityError

Raised when an authentication token is invalid or expired.

Source code in dataexcept/security_exceptions.py

class InvalidTokenError(SecurityError):
    """Raised when an authentication token is invalid or expired."""

    def __init__(
        self,
        token: str | None = None,
        message: str | None = None,
    ) -> None:
        """Initialize InvalidTokenError.

        Args:
            token: The problematic token.
            message: Optional custom error message.
        """
        # The raw token is never stored or rendered: this exception is often
        # logged, and the caller already holds the value it passed in.
        self.token = redact_secret(token)
        default = "Invalid authentication token"
        if token:
            default += f": {self.token}"
        # The library was handed the secret, so it can be removed even from a
        # message the caller wrote themselves.
        super().__init__(remove_secret(message or default, token))

Logging helpers

logging_helpers

Helper functions for logging exceptions consistently.

log_exception

log_exception(exc: Exception, logger: Optional[Logger] = None, level: int = logging.ERROR, context: Context | None = None) -> None

Log exc at the given log level using logger.

If logger is None a module level logger is used.

DataExcept redacts what it renders, but a wrapped third-party exception renders itself: an HTTP client's error may quote the credential-bearing URL it was called with, and exc_info makes logging print that whole chain. When the chain contains a URL the traceback is formatted and scrubbed here; otherwise the structured exc_info path is used unchanged, so ordinary exceptions keep the shape log aggregators expect.

Source code in dataexcept/logging_helpers.py

def log_exception(
    exc: Exception,
    logger: Optional[logging.Logger] = None,
    level: int = logging.ERROR,
    context: Context | None = None,
) -> None:
    """Log *exc* at the given log *level* using *logger*.

    If *logger* is ``None`` a module level logger is used.

    DataExcept redacts what it renders, but a wrapped third-party exception
    renders itself: an HTTP client's error may quote the credential-bearing URL
    it was called with, and ``exc_info`` makes logging print that whole chain.
    When the chain contains a URL the traceback is formatted and scrubbed here;
    otherwise the structured ``exc_info`` path is used unchanged, so ordinary
    exceptions keep the shape log aggregators expect.
    """
    if logger is None:
        logger = logging.getLogger(__name__)
    extra = _build_extra(context)

    if _chain_mentions_a_url(exc):
        formatted = "".join(
            traceback.format_exception(type(exc), exc, exc.__traceback__)
        )
        keep_path = getattr(type(exc), "_keep_url_path", True)
        scrubbed = redact_urls_in_text(formatted, keep_path=keep_path).rstrip()
        logger.log(level, "%s\n%s", exc, scrubbed, extra=extra)
        return

    exc_info = (type(exc), exc, exc.__traceback__)
    logger.log(level, "%s", exc, exc_info=exc_info, extra=extra)

log_and_raise

log_and_raise(logger: Optional[Logger] = None, level: int = logging.ERROR, context: Context | None = None) -> Iterator[None]

Context manager that logs and re-raises exceptions preserving traceback.

Source code in dataexcept/logging_helpers.py

@contextlib.contextmanager
def log_and_raise(
    logger: Optional[logging.Logger] = None,
    level: int = logging.ERROR,
    context: Context | None = None,
) -> Iterator[None]:
    """Context manager that logs and re-raises exceptions preserving traceback."""
    try:
        yield
    except Exception as exc:
        log_exception(exc, logger=logger, level=level, context=context)
        raise

log_then_raise

log_then_raise(exc: Exception, logger: Optional[Logger] = None, level: int = logging.ERROR, context: Context | None = None) -> None

Log exc and immediately raise it.

This helper mirrors the pre-context-manager API for scenarios where adding a with block would be too intrusive. Prefer :func:log_and_raise whenever possible so tracebacks remain untouched.

Source code in dataexcept/logging_helpers.py

def log_then_raise(
    exc: Exception,
    logger: Optional[logging.Logger] = None,
    level: int = logging.ERROR,
    context: Context | None = None,
) -> None:
    """Log *exc* and immediately raise it.

    This helper mirrors the pre-context-manager API for scenarios where adding a
    ``with`` block would be too intrusive. Prefer :func:`log_and_raise` whenever
    possible so tracebacks remain untouched.
    """
    log_exception(exc, logger=logger, level=level, context=context)
    raise exc

Command-line entry point

__main__

Command line interface for the DataExcept package.

main

main(argv: list[str] | None = None) -> None

Entry point for the dataexcept command.

Source code in dataexcept/__main__.py

def main(argv: list[str] | None = None) -> None:
    """Entry point for the ``dataexcept`` command."""
    parser = argparse.ArgumentParser(
        # Without this, `python -m dataexcept --version` reports "__main__.py".
        prog="dataexcept",
        description="Utilities for DataExcept",
    )
    parser.add_argument(
        "--version",
        action="version",
        version=f"%(prog)s {__version__}",
    )
    subparsers = parser.add_subparsers(dest="command")
    subparsers.add_parser("list", help="List available exception classes")

    args = parser.parse_args(argv)

    if args.command == "list":
        _list_exceptions()
    else:  # pragma: no cover - help message
        parser.print_help()