webber.xcoms

Webber classes and functions related to inter-function data exchange.

Current iteration dependent on Webber tasks running on a common machine. Uses concurrent.futures.ThreadPoolExecutor to maintain data as return results.

Last Updated: pre-0.0.1

 1"""
 2Webber classes and functions related to inter-function data exchange.
 3
 4Current iteration dependent on Webber tasks running on a common machine.
 5Uses `concurrent.futures.ThreadPoolExecutor` to maintain data as return results.
 6
 7Last Updated: pre-0.0.1
 8"""
 9__all__ = ["InvalidCallable", "Promise"]
10
11import typing as _T
12
13class InvalidCallable(Exception):
14    """Requested Webber Promise is invalid in DAG's given scope/context."""
15    def __init__(self, *args):
16        super().__init__()
17        if args:
18            self.message = args[0]
19        else:
20            self.message = None
21
22    def __str__(self):
23        if self.message:
24            return f"{self.message}"
25        return "InvalidCallable"
26
27
28class Promise:
29    """A simple object class used to handle function intercoms for Webber's DAG executor."""
30    key: _T.Union[str, _T.Callable]
31    def __init__(self, _key: _T.Union[str, _T.Callable]) -> None:
32        """Initializing Promise using a function identifier (ID string or callable)."""
33        if not isinstance(_key, _T.Callable) and not isinstance(_key, str):
34            err_msg = "Keys must be string IDs or callables to be assigned to a Webber Promise"
35            raise TypeError(err_msg)
36        self.key = _key
class InvalidCallable(builtins.Exception): 
14class InvalidCallable(Exception):
15    """Requested Webber Promise is invalid in DAG's given scope/context."""
16    def __init__(self, *args):
17        super().__init__()
18        if args:
19            self.message = args[0]
20        else:
21            self.message = None
22
23    def __str__(self):
24        if self.message:
25            return f"{self.message}"
26        return "InvalidCallable"

Requested Webber Promise is invalid in DAG's given scope/context.

InvalidCallable(*args) 
16    def __init__(self, *args):
17        super().__init__()
18        if args:
19            self.message = args[0]
20        else:
21            self.message = None
Inherited Members
builtins.BaseException
with_traceback
add_note
args
class Promise: 
29class Promise:
30    """A simple object class used to handle function intercoms for Webber's DAG executor."""
31    key: _T.Union[str, _T.Callable]
32    def __init__(self, _key: _T.Union[str, _T.Callable]) -> None:
33        """Initializing Promise using a function identifier (ID string or callable)."""
34        if not isinstance(_key, _T.Callable) and not isinstance(_key, str):
35            err_msg = "Keys must be string IDs or callables to be assigned to a Webber Promise"
36            raise TypeError(err_msg)
37        self.key = _key

A simple object class used to handle function intercoms for Webber's DAG executor.

Promise(_key: Union[str, Callable]) 
32    def __init__(self, _key: _T.Union[str, _T.Callable]) -> None:
33        """Initializing Promise using a function identifier (ID string or callable)."""
34        if not isinstance(_key, _T.Callable) and not isinstance(_key, str):
35            err_msg = "Keys must be string IDs or callables to be assigned to a Webber Promise"
36            raise TypeError(err_msg)
37        self.key = _key

Initializing Promise using a function identifier (ID string or callable).

key: Union[str, Callable]