Source code for src.util.exceptions

"""All framework-specific exceptions are placed in a single module to simplify
import os
import sys
import errno
from subprocess import CalledProcessError

import logging
_log = logging.getLogger(__name__)

[docs] def exit_on_exception(exc, msg=None): """Prints information about a fatal exception to the console before exiting. Use case is in user-facing subcommands (``mdtf install`` etc.), since we have more sophisticated logging in the framework itself. Args: exc (:py:class:`Exception`): Exception to print. msg (str): Optional, additional message to print. """ # if subprocess failed, will have already logged its own info print(f'ERROR: caught exception {repr(exc)}') if msg: print(msg) exit_handler(code=1)
[docs] def exit_handler(code=1, msg=None): """Wraps all calls to :py:func:`sys.exit`; could do additional cleanup not handled by atexit() here. """ if msg: print(msg) sys.exit(code)
[docs] def chain_exc(exc, new_msg, new_exc_class=None): """Raise a new exception from an existing one, in order to give more context for debugging. See Python documentation on `exception chaining <>`__. Args: exc (:py:class:`Exception`): Incoming exception to chain new exception from. new_msg (str): Message for new Exception. new_exc_class (class): Optional. Class of new exception to raise. If not provided, raises a new exception of the same type as *exc*. """ if new_exc_class is None: new_exc_class = type(exc) try: if new_msg.istitle(): new_msg = new_msg[0].lower() + new_msg[1:] if new_msg.endswith('.'): new_msg = new_msg[:-1] new_msg = f"{exc_descriptor(exc)} while {new_msg.lstrip()}: {repr(exc)}." raise new_exc_class(new_msg) from exc except Exception as chained_exc: return chained_exc
[docs] def exc_descriptor(exc): # MDTFEvents are raised during normal program operation; use correct wording # for log messages so user doesn't think it's an error if isinstance(exc, MDTFEvent): return "Received event" else: return "Caught exception"
[docs] class TimeoutAlarm(Exception): """Dummy exception raised if a subprocess times out.""" # NOTE py3 builds timeout into subprocess; fix this pass
[docs] class MDTFBaseException(Exception): """Base class to describe all MDTF-specific errors that can happen during the framework's operation.""" def __repr__(self): # full repr of attrs of child classes may take lots of space to print; # instead just print message return f'{self.__class__.__name__}("{str(self)}")'
[docs] class ChildFailureEvent(MDTFBaseException): """Exception raised when a member of the object hierarchy is deactivated because all its child objects have failed. """ def __init__(self, obj): self.obj = obj def __str__(self): return (f"Deactivating {self.obj.full_name} due to failure of all " f"child objects.")
[docs] class PropagatedEvent(MDTFBaseException): """Exception passed between members of the object hierarchy when a parent object (:class:`~core.MDTFObjectBase`) has been deactivated and needs to deactivate its children. """ def __init__(self, exc, parent): self.exc = exc self.parent = parent def __str__(self): return (f"{exc_descriptor(self.exc)} {repr(self.exc)} from deactivation " f"of parent {self.parent.full_name}.")
[docs] class MDTFFileNotFoundError(FileNotFoundError, MDTFBaseException): """Wrapper for :py:class:`FileNotFoundError` which handles error codes so we don't have to remember to import :py:mod:`errno` everywhere. """ def __init__(self, path): super(MDTFFileNotFoundError, self).__init__( errno.ENOENT, os.strerror(errno.ENOENT), path )
[docs] class MDTFFileExistsError(FileExistsError, MDTFBaseException): """Wrapper for :py:class:`FileExistsError` which handles error codes so we don't have to remember to import :py:mod:`errno` everywhere. """ def __init__(self, path): super(MDTFFileExistsError, self).__init__( errno.EEXIST, os.strerror(errno.EEXIST), path )
[docs] class MDTFCalledProcessError(CalledProcessError, MDTFBaseException): """Wrapper for :py:class:`subprocess.CalledProcessError`.""" pass
[docs] class WormKeyError(KeyError, MDTFBaseException): """Raised when attempting to overwrite or delete an entry in a :class:`~src.util.basic.WormDict`. """ pass
[docs] class DataclassParseError(ValueError, MDTFBaseException): """Raised when parsing input data fails on a :func:`~src.util.dataclass.mdtf_dataclass` or :func:`~src.util.dataclass.regex_dataclass`. """ pass
[docs] class RegexParseError(ValueError, MDTFBaseException): """Raised when parsing input data fails on a :func:`~src.util.dataclass.RegexPattern`. """ pass
[docs] class RegexSuppressedError(ValueError, MDTFBaseException): """Raised when parsing input data fails on a :func:`~src.util.dataclass.RegexPattern`, but we've decided to supress error based on the associated RegexPattern's match_error_filter attribute. """ pass
[docs] class UnitsError(ValueError, MDTFBaseException): """Raised when trying to convert between quantities with physically inequivalent units. """ pass
[docs] class ConventionError(MDTFBaseException): """Exception raised by a duplicate variable convention name.""" def __init__(self, conv_name): self.conv_name = conv_name def __str__(self): return f"Error in the definition of convention '{self.conv_name}'."
[docs] class MixedDatePrecisionException(MDTFBaseException): """Exception raised when we attempt to operate on :class:`Date` or :class:`DateRange` objects with differing levels of precision, which shouldn't happen with data sampled at a single frequency. """ def __init__(self, func_name='', msg=''): self.func_name = func_name self.msg = msg def __str__(self): return ("Attempted datelabel method '{}' on FXDate " "placeholder: {}.").format(self.func_name, self.msg)
[docs] class FXDateException(MDTFBaseException): """Exception raised when :class:`FXDate` or :class:`FXDateRange` classes, which are placeholder/sentinel classes used to indicate static data with no time dependence, are accessed like real :class:`Date` or :class:`DateRange` objects. """ def __init__(self, func_name='', msg=''): self.func_name = func_name self.msg = msg def __str__(self): return ("Attempted datelabel method '{}' on FXDate " "placeholder: {}.").format(self.func_name, self.msg)
[docs] class DataRequestError(MDTFBaseException): """Dummy class used for fatal errors that take place during the data query/fetch/preprocess stage of the framework. """ pass
[docs] class MDTFEvent(MDTFBaseException): """Dummy class to denote non-fatal errors, specifically "events" that are passed during the data query/fetch/preprocess stage of the framework. """ pass
[docs] class FatalErrorEvent(MDTFBaseException): """Dummy class used to "convert" :class:`MDTFEvent`\s to fatal errors (resulting in deactivation of a variable, pod or case.) via exception chaining. """ pass
[docs] class DataProcessingEvent(MDTFEvent): """Base class and common formatting code for events raised in data query/fetch. These should *not* be used for fatal errors (when a variable or POD is deactivated.) """ def __init__(self, msg="", dataset=None): self.msg = msg self.dataset = dataset def __str__(self): # if self.dataset is not None: # if hasattr(self.dataset, 'remote_path'): # data_id = self.dataset.remote_path # elif hasattr(self.dataset, 'name'): # data_id = # else: # data_id = str(self.dataset) return self.msg
[docs] class DataQueryEvent(DataProcessingEvent): """Exception signaling a failure to find requested data in the remote location. """ pass
[docs] class DataExperimentEvent(DataProcessingEvent): """Exception signaling a failure to uniquely select an experiment for all variables based on query results. """ pass
[docs] class DataFetchEvent(DataProcessingEvent): """Exception signaling a failure to obtain data from the remote location. """ pass
[docs] class DataPreprocessEvent(DataProcessingEvent): """Exception signaling an error in preprocessing data after it's been fetched, but before any PODs run. """ pass
[docs] class MetadataEvent(DataProcessingEvent): """Exception signaling discrepancies in variable metadata. """ pass
[docs] class MetadataError(MDTFBaseException): """Exception signaling unrecoverable errors in variable metadata. """ pass
[docs] class UnitsUndefinedError(MetadataError): """Exception signaling unrecoverable errors in variable metadata. """ pass
[docs] class GenericDataSourceEvent(DataProcessingEvent): """Exception signaling a failure originating in the DataSource query/fetch pipeline whose cause doesn't fall into the above categories. """ pass
[docs] class PodExceptionBase(MDTFBaseException): """Base class and common formatting code for exceptions affecting a single POD. """ _error_str = "" def __init__(self, msg=None, pod=None): self.pod = pod self.msg = msg def __str__(self): s = self._error_str if self.pod is not None: if hasattr(self.pod, 'full_name'): pod_name = self.pod.full_name else: pod_name = f"'{self.pod}'" s += f" for POD {pod_name}" if self.msg is not None: s += f": {self.msg}" if not s.endswith('.'): s += "." return s
[docs] class PodConfigError(PodExceptionBase): """Exception raised if we can't parse info in a POD's settings.jsonc file. (Covers issues with the file format/schema; malformed JSONC will raise a :py:class:`~json.JSONDecodeError` when :func:`~util.parse_json` attempts to parse the file. """ _error_str = "Couldn't parse the settings.jsonc file"
[docs] class PodConfigEvent(MDTFEvent): """Exception raised during non-fatal events in resolving POD configuration. """ pass
[docs] class PodDataError(PodExceptionBase): """Exception raised if POD doesn't have required data to run. """ _error_str = "Requested data not available"
[docs] class PodRuntimeError(PodExceptionBase): """Exception raised if POD doesn't have required resources to run. """ _error_str = "Error in setting the runtime environment"
[docs] class PodExecutionError(PodExceptionBase): """Exception raised if POD exits with non-zero retcode or otherwise raises an error during execution. """ _error_str = "Error during POD execution"