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."""

    _keep_url_path = True
    _default_failure_metadata = FailureMetadata()

    def __init__(self, *args: Any) -> None:
        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:]

        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)
        for attribute in _CAUSE_ATTRIBUTES:
            candidate = getattr(self, attribute, None)
            if isinstance(candidate, BaseException):
                self.__cause__ = candidate
                break

    @property
    def failure_metadata(self) -> FailureMetadata:
        override = self.__dict__.get("_failure_metadata_override")
        if isinstance(override, FailureMetadata):
            return override
        return type(self)._default_failure_metadata

    @property
    def failure_kind(self) -> FailureKind:
        return self.failure_metadata.failure_kind

    @property
    def retryable(self) -> bool | None:
        return self.failure_metadata.retryable

    @property
    def retry_after_seconds(self) -> float | None:
        return self.failure_metadata.retry_after_seconds

    def with_failure_metadata(self, metadata: FailureMetadata) -> "DataExceptError":
        """Attach backend-informed metadata and return ``self`` for chaining."""
        if not isinstance(metadata, FailureMetadata):
            raise TypeError("metadata must be a FailureMetadata instance")
        self._failure_metadata_override = metadata
        return self

    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__,
            ),
        )

with_failure_metadata

with_failure_metadata(metadata: FailureMetadata) -> 'DataExceptError'

Attach backend-informed metadata and return self for chaining.

Source code in dataexcept/base.py

def with_failure_metadata(self, metadata: FailureMetadata) -> "DataExceptError":
    """Attach backend-informed metadata and return ``self`` for chaining."""
    if not isinstance(metadata, FailureMetadata):
        raise TypeError("metadata must be a FailureMetadata instance")
    self._failure_metadata_override = metadata
    return self

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.

Source code in dataexcept/base.py

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

    __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)

BrokerConnectionError

Bases: MessageBrokerError

Raised when a connection to the broker cannot be established.

Source code in dataexcept/broker_exceptions.py

class BrokerConnectionError(MessageBrokerError):
    """Raised when a connection to the broker cannot be established."""

    def __init__(
        self,
        broker: str,
        message: Optional[str] = None,
        *,
        cause: Optional[Exception] = None,
    ) -> None:
        """Initialize BrokerConnectionError.

        Args:
            broker: The broker being connected to. A bootstrap address, a
                cluster name, or a URL -- redacted when it is a URL, since one
                routinely carries credentials.
            message: Optional custom error message.
            cause: Optional underlying exception, also set as ``__cause__``.
        """
        self.broker = redact_if_url(broker)
        self.cause = resolve_cause(cause=cause)
        default = f"Failed to connect to message broker '{self.broker}'"
        if self.cause:
            default += f": {self.cause}"
        super().__init__(message or default)

BrokerTimeoutError

Bases: MessageBrokerError

Raised when a broker operation exceeds its time limit.

Source code in dataexcept/broker_exceptions.py

class BrokerTimeoutError(MessageBrokerError):
    """Raised when a broker operation exceeds its time limit."""

    def __init__(
        self,
        broker: str,
        operation: Optional[str] = None,
        timeout_seconds: Optional[float] = None,
        message: Optional[str] = None,
        *,
        cause: Optional[Exception] = None,
    ) -> None:
        """Initialize BrokerTimeoutError.

        Args:
            broker: The broker the operation was waiting on.
            operation: What was being attempted, such as ``"publish"``.
            timeout_seconds: The limit that was exceeded.
            message: Optional custom error message.
            cause: Optional underlying exception, also set as ``__cause__``.
        """
        self.broker = redact_if_url(broker)
        self.operation = operation
        self.timeout_seconds = timeout_seconds
        self.cause = resolve_cause(cause=cause)
        attempted = (
            f"Broker operation '{operation}'" if operation else "Broker operation"
        )
        default = f"{attempted} timed out"
        if timeout_seconds is not None:
            default += f" after {timeout_seconds}s"
        default += _on_broker(self.broker)
        if self.cause:
            default += f": {self.cause}"
        super().__init__(message or default)

MessageAcknowledgementError

Bases: MessageBrokerError

Raised when acknowledging or committing a message fails.

Distinct from a consume failure on purpose: the message was read and processed, and it is the record of that which did not stick -- so it will be delivered again.

Source code in dataexcept/broker_exceptions.py

class MessageAcknowledgementError(MessageBrokerError):
    """Raised when acknowledging or committing a message fails.

    Distinct from a consume failure on purpose: the message was read and
    processed, and it is the record of that which did not stick -- so it will
    be delivered again.
    """

    def __init__(
        self,
        topic: str,
        broker: Optional[str] = None,
        partition: Optional[int] = None,
        offset: Optional[int] = None,
        consumer_group: Optional[str] = None,
        message: Optional[str] = None,
        *,
        cause: Optional[Exception] = None,
    ) -> None:
        """Initialize MessageAcknowledgementError.

        Args:
            topic: The topic, queue or subject the message came from.
            broker: Optional broker the acknowledgement was sent to.
            partition: Optional partition the message was read from.
            offset: Optional offset that failed to commit.
            consumer_group: Optional consumer group committing the offset.
            message: Optional custom error message.
            cause: Optional underlying exception, also set as ``__cause__``.
        """
        self.topic = topic
        self.broker = redact_if_url(broker)
        self.partition = partition
        self.offset = offset
        self.consumer_group = consumer_group
        self.cause = resolve_cause(cause=cause)
        position = _describe_position(topic, partition, offset, consumer_group)
        default = f"Failed to acknowledge {position}{_on_broker(self.broker)}"
        if self.cause:
            default += f": {self.cause}"
        super().__init__(message or default)

MessageBrokerError

Bases: DataExceptError

Base exception for message-broker failures.

Catching this catches every broker failure the library raises, without catching a database or HTTP one.

Source code in dataexcept/broker_exceptions.py

class MessageBrokerError(DataExceptError):
    """Base exception for message-broker failures.

    Catching this catches every broker failure the library raises, without
    catching a database or HTTP one.
    """

    pass

MessageConsumeError

Bases: MessageBrokerError

Raised when consuming a message fails.

Source code in dataexcept/broker_exceptions.py

class MessageConsumeError(MessageBrokerError):
    """Raised when consuming a message fails."""

    def __init__(
        self,
        topic: str,
        broker: Optional[str] = None,
        partition: Optional[int] = None,
        offset: Optional[int] = None,
        consumer_group: Optional[str] = None,
        message: Optional[str] = None,
        *,
        cause: Optional[Exception] = None,
    ) -> None:
        """Initialize MessageConsumeError.

        Args:
            topic: The topic, queue or subject consumed from.
            broker: Optional broker the consumer was reading from.
            partition: Optional partition being read.
            offset: Optional offset the failure happened at.
            consumer_group: Optional consumer group the reader belongs to.
            message: Optional custom error message.
            cause: Optional underlying exception, also set as ``__cause__``.
        """
        self.topic = topic
        self.broker = redact_if_url(broker)
        self.partition = partition
        self.offset = offset
        self.consumer_group = consumer_group
        self.cause = resolve_cause(cause=cause)
        position = _describe_position(topic, partition, offset, consumer_group)
        default = f"Failed to consume from {position}{_on_broker(self.broker)}"
        if self.cause:
            default += f": {self.cause}"
        super().__init__(message or default)

MessagePublishError

Bases: MessageBrokerError

Raised when publishing a message fails.

Source code in dataexcept/broker_exceptions.py

class MessagePublishError(MessageBrokerError):
    """Raised when publishing a message fails."""

    def __init__(
        self,
        topic: str,
        broker: Optional[str] = None,
        partition: Optional[int] = None,
        message: Optional[str] = None,
        *,
        cause: Optional[Exception] = None,
    ) -> None:
        """Initialize MessagePublishError.

        Args:
            topic: The topic, queue or subject published to.
            broker: Optional broker the publish was addressed to.
            partition: Optional partition the message was keyed to.
            message: Optional custom error message.
            cause: Optional underlying exception, also set as ``__cause__``.
        """
        self.topic = topic
        self.broker = redact_if_url(broker)
        self.partition = partition
        self.cause = resolve_cause(cause=cause)
        position = _describe_position(topic, partition)
        default = f"Failed to publish to {position}{_on_broker(self.broker)}"
        if self.cause:
            default += f": {self.cause}"
        super().__init__(message or default)

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,
        *,
        cause: Exception | None = None,
    ) -> None:
        """Initialize QueryExecutionError.

        Args:
            query: SQL query string.
            original: Legacy alias for ``cause``.
            cause: Optional underlying exception.
        """
        resolved = resolve_cause(cause=cause, original=original)
        self.query = query
        self.original = resolved
        self.cause = resolved
        msg = f"Query failed: {query}"
        if resolved:
            msg += f" ({resolved})"
        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,
        *,
        cause: Exception | None = None,
    ) -> None:
        """Initialize TransactionError.

        Args:
            transaction_id: Identifier for the transaction.
            message: Optional custom error message.
            cause: Optional underlying exception.
        """
        self.transaction_id = transaction_id
        self.cause = cause
        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."""

    _default_failure_metadata = FailureMetadata(
        failure_kind="permanent",
        retryable=False,
    )

    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."""

    _default_failure_metadata = FailureMetadata(
        failure_kind="permanent",
        retryable=False,
    )

    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.

Parameters:

Name	Type	Description	Default
data	bytes | None	The offending bytes. Optional, and kept verbatim when given: pass it only when the payload is yours and safe to retain.	None
format	str | None	What the payload was meant to be, such as "json".	None
message	str | None	Overrides the generated message entirely.	None
source	str | None	Where the payload came from -- an endpoint, path or queue. Redacted when it is a URL, left alone when it is a file path.	None
preview	str | bytes | bytearray | None	A bounded excerpt of the payload, truncated to :data:dataexcept._previews.MAX_PREVIEW_LENGTH characters.	None
cause	Exception | None	The underlying exception, also set as __cause__.	None

Source code in dataexcept/exceptions/parsing.py

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

    Args:
        data: The offending bytes. Optional, and kept verbatim when given:
            pass it only when the payload is yours and safe to retain.
        format: What the payload was meant to be, such as ``"json"``.
        message: Overrides the generated message entirely.
        source: Where the payload came from -- an endpoint, path or queue.
            Redacted when it is a URL, left alone when it is a file path.
        preview: A bounded excerpt of the payload, truncated to
            :data:`dataexcept._previews.MAX_PREVIEW_LENGTH` characters.
        cause: The underlying exception, also set as ``__cause__``.
    """

    def __init__(
        self,
        data: bytes | None = None,
        format: str | None = None,
        message: str | None = None,
        *,
        source: str | None = None,
        preview: str | bytes | bytearray | None = None,
        cause: Exception | None = None,
    ) -> None:
        self.data = data
        self.format = format
        self.source = redact_if_url(source)
        self.preview = bounded_preview(preview)
        self.cause = resolve_cause(cause=cause)
        self.message = message or self._default_message()
        super().__init__(self.message)

    def _default_message(self) -> str:
        if self.data is not None and self.format is not None and self.source is None:
            # Exactly the call the historical signature accepted, so exactly
            # the message it produced.
            described = f"Failed to deserialize data from {self.format}"
        else:
            described = (
                f"Failed to deserialize {_describe_target(self.format, self.source)}"
            )
        return f"{described}: {self.cause}" if self.cause else described

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,
        *,
        cause: Exception | None = None,
    ):
        self.operation = operation
        self.timeout = timeout
        self.cause = cause
        msg = f"Operation '{operation}' timed out after {timeout} seconds"
        super().__init__(msg)

ParsingError

Bases: JobError

Raised when parsing of input data fails.

Parameters:

Name	Type	Description	Default
text	str | None	The offending content. Optional, and kept verbatim when given: pass it only when the content is yours and safe to retain.	None
message	str | None	Overrides the generated message entirely.	None
source	str | None	Where the content came from -- an endpoint, path or queue. Redacted when it is a URL, left alone when it is a file path.	None
format	str | None	What the content was meant to be, such as "json".	None
preview	str | bytes | bytearray | None	A bounded excerpt of the content, truncated to :data:dataexcept._previews.MAX_PREVIEW_LENGTH characters.	None
cause	Exception | None	The underlying exception, also set as __cause__.	None

Source code in dataexcept/exceptions/parsing.py

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

    Args:
        text: The offending content. Optional, and kept verbatim when given:
            pass it only when the content is yours and safe to retain.
        message: Overrides the generated message entirely.
        source: Where the content came from -- an endpoint, path or queue.
            Redacted when it is a URL, left alone when it is a file path.
        format: What the content was meant to be, such as ``"json"``.
        preview: A bounded excerpt of the content, truncated to
            :data:`dataexcept._previews.MAX_PREVIEW_LENGTH` characters.
        cause: The underlying exception, also set as ``__cause__``.
    """

    def __init__(
        self,
        text: str | None = None,
        message: str | None = None,
        *,
        source: str | None = None,
        format: str | None = None,
        preview: str | bytes | bytearray | None = None,
        cause: Exception | None = None,
    ) -> None:
        self.text = text
        self.source = redact_if_url(source)
        self.format = format
        self.preview = bounded_preview(preview)
        self.cause = resolve_cause(cause=cause)
        self.message = message or self._default_message()
        super().__init__(self.message)

    def _default_message(self) -> str:
        if self.text is not None and self.source is None and self.format is None:
            # The historical message, with the payload bounded. Truncating the
            # text before repr keeps the quotes balanced.
            described = f"Failed to parse text: {bounded_preview(self.text)!r}"
        else:
            described = f"Failed to parse {_describe_target(self.format, self.source)}"
        return f"{described}: {self.cause}" if self.cause else described

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,
        *,
        cause: Exception | None = None,
    ):
        resolved = resolve_cause(
            cause=cause,
            original_exception=original_exception,
        )
        self.service_name = service_name
        self.original_exception = resolved
        self.cause = resolved
        msg = f"Failed to connect to service '{service_name}'"
        if resolved:
            msg += f": {resolved}"
        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."""

    _default_failure_metadata = FailureMetadata(
        failure_kind="permanent",
        retryable=False,
    )

    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)

FailureMetadata dataclass

Describe recovery-relevant properties of an operational failure.

failure_kind describes whether the underlying condition is known to be transient, permanent for the same operation/payload, or unknown. retryable is deliberately independent: DataExcept describes the failure, while the calling application still owns retry policy.

Source code in dataexcept/failure_metadata.py

@dataclass(frozen=True)
class FailureMetadata:
    """Describe recovery-relevant properties of an operational failure.

    ``failure_kind`` describes whether the underlying condition is known to be
    transient, permanent for the same operation/payload, or unknown.
    ``retryable`` is deliberately independent: DataExcept describes the
    failure, while the calling application still owns retry policy.
    """

    failure_kind: FailureKind = "unknown"
    retryable: bool | None = None
    retry_after_seconds: float | None = None

    def __post_init__(self) -> None:
        if self.failure_kind not in _VALID_FAILURE_KINDS:
            raise ValueError(
                "failure_kind must be 'transient', 'permanent', or 'unknown'"
            )
        if self.retryable is not None and not isinstance(self.retryable, bool):
            raise TypeError("retryable must be bool or None")
        if self.retry_after_seconds is None:
            return
        if isinstance(self.retry_after_seconds, bool) or not isinstance(
            self.retry_after_seconds, (int, float)
        ):
            raise TypeError("retry_after_seconds must be a number or None")
        seconds = float(self.retry_after_seconds)
        if not math.isfinite(seconds) or seconds < 0:
            raise ValueError("retry_after_seconds must be finite and non-negative")
        object.__setattr__(self, "retry_after_seconds", seconds)

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)

OperationContext dataclass

Identifiers that describe where an operation is executing.

system, component and operation are intended to stay low-cardinality and are suitable for filtering. Request, job, correlation, trace and span identifiers are correlation data and should not normally be turned into metric dimensions or indexed tags.

Every value is optional because inventing provenance is worse than omitting it. URL-shaped values are redacted before they can leave this object.

Source code in dataexcept/observability.py

@dataclass(frozen=True, slots=True)
class OperationContext:
    """Identifiers that describe where an operation is executing.

    ``system``, ``component`` and ``operation`` are intended to stay
    low-cardinality and are suitable for filtering.  Request, job, correlation,
    trace and span identifiers are correlation data and should not normally be
    turned into metric dimensions or indexed tags.

    Every value is optional because inventing provenance is worse than omitting
    it.  URL-shaped values are redacted before they can leave this object.
    """

    system: str | None = None
    component: str | None = None
    operation: str | None = None
    request_id: str | None = None
    job_id: str | None = None
    correlation_id: str | None = None
    trace_id: str | None = None
    span_id: str | None = None

    def __post_init__(self) -> None:
        for field in fields(self):
            value = getattr(self, field.name)
            if value is None:
                continue
            if not isinstance(value, str):
                raise TypeError(f"{field.name} must be a string or None")
            if not value.strip():
                raise ValueError(f"{field.name} must not be empty")
            object.__setattr__(
                self,
                field.name,
                redact_urls_in_text(value, keep_path=False),
            )

    def to_dict(self) -> dict[str, str]:
        """Return only the identifiers that are actually present."""
        return {
            field.name: value
            for field in fields(self)
            if (value := getattr(self, field.name)) is not None
        }

    def index_fields(self) -> dict[str, str]:
        """Return only low-cardinality fields suitable for filtering."""
        return {
            key: value
            for key in ("system", "component", "operation")
            if (value := getattr(self, key)) is not None
        }

to_dict

to_dict() -> dict[str, str]

Return only the identifiers that are actually present.

Source code in dataexcept/observability.py

def to_dict(self) -> dict[str, str]:
    """Return only the identifiers that are actually present."""
    return {
        field.name: value
        for field in fields(self)
        if (value := getattr(self, field.name)) is not None
    }

index_fields

index_fields() -> dict[str, str]

Return only low-cardinality fields suitable for filtering.

Source code in dataexcept/observability.py

def index_fields(self) -> dict[str, str]:
    """Return only low-cardinality fields suitable for filtering."""
    return {
        key: value
        for key in ("system", "component", "operation")
        if (value := getattr(self, key)) is not None
    }

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:
        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:
        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."""

    _default_failure_metadata = FailureMetadata(
        failure_kind="permanent",
        retryable=False,
    )

    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."""

    _default_failure_metadata = FailureMetadata(
        failure_kind="permanent",
        retryable=False,
    )

    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,
        *,
        cause: Exception | None = None,
    ) -> None:
        default = f"Storage {operation} failed at location: '{location}'."
        self.location = redact_if_url(location)
        self.operation = operation
        self.cause = resolve_cause(cause=cause)
        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, operation_context: OperationContext | 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,
    operation_context: OperationContext | 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,
            operation_context=operation_context,
        )
        raise

log_exception

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

Log exc at the given log level using logger.

If logger is None a module level logger is used.

context remains the free-form application context. operation_context carries the stable request/job/tool-call identifiers shared with tracing and error-tracker integrations.

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.

Logging is fail-open: context conversion, traceback rendering or the logger itself may fail, but that failure is swallowed so observability can never replace the exception the caller was already handling.

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,
    operation_context: OperationContext | None = None,
) -> None:
    """Log *exc* at the given log *level* using *logger*.

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

    ``context`` remains the free-form application context. ``operation_context``
    carries the stable request/job/tool-call identifiers shared with tracing and
    error-tracker integrations.

    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.

    Logging is fail-open: context conversion, traceback rendering or the logger
    itself may fail, but that failure is swallowed so observability can never
    replace the exception the caller was already handling.
    """
    try:
        if logger is None:
            logger = logging.getLogger(__name__)
        extra = _build_extra(context, operation_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)
    except Exception:
        return

log_then_raise

log_then_raise(exc: Exception, logger: Optional[Logger] = None, level: int = logging.ERROR, context: Context | None = None, operation_context: OperationContext | 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,
    operation_context: OperationContext | 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,
        operation_context=operation_context,
    )
    raise exc

exception_to_observability_event

exception_to_observability_event(exc: BaseException, *, operation_context: OperationContext | None = None, include_attributes: bool = True, max_depth: int = 8) -> dict[str, Any]

Return a strict-JSON-safe failure event shared by observability adapters.

The exception payload is the canonical DataExcept envelope. Operation metadata is separate so adapters can index stable fields while retaining high-cardinality correlation identifiers without flattening them into tags.

Source code in dataexcept/observability.py

def exception_to_observability_event(
    exc: BaseException,
    *,
    operation_context: OperationContext | None = None,
    include_attributes: bool = True,
    max_depth: int = 8,
) -> dict[str, Any]:
    """Return a strict-JSON-safe failure event shared by observability adapters.

    The exception payload is the canonical DataExcept envelope.  Operation
    metadata is separate so adapters can index stable fields while retaining
    high-cardinality correlation identifiers without flattening them into tags.
    """
    if operation_context is not None and not isinstance(
        operation_context, OperationContext
    ):
        raise TypeError("operation_context must be an OperationContext or None")

    event: dict[str, Any] = {
        "event": "exception",
        "exception": exception_to_dict(
            exc,
            include_attributes=include_attributes,
            max_depth=max_depth,
        ),
    }

    if operation_context is not None:
        operation = operation_context.to_dict()
        if operation:
            event["operation"] = operation

    return event

envelope_to_pino

envelope_to_pino(envelope: Mapping[str, Any], *, stack: Optional[str] = None) -> dict[str, Any]

Project envelope into the value Pino logs under its error key.

Pass stack only when a real stack representation is in hand -- the traceback the failure actually carried, or a stack a caller received across a boundary. It is redacted like every other exported string.

Source code in dataexcept/pino.py

def envelope_to_pino(
    envelope: Mapping[str, Any],
    *,
    stack: Optional[str] = None,
) -> dict[str, Any]:
    """Project *envelope* into the value Pino logs under its error key.

    Pass ``stack`` only when a real stack representation is in hand -- the
    traceback the failure actually carried, or a stack a caller received across
    a boundary. It is redacted like every other exported string.
    """
    if not isinstance(envelope, Mapping):
        raise TypeError("envelope must be a mapping")
    if stack is not None and not isinstance(stack, str):
        raise TypeError("stack must be a string or None")

    record = _project(envelope, depth=0, seen=set())
    if stack is None:
        return record
    return _with_stack(record, redact_urls_in_text(stack, keep_path=False))

exception_to_pino

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

Return exc in Pino's shape, by way of its envelope.

include_stack renders the exception's own traceback when it has one. The field is omitted rather than invented when it does not.

Source code in dataexcept/pino.py

def exception_to_pino(
    exc: BaseException,
    *,
    include_attributes: bool = True,
    max_depth: int = 8,
    include_stack: bool = False,
) -> dict[str, Any]:
    """Return *exc* in Pino's shape, by way of its envelope.

    ``include_stack`` renders the exception's own traceback when it has one.
    The field is omitted rather than invented when it does not.
    """
    envelope = exception_to_dict(
        exc,
        include_attributes=include_attributes,
        max_depth=max_depth,
    )
    return envelope_to_pino(
        envelope,
        stack=_rendered_stack(exc) if include_stack else None,
    )

envelope_schema

envelope_schema() -> dict[str, Any]

Return the JSON Schema describing an exception envelope.

The result is a fresh copy on every call, so a caller that registers it with a validator, or annotates it for an API specification, cannot corrupt the copy the next caller gets.

Source code in dataexcept/schema.py

def envelope_schema() -> dict[str, Any]:
    """Return the JSON Schema describing an exception envelope.

    The result is a fresh copy on every call, so a caller that registers it
    with a validator, or annotates it for an API specification, cannot corrupt
    the copy the next caller gets.
    """
    return deepcopy(_loaded_schema(_ENVELOPE_FILENAME))

pino_profile_schema

pino_profile_schema() -> dict[str, Any]

Return the JSON Schema describing the Pino projection of an envelope.

A fresh copy on every call, for the same reason as :func:envelope_schema.

Source code in dataexcept/schema.py

def pino_profile_schema() -> dict[str, Any]:
    """Return the JSON Schema describing the Pino projection of an envelope.

    A fresh copy on every call, for the same reason as
    :func:`envelope_schema`.
    """
    return deepcopy(_loaded_schema(_PINO_FILENAME))

enrich_sentry_event

enrich_sentry_event(event: dict[str, Any], hint: Mapping[str, Any], *, operation_context: OperationContext | None = None, include_attributes: bool = True, max_depth: int = 8) -> dict[str, Any]

Return a Sentry event enriched with DataExcept failure metadata.

Full operation context, including correlation identifiers, is stored under contexts.dataexcept_operation. Only the low-cardinality operation fields become tags so request/job/trace identifiers do not become indexed tags.

Source code in dataexcept/sentry.py

def enrich_sentry_event(
    event: dict[str, Any],
    hint: Mapping[str, Any],
    *,
    operation_context: OperationContext | None = None,
    include_attributes: bool = True,
    max_depth: int = 8,
) -> dict[str, Any]:
    """Return a Sentry event enriched with DataExcept failure metadata.

    Full operation context, including correlation identifiers, is stored under
    ``contexts.dataexcept_operation``. Only the low-cardinality operation fields
    become tags so request/job/trace identifiers do not become indexed tags.
    """
    if not isinstance(event, dict):
        raise TypeError("event must be a dictionary")
    if not isinstance(hint, Mapping):
        raise TypeError("hint must be a mapping")
    if operation_context is not None and not isinstance(
        operation_context, OperationContext
    ):
        raise TypeError("operation_context must be an OperationContext or None")

    exc = _exception_from_hint(hint)
    if exc is None:
        return event

    envelope = exception_to_dict(
        exc,
        include_attributes=include_attributes,
        max_depth=max_depth,
    )
    enriched = dict(event)

    existing_contexts = event.get("contexts")
    contexts = dict(existing_contexts) if isinstance(existing_contexts, Mapping) else {}
    contexts["dataexcept"] = envelope
    if operation_context is not None:
        serialized = operation_context.to_dict()
        if serialized:
            contexts["dataexcept_operation"] = serialized
    enriched["contexts"] = contexts

    existing_tags = event.get("tags")
    tags = dict(existing_tags) if isinstance(existing_tags, Mapping) else {}
    tags.update(_tags_from_envelope(envelope))
    tags.update(_operation_tags(operation_context))
    enriched["tags"] = tags

    return enriched

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.

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*."""
    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], /, *, failure_metadata: FailureMetadata | None = None, **kwargs: Any) -> DataExceptError

Build target from original, recording it as the cause.

Extra keyword arguments are passed through to the target constructor. If the target accepts a cause parameter, original is injected unless the caller already supplied cause, original or original_exception. The resulting exception is always chained to original via __cause__.

failure_metadata optionally overrides the target class's conservative default when the integration has backend-specific evidence about whether the failure is transient or retryable.

Source code in dataexcept/wrapping.py

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

    Extra keyword arguments are passed through to the target constructor. If
    the target accepts a cause parameter, *original* is injected unless the
    caller already supplied ``cause``, ``original`` or ``original_exception``.
    The resulting exception is always chained to *original* via ``__cause__``.

    ``failure_metadata`` optionally overrides the target class's conservative
    default when the integration has backend-specific evidence about whether
    the failure is transient or retryable.
    """
    if failure_metadata is not None and not isinstance(
        failure_metadata, FailureMetadata
    ):
        raise TypeError("failure_metadata must be FailureMetadata or None")

    if _explicit_cause_parameter(kwargs) is None:
        parameter = _cause_parameter(target)
        if parameter is not None:
            kwargs[parameter] = original

    exception = target(**kwargs)
    exception.__cause__ = original
    if failure_metadata is not None:
        exception.with_failure_metadata(failure_metadata)
    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."""

    _default_failure_metadata = FailureMetadata(
        failure_kind="permanent",
        retryable=False,
    )

    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."""

    _default_failure_metadata = FailureMetadata(
        failure_kind="permanent",
        retryable=False,
    )

    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,
        *,
        cause: Exception | None = None,
    ):
        self.operation = operation
        self.timeout = timeout
        self.cause = cause
        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,
        *,
        cause: Exception | None = None,
    ):
        resolved = resolve_cause(
            cause=cause,
            original_exception=original_exception,
        )
        self.service_name = service_name
        self.original_exception = resolved
        self.cause = resolved
        msg = f"Failed to connect to service '{service_name}'"
        if resolved:
            msg += f": {resolved}"
        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.

Parameters:

Name	Type	Description	Default
data	bytes | None	The offending bytes. Optional, and kept verbatim when given: pass it only when the payload is yours and safe to retain.	None
format	str | None	What the payload was meant to be, such as "json".	None
message	str | None	Overrides the generated message entirely.	None
source	str | None	Where the payload came from -- an endpoint, path or queue. Redacted when it is a URL, left alone when it is a file path.	None
preview	str | bytes | bytearray | None	A bounded excerpt of the payload, truncated to :data:dataexcept._previews.MAX_PREVIEW_LENGTH characters.	None
cause	Exception | None	The underlying exception, also set as __cause__.	None

Source code in dataexcept/exceptions/parsing.py

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

    Args:
        data: The offending bytes. Optional, and kept verbatim when given:
            pass it only when the payload is yours and safe to retain.
        format: What the payload was meant to be, such as ``"json"``.
        message: Overrides the generated message entirely.
        source: Where the payload came from -- an endpoint, path or queue.
            Redacted when it is a URL, left alone when it is a file path.
        preview: A bounded excerpt of the payload, truncated to
            :data:`dataexcept._previews.MAX_PREVIEW_LENGTH` characters.
        cause: The underlying exception, also set as ``__cause__``.
    """

    def __init__(
        self,
        data: bytes | None = None,
        format: str | None = None,
        message: str | None = None,
        *,
        source: str | None = None,
        preview: str | bytes | bytearray | None = None,
        cause: Exception | None = None,
    ) -> None:
        self.data = data
        self.format = format
        self.source = redact_if_url(source)
        self.preview = bounded_preview(preview)
        self.cause = resolve_cause(cause=cause)
        self.message = message or self._default_message()
        super().__init__(self.message)

    def _default_message(self) -> str:
        if self.data is not None and self.format is not None and self.source is None:
            # Exactly the call the historical signature accepted, so exactly
            # the message it produced.
            described = f"Failed to deserialize data from {self.format}"
        else:
            described = (
                f"Failed to deserialize {_describe_target(self.format, self.source)}"
            )
        return f"{described}: {self.cause}" if self.cause else described

ParsingError

Bases: JobError

Raised when parsing of input data fails.

Parameters:

Name	Type	Description	Default
text	str | None	The offending content. Optional, and kept verbatim when given: pass it only when the content is yours and safe to retain.	None
message	str | None	Overrides the generated message entirely.	None
source	str | None	Where the content came from -- an endpoint, path or queue. Redacted when it is a URL, left alone when it is a file path.	None
format	str | None	What the content was meant to be, such as "json".	None
preview	str | bytes | bytearray | None	A bounded excerpt of the content, truncated to :data:dataexcept._previews.MAX_PREVIEW_LENGTH characters.	None
cause	Exception | None	The underlying exception, also set as __cause__.	None

Source code in dataexcept/exceptions/parsing.py

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

    Args:
        text: The offending content. Optional, and kept verbatim when given:
            pass it only when the content is yours and safe to retain.
        message: Overrides the generated message entirely.
        source: Where the content came from -- an endpoint, path or queue.
            Redacted when it is a URL, left alone when it is a file path.
        format: What the content was meant to be, such as ``"json"``.
        preview: A bounded excerpt of the content, truncated to
            :data:`dataexcept._previews.MAX_PREVIEW_LENGTH` characters.
        cause: The underlying exception, also set as ``__cause__``.
    """

    def __init__(
        self,
        text: str | None = None,
        message: str | None = None,
        *,
        source: str | None = None,
        format: str | None = None,
        preview: str | bytes | bytearray | None = None,
        cause: Exception | None = None,
    ) -> None:
        self.text = text
        self.source = redact_if_url(source)
        self.format = format
        self.preview = bounded_preview(preview)
        self.cause = resolve_cause(cause=cause)
        self.message = message or self._default_message()
        super().__init__(self.message)

    def _default_message(self) -> str:
        if self.text is not None and self.source is None and self.format is None:
            # The historical message, with the payload bounded. Truncating the
            # text before repr keeps the quotes balanced.
            described = f"Failed to parse text: {bounded_preview(self.text)!r}"
        else:
            described = f"Failed to parse {_describe_target(self.format, self.source)}"
        return f"{described}: {self.cause}" if self.cause else described

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."""

    _default_failure_metadata = FailureMetadata(
        failure_kind="permanent",
        retryable=False,
    )

    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:
        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,
        *,
        cause: Exception | None = None,
    ) -> None:
        default = f"Storage {operation} failed at location: '{location}'."
        self.location = redact_if_url(location)
        self.operation = operation
        self.cause = resolve_cause(cause=cause)
        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."""

    _default_failure_metadata = FailureMetadata(
        failure_kind="permanent",
        retryable=False,
    )

    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."""

    _default_failure_metadata = FailureMetadata(
        failure_kind="permanent",
        retryable=False,
    )

    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:
        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,
        *,
        cause: Exception | None = None,
    ) -> None:
        """Initialize QueryExecutionError.

        Args:
            query: SQL query string.
            original: Legacy alias for ``cause``.
            cause: Optional underlying exception.
        """
        resolved = resolve_cause(cause=cause, original=original)
        self.query = query
        self.original = resolved
        self.cause = resolved
        msg = f"Query failed: {query}"
        if resolved:
            msg += f" ({resolved})"
        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,
        *,
        cause: Exception | None = None,
    ) -> None:
        """Initialize TransactionError.

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

Message broker exceptions

broker_exceptions

Custom exceptions for message-broker operations.

A broker is a data-engineering boundary like a database or an object store, and it fails in ways that are specific to it: a publish is rejected, a consumer cannot reach its group, an offset is never committed. Mapping those onto ServiceConnectionError and OperationTimeoutError loses which of them happened, and with it the topic, partition, offset and consumer group that say where.

The hierarchy is deliberately about the operation rather than the product. Kafka, RabbitMQ, Pulsar and NATS disagree about almost everything else, but all four connect, publish, consume and acknowledge, so a pipeline can catch these without knowing which broker or client library is underneath -- and DataExcept depends on none of them.

Failure metadata stays at the conservative default. A broker refusing a publish may be a leader election that resolves in a second or a topic that does not exist, and the exception cannot tell which. An integration that does know -- because it read the broker's own error code -- attaches that with with_failure_metadata or wrap(..., failure_metadata=...).

MessageBrokerError

Bases: DataExceptError

Base exception for message-broker failures.

Catching this catches every broker failure the library raises, without catching a database or HTTP one.

Source code in dataexcept/broker_exceptions.py

class MessageBrokerError(DataExceptError):
    """Base exception for message-broker failures.

    Catching this catches every broker failure the library raises, without
    catching a database or HTTP one.
    """

    pass

BrokerConnectionError

Bases: MessageBrokerError

Raised when a connection to the broker cannot be established.

Source code in dataexcept/broker_exceptions.py

class BrokerConnectionError(MessageBrokerError):
    """Raised when a connection to the broker cannot be established."""

    def __init__(
        self,
        broker: str,
        message: Optional[str] = None,
        *,
        cause: Optional[Exception] = None,
    ) -> None:
        """Initialize BrokerConnectionError.

        Args:
            broker: The broker being connected to. A bootstrap address, a
                cluster name, or a URL -- redacted when it is a URL, since one
                routinely carries credentials.
            message: Optional custom error message.
            cause: Optional underlying exception, also set as ``__cause__``.
        """
        self.broker = redact_if_url(broker)
        self.cause = resolve_cause(cause=cause)
        default = f"Failed to connect to message broker '{self.broker}'"
        if self.cause:
            default += f": {self.cause}"
        super().__init__(message or default)

BrokerTimeoutError

Bases: MessageBrokerError

Raised when a broker operation exceeds its time limit.

Source code in dataexcept/broker_exceptions.py

class BrokerTimeoutError(MessageBrokerError):
    """Raised when a broker operation exceeds its time limit."""

    def __init__(
        self,
        broker: str,
        operation: Optional[str] = None,
        timeout_seconds: Optional[float] = None,
        message: Optional[str] = None,
        *,
        cause: Optional[Exception] = None,
    ) -> None:
        """Initialize BrokerTimeoutError.

        Args:
            broker: The broker the operation was waiting on.
            operation: What was being attempted, such as ``"publish"``.
            timeout_seconds: The limit that was exceeded.
            message: Optional custom error message.
            cause: Optional underlying exception, also set as ``__cause__``.
        """
        self.broker = redact_if_url(broker)
        self.operation = operation
        self.timeout_seconds = timeout_seconds
        self.cause = resolve_cause(cause=cause)
        attempted = (
            f"Broker operation '{operation}'" if operation else "Broker operation"
        )
        default = f"{attempted} timed out"
        if timeout_seconds is not None:
            default += f" after {timeout_seconds}s"
        default += _on_broker(self.broker)
        if self.cause:
            default += f": {self.cause}"
        super().__init__(message or default)

MessagePublishError

Bases: MessageBrokerError

Raised when publishing a message fails.

Source code in dataexcept/broker_exceptions.py

class MessagePublishError(MessageBrokerError):
    """Raised when publishing a message fails."""

    def __init__(
        self,
        topic: str,
        broker: Optional[str] = None,
        partition: Optional[int] = None,
        message: Optional[str] = None,
        *,
        cause: Optional[Exception] = None,
    ) -> None:
        """Initialize MessagePublishError.

        Args:
            topic: The topic, queue or subject published to.
            broker: Optional broker the publish was addressed to.
            partition: Optional partition the message was keyed to.
            message: Optional custom error message.
            cause: Optional underlying exception, also set as ``__cause__``.
        """
        self.topic = topic
        self.broker = redact_if_url(broker)
        self.partition = partition
        self.cause = resolve_cause(cause=cause)
        position = _describe_position(topic, partition)
        default = f"Failed to publish to {position}{_on_broker(self.broker)}"
        if self.cause:
            default += f": {self.cause}"
        super().__init__(message or default)

MessageConsumeError

Bases: MessageBrokerError

Raised when consuming a message fails.

Source code in dataexcept/broker_exceptions.py

class MessageConsumeError(MessageBrokerError):
    """Raised when consuming a message fails."""

    def __init__(
        self,
        topic: str,
        broker: Optional[str] = None,
        partition: Optional[int] = None,
        offset: Optional[int] = None,
        consumer_group: Optional[str] = None,
        message: Optional[str] = None,
        *,
        cause: Optional[Exception] = None,
    ) -> None:
        """Initialize MessageConsumeError.

        Args:
            topic: The topic, queue or subject consumed from.
            broker: Optional broker the consumer was reading from.
            partition: Optional partition being read.
            offset: Optional offset the failure happened at.
            consumer_group: Optional consumer group the reader belongs to.
            message: Optional custom error message.
            cause: Optional underlying exception, also set as ``__cause__``.
        """
        self.topic = topic
        self.broker = redact_if_url(broker)
        self.partition = partition
        self.offset = offset
        self.consumer_group = consumer_group
        self.cause = resolve_cause(cause=cause)
        position = _describe_position(topic, partition, offset, consumer_group)
        default = f"Failed to consume from {position}{_on_broker(self.broker)}"
        if self.cause:
            default += f": {self.cause}"
        super().__init__(message or default)

MessageAcknowledgementError

Bases: MessageBrokerError

Raised when acknowledging or committing a message fails.

Distinct from a consume failure on purpose: the message was read and processed, and it is the record of that which did not stick -- so it will be delivered again.

Source code in dataexcept/broker_exceptions.py

class MessageAcknowledgementError(MessageBrokerError):
    """Raised when acknowledging or committing a message fails.

    Distinct from a consume failure on purpose: the message was read and
    processed, and it is the record of that which did not stick -- so it will
    be delivered again.
    """

    def __init__(
        self,
        topic: str,
        broker: Optional[str] = None,
        partition: Optional[int] = None,
        offset: Optional[int] = None,
        consumer_group: Optional[str] = None,
        message: Optional[str] = None,
        *,
        cause: Optional[Exception] = None,
    ) -> None:
        """Initialize MessageAcknowledgementError.

        Args:
            topic: The topic, queue or subject the message came from.
            broker: Optional broker the acknowledgement was sent to.
            partition: Optional partition the message was read from.
            offset: Optional offset that failed to commit.
            consumer_group: Optional consumer group committing the offset.
            message: Optional custom error message.
            cause: Optional underlying exception, also set as ``__cause__``.
        """
        self.topic = topic
        self.broker = redact_if_url(broker)
        self.partition = partition
        self.offset = offset
        self.consumer_group = consumer_group
        self.cause = resolve_cause(cause=cause)
        position = _describe_position(topic, partition, offset, consumer_group)
        default = f"Failed to acknowledge {position}{_on_broker(self.broker)}"
        if self.cause:
            default += f": {self.cause}"
        super().__init__(message or default)

Message broker observability context

broker_context

Framework-neutral broker and stream-processing observability context.

The adapter accepts plain message metadata so Kafka, RabbitMQ, Pulsar, NATS, stream processors and custom brokers can share one dependency-free boundary model. Message bodies and application payloads are intentionally excluded.

BrokerContext

Broker operation context plus optional message coordinates.

Source code in dataexcept/broker_context.py

class BrokerContext:
    """Broker operation context plus optional message coordinates."""

    __slots__ = (
        "consumer_group",
        "message_id",
        "offset",
        "operation_context",
        "partition",
        "topic",
        "trace_context",
    )

    def __init__(
        self,
        operation_context: OperationContext,
        trace_context: W3CTraceContext | None,
        *,
        topic: str,
        partition: int | None,
        offset: int | None,
        consumer_group: str | None,
        message_id: str | None,
    ) -> None:
        self.operation_context = operation_context
        self.trace_context = trace_context
        self.topic = topic
        self.partition = partition
        self.offset = offset
        self.consumer_group = consumer_group
        self.message_id = message_id

broker_context_from_message

broker_context_from_message(operation: str, topic: str, *, metadata: Mapping[str, object] | None = None, system: str | None = 'broker', component: str | None = None, correlation_id: str | None = None, partition: int | None = None, offset: int | None = None, consumer_group: str | None = None, message_id: str | None = None) -> BrokerContext

Build observability context for a broker or stream-processing boundary.

The stable operation identity is " ", where operation is one of publish, consume or acknowledge. Partition, offset, consumer-group and message identifiers remain wrapper metadata instead of becoming low-cardinality operation labels. Incoming W3C Trace Context is preserved when present.

Source code in dataexcept/broker_context.py

def broker_context_from_message(
    operation: str,
    topic: str,
    *,
    metadata: Mapping[str, object] | None = None,
    system: str | None = "broker",
    component: str | None = None,
    correlation_id: str | None = None,
    partition: int | None = None,
    offset: int | None = None,
    consumer_group: str | None = None,
    message_id: str | None = None,
) -> BrokerContext:
    """Build observability context for a broker or stream-processing boundary.

    The stable operation identity is "<operation> <topic>", where operation is
    one of publish, consume or acknowledge. Partition, offset, consumer-group
    and message identifiers remain wrapper metadata instead of becoming
    low-cardinality operation labels. Incoming W3C Trace Context is preserved
    when present.
    """
    if not isinstance(operation, str):
        raise TypeError("operation must be a string")
    normalized_operation = operation.strip().lower()
    if normalized_operation not in _VALID_OPERATIONS:
        raise ValueError("operation must be one of publish, consume or acknowledge")

    topic_name = _single_line(topic, "topic")
    correlation_identifier = _optional_single_line(
        correlation_id,
        "correlation_id",
    )
    group = _optional_single_line(consumer_group, "consumer_group")
    message = _optional_single_line(message_id, "message_id")
    partition_number = _optional_non_negative_int(partition, "partition")
    offset_number = _optional_non_negative_int(offset, "offset")

    if metadata is None:
        metadata = {}
    if not isinstance(metadata, Mapping):
        raise TypeError("metadata must be a mapping or None")

    operation_context = OperationContext(
        system=system,
        component=component,
        operation=f"{normalized_operation} {topic_name}",
        correlation_id=correlation_identifier,
    )

    trace_context = trace_context_from_mapping(metadata)
    if trace_context is not None:
        operation_context = trace_context.to_operation_context(operation_context)

    return BrokerContext(
        operation_context,
        trace_context,
        topic=topic_name,
        partition=partition_number,
        offset=offset_number,
        consumer_group=group,
        message_id=message,
    )

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))

Observability model

observability

Product-neutral observability context for failures crossing boundaries.

Frameworks call the same concepts by different names: request, task, job, workflow step, tool call, invocation. DataExcept keeps the shared part small and explicit so adapters can project it into logs, traces, error trackers or protocol-specific metadata without making those frameworks runtime requirements.

OperationContext dataclass

Identifiers that describe where an operation is executing.

system, component and operation are intended to stay low-cardinality and are suitable for filtering. Request, job, correlation, trace and span identifiers are correlation data and should not normally be turned into metric dimensions or indexed tags.

Every value is optional because inventing provenance is worse than omitting it. URL-shaped values are redacted before they can leave this object.

Source code in dataexcept/observability.py

@dataclass(frozen=True, slots=True)
class OperationContext:
    """Identifiers that describe where an operation is executing.

    ``system``, ``component`` and ``operation`` are intended to stay
    low-cardinality and are suitable for filtering.  Request, job, correlation,
    trace and span identifiers are correlation data and should not normally be
    turned into metric dimensions or indexed tags.

    Every value is optional because inventing provenance is worse than omitting
    it.  URL-shaped values are redacted before they can leave this object.
    """

    system: str | None = None
    component: str | None = None
    operation: str | None = None
    request_id: str | None = None
    job_id: str | None = None
    correlation_id: str | None = None
    trace_id: str | None = None
    span_id: str | None = None

    def __post_init__(self) -> None:
        for field in fields(self):
            value = getattr(self, field.name)
            if value is None:
                continue
            if not isinstance(value, str):
                raise TypeError(f"{field.name} must be a string or None")
            if not value.strip():
                raise ValueError(f"{field.name} must not be empty")
            object.__setattr__(
                self,
                field.name,
                redact_urls_in_text(value, keep_path=False),
            )

    def to_dict(self) -> dict[str, str]:
        """Return only the identifiers that are actually present."""
        return {
            field.name: value
            for field in fields(self)
            if (value := getattr(self, field.name)) is not None
        }

    def index_fields(self) -> dict[str, str]:
        """Return only low-cardinality fields suitable for filtering."""
        return {
            key: value
            for key in ("system", "component", "operation")
            if (value := getattr(self, key)) is not None
        }

to_dict

to_dict() -> dict[str, str]

Return only the identifiers that are actually present.

Source code in dataexcept/observability.py

def to_dict(self) -> dict[str, str]:
    """Return only the identifiers that are actually present."""
    return {
        field.name: value
        for field in fields(self)
        if (value := getattr(self, field.name)) is not None
    }

index_fields

index_fields() -> dict[str, str]

Return only low-cardinality fields suitable for filtering.

Source code in dataexcept/observability.py

def index_fields(self) -> dict[str, str]:
    """Return only low-cardinality fields suitable for filtering."""
    return {
        key: value
        for key in ("system", "component", "operation")
        if (value := getattr(self, key)) is not None
    }

exception_to_observability_event

exception_to_observability_event(exc: BaseException, *, operation_context: OperationContext | None = None, include_attributes: bool = True, max_depth: int = 8) -> dict[str, Any]

Return a strict-JSON-safe failure event shared by observability adapters.

The exception payload is the canonical DataExcept envelope. Operation metadata is separate so adapters can index stable fields while retaining high-cardinality correlation identifiers without flattening them into tags.

Source code in dataexcept/observability.py

def exception_to_observability_event(
    exc: BaseException,
    *,
    operation_context: OperationContext | None = None,
    include_attributes: bool = True,
    max_depth: int = 8,
) -> dict[str, Any]:
    """Return a strict-JSON-safe failure event shared by observability adapters.

    The exception payload is the canonical DataExcept envelope.  Operation
    metadata is separate so adapters can index stable fields while retaining
    high-cardinality correlation identifiers without flattening them into tags.
    """
    if operation_context is not None and not isinstance(
        operation_context, OperationContext
    ):
        raise TypeError("operation_context must be an OperationContext or None")

    event: dict[str, Any] = {
        "event": "exception",
        "exception": exception_to_dict(
            exc,
            include_attributes=include_attributes,
            max_depth=max_depth,
        ),
    }

    if operation_context is not None:
        operation = operation_context.to_dict()
        if operation:
            event["operation"] = operation

    return event

OpenTelemetry adapter

opentelemetry

OpenTelemetry-compatible exception attributes without an OTel dependency.

OpenTelemetry defines stable exception.type, exception.message and exception.stacktrace attributes for exceptions. DataExcept can supply those values from its redacted envelope, plus its recovery metadata and product-neutral operation context, without importing opentelemetry itself.

ExceptionRecorder

Bases: Protocol

Minimal structural interface required by :func:record_otel_exception.

Source code in dataexcept/opentelemetry.py

class ExceptionRecorder(Protocol):
    """Minimal structural interface required by :func:`record_otel_exception`."""

    def record_exception(
        self,
        exception: BaseException,
        attributes: Mapping[str, OtelAttributeValue] | None = None,
    ) -> None:
        """Record *exception* with optional event attributes."""

record_exception

record_exception(exception: BaseException, attributes: Mapping[str, OtelAttributeValue] | None = None) -> None

Record exception with optional event attributes.

Source code in dataexcept/opentelemetry.py

def record_exception(
    self,
    exception: BaseException,
    attributes: Mapping[str, OtelAttributeValue] | None = None,
) -> None:
    """Record *exception* with optional event attributes."""

exception_to_otel_attributes

exception_to_otel_attributes(exc: BaseException, *, operation_context: OperationContext | None = None, include_attributes: bool = True, max_depth: int = 8, include_stacktrace: bool = True, include_envelope: bool = False) -> dict[str, OtelAttributeValue]

Return OpenTelemetry-compatible attributes describing exc.

trace_id and span_id from :class:OperationContext are deliberately not duplicated as custom attributes. OpenTelemetry already carries them in native span context; DataExcept only projects operation and correlation data.

Source code in dataexcept/opentelemetry.py

def exception_to_otel_attributes(
    exc: BaseException,
    *,
    operation_context: OperationContext | None = None,
    include_attributes: bool = True,
    max_depth: int = 8,
    include_stacktrace: bool = True,
    include_envelope: bool = False,
) -> dict[str, OtelAttributeValue]:
    """Return OpenTelemetry-compatible attributes describing *exc*.

    ``trace_id`` and ``span_id`` from :class:`OperationContext` are deliberately
    not duplicated as custom attributes. OpenTelemetry already carries them in
    native span context; DataExcept only projects operation and correlation data.
    """
    envelope = exception_to_dict(
        exc,
        include_attributes=include_attributes,
        max_depth=max_depth,
    )

    attributes: dict[str, OtelAttributeValue] = {
        "exception.type": _qualified_type_name(exc),
        "exception.message": str(envelope.get("message", "")),
    }

    if include_stacktrace:
        stacktrace = _rendered_stacktrace(exc)
        if stacktrace is not None:
            attributes["exception.stacktrace"] = stacktrace

    attributes.update(_failure_attributes(envelope))
    attributes.update(_operation_attributes(operation_context))

    if include_envelope:
        attributes["dataexcept.envelope.schema"] = ENVELOPE_SCHEMA_ID
        attributes["dataexcept.envelope"] = json.dumps(
            envelope,
            ensure_ascii=False,
            allow_nan=False,
            separators=(",", ":"),
            sort_keys=True,
        )

    return attributes

record_otel_exception

record_otel_exception(span: ExceptionRecorder, exc: BaseException, *, operation_context: OperationContext | None = None, include_attributes: bool = True, max_depth: int = 8, include_stacktrace: bool = True, include_envelope: bool = False) -> None

Record exc without allowing telemetry failure to escape.

Attribute conversion remains strict through :func:exception_to_otel_attributes. This emission helper is different: it is intended for use while handling an existing failure, so conversion or recorder errors are swallowed rather than replacing that failure.

Source code in dataexcept/opentelemetry.py

def record_otel_exception(
    span: ExceptionRecorder,
    exc: BaseException,
    *,
    operation_context: OperationContext | None = None,
    include_attributes: bool = True,
    max_depth: int = 8,
    include_stacktrace: bool = True,
    include_envelope: bool = False,
) -> None:
    """Record *exc* without allowing telemetry failure to escape.

    Attribute conversion remains strict through
    :func:`exception_to_otel_attributes`. This emission helper is different:
    it is intended for use while handling an existing failure, so conversion or
    recorder errors are swallowed rather than replacing that failure.
    """
    try:
        attributes = exception_to_otel_attributes(
            exc,
            operation_context=operation_context,
            include_attributes=include_attributes,
            max_depth=max_depth,
            include_stacktrace=include_stacktrace,
            include_envelope=include_envelope,
        )
        span.record_exception(exc, attributes=attributes)
    except Exception:
        return

Sentry adapter

sentry

Enrich Sentry error events with structured DataExcept metadata.

enrich_sentry_event

enrich_sentry_event(event: dict[str, Any], hint: Mapping[str, Any], *, operation_context: OperationContext | None = None, include_attributes: bool = True, max_depth: int = 8) -> dict[str, Any]

Return a Sentry event enriched with DataExcept failure metadata.

Full operation context, including correlation identifiers, is stored under contexts.dataexcept_operation. Only the low-cardinality operation fields become tags so request/job/trace identifiers do not become indexed tags.

Source code in dataexcept/sentry.py

def enrich_sentry_event(
    event: dict[str, Any],
    hint: Mapping[str, Any],
    *,
    operation_context: OperationContext | None = None,
    include_attributes: bool = True,
    max_depth: int = 8,
) -> dict[str, Any]:
    """Return a Sentry event enriched with DataExcept failure metadata.

    Full operation context, including correlation identifiers, is stored under
    ``contexts.dataexcept_operation``. Only the low-cardinality operation fields
    become tags so request/job/trace identifiers do not become indexed tags.
    """
    if not isinstance(event, dict):
        raise TypeError("event must be a dictionary")
    if not isinstance(hint, Mapping):
        raise TypeError("hint must be a mapping")
    if operation_context is not None and not isinstance(
        operation_context, OperationContext
    ):
        raise TypeError("operation_context must be an OperationContext or None")

    exc = _exception_from_hint(hint)
    if exc is None:
        return event

    envelope = exception_to_dict(
        exc,
        include_attributes=include_attributes,
        max_depth=max_depth,
    )
    enriched = dict(event)

    existing_contexts = event.get("contexts")
    contexts = dict(existing_contexts) if isinstance(existing_contexts, Mapping) else {}
    contexts["dataexcept"] = envelope
    if operation_context is not None:
        serialized = operation_context.to_dict()
        if serialized:
            contexts["dataexcept_operation"] = serialized
    enriched["contexts"] = contexts

    existing_tags = event.get("tags")
    tags = dict(existing_tags) if isinstance(existing_tags, Mapping) else {}
    tags.update(_tags_from_envelope(envelope))
    tags.update(_operation_tags(operation_context))
    enriched["tags"] = tags

    return enriched

W3C Trace Context

trace_context

W3C Trace Context parsing without a tracing runtime dependency.

DataExcept only needs enough trace context to keep failures correlated across execution boundaries. It does not start spans or generate identifiers. The raw carrier values are preserved for forwarding, while parsed identifiers are available to logging, error-tracker and protocol adapters.

W3CTraceContext dataclass

Parsed W3C traceparent plus optional propagation companions.

parent_id is the caller's span identifier from the incoming carrier. It is deliberately not treated as the local OperationContext.span_id. traceparent is retained verbatim so pass-through code can forward an unknown future version without reconstructing fields it does not understand.

Source code in dataexcept/trace_context.py

@dataclass(frozen=True, slots=True)
class W3CTraceContext:
    """Parsed W3C ``traceparent`` plus optional propagation companions.

    ``parent_id`` is the caller's span identifier from the incoming carrier.
    It is deliberately not treated as the local ``OperationContext.span_id``.
    ``traceparent`` is retained verbatim so pass-through code can forward an
    unknown future version without reconstructing fields it does not understand.
    """

    traceparent: str
    version: str
    trace_id: str
    parent_id: str
    trace_flags: str
    tracestate: str | None = None
    baggage: str | None = None

    @property
    def sampled(self) -> bool:
        """Whether the W3C sampled bit is set in ``trace_flags``."""
        return bool(int(self.trace_flags, 16) & 0x01)

    def to_carrier(self) -> dict[str, str]:
        """Return propagation fields without changing their received values."""
        carrier = {"traceparent": self.traceparent}
        if self.tracestate is not None:
            carrier["tracestate"] = self.tracestate
        if self.baggage is not None:
            carrier["baggage"] = self.baggage
        return carrier

    def to_operation_context(
        self,
        context: OperationContext | None = None,
    ) -> OperationContext:
        """Return *context* correlated with this trace without inventing a span.

        Existing operation metadata is preserved.  A conflicting trace ID is
        rejected because silently replacing provenance would make correlation
        less trustworthy than omitting it.
        """
        if context is None:
            return OperationContext(trace_id=self.trace_id)
        if not isinstance(context, OperationContext):
            raise TypeError("context must be an OperationContext or None")
        if context.trace_id is not None and context.trace_id != self.trace_id:
            raise ValueError("context trace_id conflicts with traceparent")
        if context.trace_id == self.trace_id:
            return context
        return replace(context, trace_id=self.trace_id)

sampled property

sampled: bool

Whether the W3C sampled bit is set in trace_flags.

to_carrier

to_carrier() -> dict[str, str]

Return propagation fields without changing their received values.

Source code in dataexcept/trace_context.py

def to_carrier(self) -> dict[str, str]:
    """Return propagation fields without changing their received values."""
    carrier = {"traceparent": self.traceparent}
    if self.tracestate is not None:
        carrier["tracestate"] = self.tracestate
    if self.baggage is not None:
        carrier["baggage"] = self.baggage
    return carrier

to_operation_context

to_operation_context(context: OperationContext | None = None) -> OperationContext

Return context correlated with this trace without inventing a span.

Existing operation metadata is preserved. A conflicting trace ID is rejected because silently replacing provenance would make correlation less trustworthy than omitting it.

Source code in dataexcept/trace_context.py

def to_operation_context(
    self,
    context: OperationContext | None = None,
) -> OperationContext:
    """Return *context* correlated with this trace without inventing a span.

    Existing operation metadata is preserved.  A conflicting trace ID is
    rejected because silently replacing provenance would make correlation
    less trustworthy than omitting it.
    """
    if context is None:
        return OperationContext(trace_id=self.trace_id)
    if not isinstance(context, OperationContext):
        raise TypeError("context must be an OperationContext or None")
    if context.trace_id is not None and context.trace_id != self.trace_id:
        raise ValueError("context trace_id conflicts with traceparent")
    if context.trace_id == self.trace_id:
        return context
    return replace(context, trace_id=self.trace_id)

parse_traceparent

parse_traceparent(value: str, *, tracestate: str | None = None, baggage: str | None = None) -> W3CTraceContext | None

Parse a W3C traceparent value, returning None when invalid.

Version 00 uses the exact 55-character format. Higher versions are accepted when their mandatory prefix is parseable; any extension remains opaque and is preserved in traceparent for forwarding. No identifiers are generated when parsing fails.

Source code in dataexcept/trace_context.py

def parse_traceparent(
    value: str,
    *,
    tracestate: str | None = None,
    baggage: str | None = None,
) -> W3CTraceContext | None:
    """Parse a W3C ``traceparent`` value, returning ``None`` when invalid.

    Version ``00`` uses the exact 55-character format. Higher versions are
    accepted when their mandatory prefix is parseable; any extension remains
    opaque and is preserved in ``traceparent`` for forwarding. No identifiers
    are generated when parsing fails.
    """
    if not isinstance(value, str):
        raise TypeError("traceparent must be a string")
    if value != value.strip() or len(value) < 55:
        return None
    if not _has_valid_separators(value):
        return None

    version = value[:2]
    trace_id = value[3:35]
    parent_id = value[36:52]
    trace_flags = value[53:55]

    if not _mandatory_fields_are_valid(version, trace_id, parent_id, trace_flags):
        return None
    if not _version_shape_is_valid(value, version):
        return None

    return W3CTraceContext(
        traceparent=value,
        version=version,
        trace_id=trace_id,
        parent_id=parent_id,
        trace_flags=trace_flags,
        tracestate=_safe_optional_header(tracestate),
        baggage=_safe_optional_header(baggage),
    )

trace_context_from_mapping

trace_context_from_mapping(carrier: Mapping[str, object]) -> W3CTraceContext | None

Extract W3C trace context from a case-insensitive mapping-like carrier.

The mapping can represent HTTP headers, RPC metadata, broker properties or protocol metadata such as MCP _meta. Invalid tracestate or baggage values are dropped without invalidating a valid traceparent.

Source code in dataexcept/trace_context.py

def trace_context_from_mapping(
    carrier: Mapping[str, object],
) -> W3CTraceContext | None:
    """Extract W3C trace context from a case-insensitive mapping-like carrier.

    The mapping can represent HTTP headers, RPC metadata, broker properties or
    protocol metadata such as MCP ``_meta``. Invalid ``tracestate`` or
    ``baggage`` values are dropped without invalidating a valid ``traceparent``.
    """
    if not isinstance(carrier, Mapping):
        raise TypeError("carrier must be a mapping")

    values: dict[str, object] = {}
    for key, value in carrier.items():
        if isinstance(key, str):
            normalized = key.lower()
            if normalized in {"traceparent", "tracestate", "baggage"}:
                values[normalized] = value

    traceparent = values.get("traceparent")
    if traceparent is None:
        return None
    if not isinstance(traceparent, str):
        return None

    return parse_traceparent(
        traceparent,
        tracestate=_safe_optional_header(values.get("tracestate")),
        baggage=_safe_optional_header(values.get("baggage")),
    )

HTTP request context

http_context

Framework-neutral HTTP request context for DataExcept observability.

The adapter works with plain header mappings and route templates, so ASGI, WSGI, serverless and RPC gateways can use the same logic without becoming runtime dependencies of DataExcept.

HttpContext

Request operation context plus optional W3C propagation context.

Source code in dataexcept/http_context.py

class HttpContext:
    """Request operation context plus optional W3C propagation context."""

    __slots__ = ("operation_context", "trace_context")

    def __init__(
        self,
        operation_context: OperationContext,
        trace_context: W3CTraceContext | None,
    ) -> None:
        self.operation_context = operation_context
        self.trace_context = trace_context

http_context_from_request

http_context_from_request(method: str, *, route: str | None = None, headers: Mapping[str, object] | None = None, system: str | None = 'http', component: str | None = None, request_id_headers: Sequence[str] = _DEFAULT_REQUEST_ID_HEADERS, correlation_id_headers: Sequence[str] = _DEFAULT_CORRELATION_ID_HEADERS) -> HttpContext

Build request observability context from plain HTTP-style inputs.

route should be a low-cardinality route template such as /users/{id}, never a raw request path. Request and correlation IDs are read from configurable headers. W3C trace context is propagated when present, but no trace or span identifiers are generated when it is absent.

Source code in dataexcept/http_context.py

def http_context_from_request(
    method: str,
    *,
    route: str | None = None,
    headers: Mapping[str, object] | None = None,
    system: str | None = "http",
    component: str | None = None,
    request_id_headers: Sequence[str] = _DEFAULT_REQUEST_ID_HEADERS,
    correlation_id_headers: Sequence[str] = _DEFAULT_CORRELATION_ID_HEADERS,
) -> HttpContext:
    """Build request observability context from plain HTTP-style inputs.

    ``route`` should be a low-cardinality route template such as
    ``/users/{id}``, never a raw request path. Request and correlation IDs are
    read from configurable headers. W3C trace context is propagated when
    present, but no trace or span identifiers are generated when it is absent.
    """
    if not isinstance(method, str):
        raise TypeError("method must be a string")
    normalized_method = method.strip().upper()
    if not _METHOD.fullmatch(normalized_method):
        raise ValueError("method must be a valid HTTP method token")
    if headers is None:
        headers = {}
    if not isinstance(headers, Mapping):
        raise TypeError("headers must be a mapping or None")

    normalized = _normalized_headers(headers)
    request_names = _validate_header_names(request_id_headers, "request_id_headers")
    correlation_names = _validate_header_names(
        correlation_id_headers,
        "correlation_id_headers",
    )

    operation_context = OperationContext(
        system=system,
        component=component,
        operation=_operation_name(normalized_method, route),
        request_id=_first_header(normalized, request_names),
        correlation_id=_first_header(normalized, correlation_names),
    )

    trace_context = trace_context_from_mapping(normalized)
    if trace_context is not None:
        operation_context = trace_context.to_operation_context(operation_context)

    return HttpContext(operation_context, trace_context)

Worker/task context

worker_context

Framework-neutral task/worker context for DataExcept observability.

The adapter accepts plain task metadata so Celery, RQ, Arq, Dramatiq, custom workers and orchestration executors can share one dependency-free boundary model. Task arguments and payloads are intentionally out of scope.

WorkerContext

Task operation context plus optional trace and retry metadata.

Source code in dataexcept/worker_context.py

class WorkerContext:
    """Task operation context plus optional trace and retry metadata."""

    __slots__ = ("attempt", "operation_context", "trace_context")

    def __init__(
        self,
        operation_context: OperationContext,
        trace_context: W3CTraceContext | None,
        attempt: int | None,
    ) -> None:
        self.operation_context = operation_context
        self.trace_context = trace_context
        self.attempt = attempt

worker_context_from_task

worker_context_from_task(task_name: str, *, job_id: str | None = None, metadata: Mapping[str, object] | None = None, system: str | None = 'worker', component: str | None = None, correlation_id: str | None = None, attempt: int | None = None, job_id_keys: Sequence[str] = _DEFAULT_JOB_ID_KEYS, correlation_id_keys: Sequence[str] = _DEFAULT_CORRELATION_ID_KEYS) -> WorkerContext

Build worker observability context from plain task metadata.

task_name should be the stable registered task name, never arguments or a rendered payload. Explicit job_id and correlation_id values take precedence over metadata fallbacks. W3C trace context is propagated when present and no identifiers are generated when it is absent.

Source code in dataexcept/worker_context.py

def worker_context_from_task(
    task_name: str,
    *,
    job_id: str | None = None,
    metadata: Mapping[str, object] | None = None,
    system: str | None = "worker",
    component: str | None = None,
    correlation_id: str | None = None,
    attempt: int | None = None,
    job_id_keys: Sequence[str] = _DEFAULT_JOB_ID_KEYS,
    correlation_id_keys: Sequence[str] = _DEFAULT_CORRELATION_ID_KEYS,
) -> WorkerContext:
    """Build worker observability context from plain task metadata.

    ``task_name`` should be the stable registered task name, never arguments or
    a rendered payload. Explicit ``job_id`` and ``correlation_id`` values take
    precedence over metadata fallbacks. W3C trace context is propagated when
    present and no identifiers are generated when it is absent.
    """
    if not isinstance(task_name, str):
        raise TypeError("task_name must be a string")
    normalized_task = task_name.strip()
    if not normalized_task:
        raise ValueError("task_name must not be empty")
    if "\r" in task_name or "\n" in task_name:
        raise ValueError("task_name must be a single line")

    if metadata is None:
        metadata = {}
    if not isinstance(metadata, Mapping):
        raise TypeError("metadata must be a mapping or None")

    normalized = _normalized_metadata(metadata)
    job_keys = _validated_keys(job_id_keys, "job_id_keys")
    correlation_keys = _validated_keys(
        correlation_id_keys,
        "correlation_id_keys",
    )

    explicit_job_id = _validated_optional_text(job_id, "job_id")
    explicit_correlation_id = _validated_optional_text(
        correlation_id,
        "correlation_id",
    )

    operation_context = OperationContext(
        system=system,
        component=component,
        operation=normalized_task,
        job_id=explicit_job_id or _first_metadata_value(normalized, job_keys),
        correlation_id=(
            explicit_correlation_id
            or _first_metadata_value(normalized, correlation_keys)
        ),
    )

    trace_context = trace_context_from_mapping(normalized)
    if trace_context is not None:
        operation_context = trace_context.to_operation_context(operation_context)

    return WorkerContext(
        operation_context,
        trace_context,
        _validate_attempt(attempt),
    )

Workflow/orchestrator context

orchestrator_context

Framework-neutral workflow/orchestrator observability context.

The adapter accepts plain workflow metadata so Airflow, Dagster, Prefect, Argo and custom schedulers can share the same dependency-free boundary model. Payloads and task arguments are intentionally excluded.

OrchestratorContext

Workflow-step context plus optional trace and retry metadata.

Source code in dataexcept/orchestrator_context.py

class OrchestratorContext:
    """Workflow-step context plus optional trace and retry metadata."""

    __slots__ = ("attempt", "operation_context", "step_run_id", "trace_context")

    def __init__(
        self,
        operation_context: OperationContext,
        trace_context: W3CTraceContext | None,
        *,
        step_run_id: str | None,
        attempt: int | None,
    ) -> None:
        self.operation_context = operation_context
        self.trace_context = trace_context
        self.step_run_id = step_run_id
        self.attempt = attempt

orchestrator_context_from_step

orchestrator_context_from_step(workflow: str, step: str, *, run_id: str | None = None, step_run_id: str | None = None, metadata: Mapping[str, object] | None = None, system: str | None = 'orchestrator', component: str | None = None, correlation_id: str | None = None, attempt: int | None = None) -> OrchestratorContext

Build observability context for one workflow/orchestrator step.

The operation name is the stable workflow:step pair. Run identifiers remain correlation metadata and are never folded into the operation name. Incoming W3C trace context is preserved when present.

Source code in dataexcept/orchestrator_context.py

def orchestrator_context_from_step(
    workflow: str,
    step: str,
    *,
    run_id: str | None = None,
    step_run_id: str | None = None,
    metadata: Mapping[str, object] | None = None,
    system: str | None = "orchestrator",
    component: str | None = None,
    correlation_id: str | None = None,
    attempt: int | None = None,
) -> OrchestratorContext:
    """Build observability context for one workflow/orchestrator step.

    The operation name is the stable ``workflow:step`` pair. Run identifiers
    remain correlation metadata and are never folded into the operation name.
    Incoming W3C trace context is preserved when present.
    """
    workflow_name = _single_line(workflow, "workflow")
    step_name = _single_line(step, "step")
    run_identifier = _optional_single_line(run_id, "run_id")
    step_run_identifier = _optional_single_line(step_run_id, "step_run_id")
    correlation_identifier = _optional_single_line(correlation_id, "correlation_id")

    if metadata is None:
        metadata = {}
    if not isinstance(metadata, Mapping):
        raise TypeError("metadata must be a mapping or None")

    operation_context = OperationContext(
        system=system,
        component=component,
        operation=f"{workflow_name}:{step_name}",
        job_id=run_identifier,
        correlation_id=correlation_identifier,
    )

    trace_context = trace_context_from_mapping(metadata)
    if trace_context is not None:
        operation_context = trace_context.to_operation_context(operation_context)

    return OrchestratorContext(
        operation_context,
        trace_context,
        step_run_id=step_run_identifier,
        attempt=_validate_attempt(attempt),
    )

Serverless invocation context

serverless_context

Framework-neutral serverless invocation context for observability.

The adapter accepts plain invocation metadata so AWS Lambda, Azure Functions, Google Cloud Functions, OpenFaaS and custom function runtimes can share one dependency-free boundary model. Event payloads are intentionally excluded.

ServerlessContext

Invocation operation context plus optional runtime metadata.

Source code in dataexcept/serverless_context.py

class ServerlessContext:
    """Invocation operation context plus optional runtime metadata."""

    __slots__ = ("cold_start", "operation_context", "runtime", "trace_context")

    def __init__(
        self,
        operation_context: OperationContext,
        trace_context: W3CTraceContext | None,
        *,
        cold_start: bool | None,
        runtime: str | None,
    ) -> None:
        self.operation_context = operation_context
        self.trace_context = trace_context
        self.cold_start = cold_start
        self.runtime = runtime

serverless_context_from_invocation

serverless_context_from_invocation(function_name: str, *, invocation_id: str | None = None, metadata: Mapping[str, object] | None = None, system: str | None = 'serverless', component: str | None = None, correlation_id: str | None = None, cold_start: bool | None = None, runtime: str | None = None, invocation_id_keys: Sequence[str] = _DEFAULT_INVOCATION_ID_KEYS, correlation_id_keys: Sequence[str] = _DEFAULT_CORRELATION_ID_KEYS) -> ServerlessContext

Build observability context for one serverless function invocation.

function_name is the stable deployed function identity. Per-invocation request IDs stay correlation metadata and event payloads are never captured. Explicit invocation and correlation IDs take precedence over metadata fallbacks. Incoming W3C trace context is preserved when present.

Source code in dataexcept/serverless_context.py

def serverless_context_from_invocation(
    function_name: str,
    *,
    invocation_id: str | None = None,
    metadata: Mapping[str, object] | None = None,
    system: str | None = "serverless",
    component: str | None = None,
    correlation_id: str | None = None,
    cold_start: bool | None = None,
    runtime: str | None = None,
    invocation_id_keys: Sequence[str] = _DEFAULT_INVOCATION_ID_KEYS,
    correlation_id_keys: Sequence[str] = _DEFAULT_CORRELATION_ID_KEYS,
) -> ServerlessContext:
    """Build observability context for one serverless function invocation.

    function_name is the stable deployed function identity. Per-invocation
    request IDs stay correlation metadata and event payloads are never captured.
    Explicit invocation and correlation IDs take precedence over metadata
    fallbacks. Incoming W3C trace context is preserved when present.
    """
    if not isinstance(function_name, str):
        raise TypeError("function_name must be a string")
    operation = function_name.strip()
    if not operation:
        raise ValueError("function_name must not be empty")
    if "\r" in function_name or "\n" in function_name:
        raise ValueError("function_name must be a single line")

    if metadata is None:
        metadata = {}
    if not isinstance(metadata, Mapping):
        raise TypeError("metadata must be a mapping or None")

    normalized = _normalized_metadata(metadata)
    invocation_keys = _validated_keys(invocation_id_keys, "invocation_id_keys")
    correlation_keys = _validated_keys(
        correlation_id_keys,
        "correlation_id_keys",
    )
    explicit_invocation_id = _validated_optional_text(
        invocation_id,
        "invocation_id",
    )
    explicit_correlation_id = _validated_optional_text(
        correlation_id,
        "correlation_id",
    )

    operation_context = OperationContext(
        system=system,
        component=component,
        operation=operation,
        request_id=(
            explicit_invocation_id or _first_metadata_value(normalized, invocation_keys)
        ),
        correlation_id=(
            explicit_correlation_id
            or _first_metadata_value(normalized, correlation_keys)
        ),
    )

    trace_context = trace_context_from_mapping(normalized)
    if trace_context is not None:
        operation_context = trace_context.to_operation_context(operation_context)

    return ServerlessContext(
        operation_context,
        trace_context,
        cold_start=_validated_cold_start(cold_start),
        runtime=_validated_optional_text(runtime, "runtime"),
    )

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, operation_context: OperationContext | None = None) -> None

Log exc at the given log level using logger.

If logger is None a module level logger is used.

context remains the free-form application context. operation_context carries the stable request/job/tool-call identifiers shared with tracing and error-tracker integrations.

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.

Logging is fail-open: context conversion, traceback rendering or the logger itself may fail, but that failure is swallowed so observability can never replace the exception the caller was already handling.

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,
    operation_context: OperationContext | None = None,
) -> None:
    """Log *exc* at the given log *level* using *logger*.

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

    ``context`` remains the free-form application context. ``operation_context``
    carries the stable request/job/tool-call identifiers shared with tracing and
    error-tracker integrations.

    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.

    Logging is fail-open: context conversion, traceback rendering or the logger
    itself may fail, but that failure is swallowed so observability can never
    replace the exception the caller was already handling.
    """
    try:
        if logger is None:
            logger = logging.getLogger(__name__)
        extra = _build_extra(context, operation_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)
    except Exception:
        return

log_and_raise

log_and_raise(logger: Optional[Logger] = None, level: int = logging.ERROR, context: Context | None = None, operation_context: OperationContext | 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,
    operation_context: OperationContext | 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,
            operation_context=operation_context,
        )
        raise

log_then_raise

log_then_raise(exc: Exception, logger: Optional[Logger] = None, level: int = logging.ERROR, context: Context | None = None, operation_context: OperationContext | 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,
    operation_context: OperationContext | 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,
        operation_context=operation_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()