Class DirtyError

gunicorn/dirty/errors.py:12–77

Base exception for all dirty arbiter errors.

class DirtyError(Exception):
    """Base exception for all dirty arbiter errors."""

    def __init__(self, message, details=None):
        self.message = message
        self.details = details or {}
        super().__init__(message)

    def __str__(self):
        if self.details:
            return f"{self.message}: {self.details}"
        return self.message

    def to_dict(self):
        """Serialize error for protocol transmission."""
        return {
            "error_type": self.__class__.__name__,
            "message": self.message,
            "details": self.details,
        }

    @classmethod
    def from_dict(cls, data):
        """Deserialize error from protocol transmission.

        Creates an error instance from a serialized dict. The returned
        error will be an instance of the appropriate subclass based on
        the error_type field, but constructed using the base DirtyError
        __init__ to preserve all details.
        """
        error_classes = {
            "DirtyError": DirtyError,
            "DirtyTimeoutError": DirtyTimeoutError,
            "DirtyConnectionError": DirtyConnectionError,
            "DirtyWorkerError": DirtyWorkerError,
            "DirtyAppError": DirtyAppError,
            "DirtyAppNotFoundError": DirtyAppNotFoundError,
            "DirtyNoWorkersAvailableError": DirtyNoWorkersAvailableError,
            "DirtyProtocolError": DirtyProtocolError,
        }
        error_type = data.get("error_type", "DirtyError")
        error_class = error_classes.get(error_type, DirtyError)

        # Create instance and set attributes directly to bypass
        # subclass __init__ complexity while preserving error type
        error = Exception.__new__(error_class)
        error.message = data.get("message", "Unknown error")
        error.details = data.get("details") or {}
        Exception.__init__(error, error.message)

        # Set subclass-specific attributes from details
        if error_class == DirtyTimeoutError:
            error.timeout = error.details.get("timeout")
        elif error_class == DirtyConnectionError:
            error.socket_path = error.details.get("socket_path")
        elif error_class == DirtyWorkerError:
            error.worker_id = error.details.get("worker_id")
            error.traceback = error.details.get("traceback")