forked from goodboy/tractor
1
0
Fork 0

Add type annots to exceptions mod

Also add a `is_multi_cancelled()` predicate to test for
`trio.MultiError`s that contain entirely cancel signals.

Resolves #125
denoise_logging
Tyler Goodlet 2020-12-25 15:07:36 -05:00
parent f4f39c29f3
commit 8522f90000
1 changed files with 20 additions and 3 deletions

View File

@ -1,6 +1,7 @@
"""
Our classy exception set.
"""
from typing import Dict, Any
import importlib
import builtins
import traceback
@ -14,7 +15,7 @@ _this_mod = importlib.import_module(__name__)
class RemoteActorError(Exception):
# TODO: local recontruction of remote exception deats
"Remote actor exception bundled locally"
def __init__(self, message, type_str, **msgdata):
def __init__(self, message, type_str, **msgdata) -> None:
super().__init__(message)
for ns in [builtins, _this_mod, trio]:
try:
@ -45,7 +46,7 @@ class ModuleNotExposed(ModuleNotFoundError):
"The requested module is not exposed for RPC"
def pack_error(exc):
def pack_error(exc: BaseException) -> Dict[str, Any]:
"""Create an "error message" for tranmission over
a channel (aka the wire).
"""
@ -57,7 +58,11 @@ def pack_error(exc):
}
def unpack_error(msg, chan=None, err_type=RemoteActorError):
def unpack_error(
msg: Dict[str, Any],
chan=None,
err_type=RemoteActorError
) -> Exception:
"""Unpack an 'error' message from the wire
into a local ``RemoteActorError``.
"""
@ -66,3 +71,15 @@ def unpack_error(msg, chan=None, err_type=RemoteActorError):
f"{chan.uid}\n" + tb_str,
**msg['error'],
)
def is_multi_cancelled(exc: BaseException) -> bool:
"""Predicate to determine if a ``trio.MultiError`` contains only
``trio.Cancelled`` sub-exceptions (and is likely the result of
cancelling a collection of subtasks.
"""
return not trio.MultiError.filter(
lambda exc: exc if not isinstance(exc, trio.Cancelled) else None,
exc,
)