Source code for WDL.Error

# pyre-strict
from typing import (
from functools import total_ordering
from contextlib import contextmanager

from . import Type

[docs]class SourcePosition( NamedTuple( "SourcePosition", [ ("uri", str), ("abspath", str), ("line", int), ("column", int), ("end_line", int), ("end_column", int), ], ) ): """ Source position attached to AST nodes and exceptions; NamedTuple of ``uri`` the filename/URI passed to :func:`WDL.load` or a WDL import statement, which may be relative; ``abspath`` the absolute filename/URI; and one-based int positions ``line`` ``end_line`` ``column`` ``end_column`` """
[docs]class SyntaxError(Exception): """Failure to lex/parse a WDL document""" pos: SourcePosition wdl_version: str declared_wdl_version: Optional[str] def __init__( self, pos: SourcePosition, msg: str, wdl_version: str, declared_wdl_version: Optional[str] ) -> None: super().__init__(msg) self.pos = pos self.wdl_version = wdl_version self.declared_wdl_version = declared_wdl_version
[docs]class ImportError(Exception): """Failure to open/retrieve an imported WDL document The ``__cause__`` attribute may hold the inner error object.""" pos: SourcePosition def __init__(self, pos: SourcePosition, import_uri: str, message: Optional[str] = None) -> None: msg = "Failed to import " + import_uri if message: msg = msg + ", " + message super().__init__(msg) self.pos = pos
TVSourceNode = TypeVar("TVSourceNode", bound="SourceNode")
[docs]@total_ordering class SourceNode: """Base class for an AST node, recording the source position""" pos: SourcePosition """ :type: SourcePosition Source position for this AST node """ def __init__(self, pos: SourcePosition) -> None: self.pos = pos def __lt__(self, rhs: TVSourceNode) -> bool: if isinstance(rhs, SourceNode): return ( self.pos.abspath, self.pos.line, self.pos.column, self.pos.end_line, self.pos.end_column, ) < ( rhs.pos.abspath, rhs.pos.line, rhs.pos.column, rhs.pos.end_line, rhs.pos.end_column, ) return False def __eq__(self, rhs: TVSourceNode) -> bool: assert isinstance(rhs, SourceNode) return self.pos == rhs.pos @property def children(self: TVSourceNode) -> Iterable[TVSourceNode]: """ :type: Iterable[SourceNode] Yield all child nodes """ return []
[docs]class ValidationError(Exception): """ Base class for a WDL validation error (when the document loads and parses, but fails typechecking or other static validity tests) """ pos: SourcePosition """:type: SourcePosition""" node: Optional[SourceNode] = None """:type: Optional[SourceNode]""" source_text: Optional[str] = None """:type: Optional[str] The complete source text of the WDL document (if available)""" def __init__(self, node: Union[SourceNode, SourcePosition], message: str) -> None: if isinstance(node, SourceNode): self.node = node self.pos = node.pos else: self.pos = node super().__init__(message)
[docs]class InvalidType(ValidationError): pass
[docs]class IndeterminateType(ValidationError): pass
[docs]class NoSuchTask(ValidationError): def __init__(self, node: Union[SourceNode, SourcePosition], name: str) -> None: super().__init__(node, "No such task/workflow: " + name)
[docs]class NoSuchCall(ValidationError): def __init__(self, node: Union[SourceNode, SourcePosition], name: str) -> None: super().__init__(node, "No such call in this workflow: " + name)
[docs]class NoSuchFunction(ValidationError): def __init__(self, node: SourceNode, name: str) -> None: super().__init__(node, "No such function: " + name)
[docs]class WrongArity(ValidationError): def __init__(self, node: SourceNode, expected: int) -> None: # avoiding circular dep: # assert isinstance(node, WDL.Expr.Apply) msg = "{} expects {} argument(s)".format(getattr(node, "function_name"), expected) super().__init__(node, msg)
[docs]class NotAnArray(ValidationError): def __init__(self, node: SourceNode) -> None: super().__init__(node, "Not an array")
[docs]class NoSuchMember(ValidationError): def __init__(self, node: SourceNode, member: str) -> None: super().__init__(node, "No such member '{}'".format(member))
[docs]class StaticTypeMismatch(ValidationError): message: str def __init__( self, node: SourceNode, expected: Type.Base, actual: Type.Base, message: str = "" ) -> None: self.expected = expected self.actual = actual self.message = message super().__init__(node, message) def __str__(self) -> str: msg = f"Expected {self.expected} instead of {self.actual}" if self.message: msg += "; " + self.message elif isinstance(self.expected, Type.Int) and isinstance(self.actual, Type.Float): msg += "; perhaps try floor() or round()" elif str(self.actual).replace("?", "") == str(self.expected): msg += ( " -- to coerce T? X into T, try select_first([X,defaultValue])" " or select_first([X]) (which might fail at runtime);" " to coerce Array[T?] X into Array[T], try select_all(X)" ) return msg
[docs]class IncompatibleOperand(ValidationError): def __init__(self, node: SourceNode, message: str) -> None: super().__init__(node, message)
[docs]class UnknownIdentifier(ValidationError): def __init__(self, node: SourceNode, message: Optional[str] = None) -> None: # avoiding circular dep: # assert isinstance(node, WDL.Expr.Ident) if not message: message = "Unknown identifier " + str(node) super().__init__(node, message)
[docs]class NoSuchInput(ValidationError): def __init__(self, node: SourceNode, name: str) -> None: super().__init__(node, "No such input " + name)
[docs]class UncallableWorkflow(ValidationError): def __init__(self, node: SourceNode, name: str) -> None: super().__init__( node, ( "Cannot call workflow {} because its calls don't supply all required inputs, " "or it lacks an output section" ).format(name), )
[docs]class MultipleDefinitions(ValidationError): pass
[docs]class StrayInputDeclaration(ValidationError): pass
[docs]class CircularDependencies(ValidationError): def __init__(self, node: SourceNode) -> None: msg = "circular dependencies" nm = next( (getattr(node, attr) for attr in ("name", "workflow_node_id") if hasattr(node, attr)), None, ) if nm: nm += " involving " + nm super().__init__(node, msg)
[docs]class MultipleValidationErrors(Exception): """Propagates several validation/typechecking errors""" exceptions: List[ValidationError] """:type: List[ValidationError]""" def __init__( self, *exceptions: List[Union[ValidationError, "MultipleValidationErrors"]] ) -> None: super().__init__() self.exceptions = [] for exn in exceptions: if isinstance(exn, ValidationError): self.exceptions.append(exn) elif isinstance(exn, MultipleValidationErrors): self.exceptions.extend(exn.exceptions) else: assert False assert self.exceptions self.exceptions = sorted(self.exceptions, key=lambda exn: getattr(exn, "pos"))
class _MultiContext: """""" _exceptions: List[Union[ValidationError, MultipleValidationErrors]] def __init__(self) -> None: self._exceptions = [] def try1(self, fn: Callable[[], Any]) -> Optional[Any]: # pyre-ignore try: return fn() except (ValidationError, MultipleValidationErrors) as exn: self._exceptions.append(exn) return None def append(self, exn: Union[ValidationError, MultipleValidationErrors]) -> None: self._exceptions.append(exn) def maybe_raise(self) -> None: if len(self._exceptions) == 1: raise self._exceptions[0] if self._exceptions: # pyre-ignore raise MultipleValidationErrors(*self._exceptions) from self._exceptions[0] @contextmanager def multi_context() -> Generator[_MultiContext, None, None]: """""" # Context manager to assist with catching and propagating multiple # validation/typechecking errors # # with WDL.Error.multi_context() as errors: # # result = errors.try1(lambda: perform_validation()) # # Returns the result of invoking the lambda. If the lambda invocation # # raises WDL.Error.ValidationError or # # WDL.Error.MultipleValidationErrors, records the error and returns # # None. (Other exceptions would halt execution and propagate # # normally.) # # errors.append(WDL.Error.NullValue()) # # errors.append() manually records one error. # # When the context closes, any exceptions recorded with errors.try1() or # errors.append() are raised at that point. Note that any exception raised # outside of errors.try1() will exit the context immediately and discard # any previously-recorded errors. # # Lastly, you can call errors.maybe_raise() to immediately propagate any # exceptions recorded so far, or if none, proceed with the remainder of # the context body. ctx = _MultiContext() yield ctx ctx.maybe_raise()
[docs]class RuntimeError(Exception): more_info: Dict[str, Any] """ Backend-specific information about an error (for example, pointer to a centralized log system) """ # pyre-ignore def __init__(self, *args, more_info: Optional[Dict[str, Any]] = None, **kwargs) -> None: super().__init__(*args, **kwargs) self.more_info = more_info if more_info else {}
[docs]class EvalError(RuntimeError): """Error evaluating a WDL expression or declaration""" pos: SourcePosition """:type: SourcePosition""" node: Optional[SourceNode] = None """:type: Optional[SourceNode]""" def __init__(self, node: Union[SourceNode, SourcePosition], message: str) -> None: if isinstance(node, SourceNode): self.node = node self.pos = node.pos else: self.pos = node super().__init__(message)
[docs]class OutOfBounds(EvalError): pass
[docs]class EmptyArray(EvalError): def __init__(self, node: SourceNode) -> None: super().__init__(node, "Empty array for Array+ input/declaration")
[docs]class NullValue(EvalError): def __init__(self, node: Union[SourceNode, SourcePosition]) -> None: super().__init__(node, "Null value")
[docs]class InputError(RuntimeError): """Error reading an input value/file""" pass