### AsyncDAG.setup(target_nodes=None, exclude_nodes=None, root_nodes=None) Source: https://mindee.github.io/tawazi/dag Runs the setup ExecNodes for the DAG. If target_nodes are provided, runs only the necessary setup ExecNodes, otherwise runs all setup ExecNodes. ```APIDOC ## AsyncDAG.setup(target_nodes=None, exclude_nodes=None, root_nodes=None) ### Description Runs the setup ExecNodes for the DAG. If target_nodes are provided, runs only the necessary setup ExecNodes, otherwise runs all setup ExecNodes. NOTE: DAG arguments should not be passed to setup ExecNodes. Only pass in constants or setup ExecNodes results. ### Method async setup ### Parameters #### Arguments - target_nodes (Optional[List[XNId]]) - Optional - The ExecNodes that the user aims to use in the DAG. This might include setup or non setup ExecNodes. If None is provided, will run all setup ExecNodes. Defaults to None. - exclude_nodes (Optional[List[XNId]]) - Optional - The ExecNodes that the user aims to exclude from the DAG. The user is responsible for ensuring that the overlapping between the target_nodes and exclude_nodes is logical. - root_nodes (Optional[List[XNId]]) - Optional - The ExecNodes that the user aims to select as ancestor nodes. The user is responsible for ensuring that the overlapping between the target_nodes, the exclude_nodes and the root nodes is logical. ``` -------------------------------- ### DAGExecution.setup Source: https://mindee.github.io/tawazi/DAGExecution Performs the same setup as DAG.setup but uses the target_nodes and exclude_nodes provided during DAGExecution initialization. ```APIDOC ## DAGExecution.setup ### Description Same thing as DAG.setup but `target_nodes` and `exclude_nodes` come from the DAGExecution's init. ### Method setup ### Parameters None ### Returns None ``` -------------------------------- ### DAG.setup(target_nodes=None, exclude_nodes=None, root_nodes=None) Source: https://mindee.github.io/tawazi/dag Runs the setup ExecNodes for the DAG, optionally targeting specific nodes. Note that DAG arguments should not be passed to setup ExecNodes. ```APIDOC ## DAG.setup(target_nodes=None, exclude_nodes=None, root_nodes=None) ### Description Run the setup ExecNodes for the DAG. If target_nodes are provided, run only the necessary setup ExecNodes, otherwise will run all setup ExecNodes. NOTE: DAG arguments should not be passed to setup ExecNodes. Only pass in constants or setup ExecNode results. ### Method setup ### Parameters - target_nodes (Optional[List[XNId]]) - Optional - The ExecNodes that the user aims to use in the DAG. This might include setup or non setup ExecNodes. If None is provided, will run all setup ExecNodes. Defaults to None. - exclude_nodes (Optional[List[XNId]]) - Optional - The ExecNodes that the user aims to exclude from the DAG. The user is responsible for ensuring that the overlapping between the target_nodes and exclude_nodes is logical. - root_nodes (Optional[List[XNId]]) - Optional - The ExecNodes that the user aims to select as ancestor nodes. The user is responsible for ensuring that the overlapping between the target_nodes, the exclude_nodes and the root nodes is logical. ``` -------------------------------- ### Executing SetupExecNode alone Source: https://mindee.github.io/tawazi You can trigger the setup process for a DAG independently using the setup() method. ```python @dag def pipeline(): cst_data = setup_node() large_data = my_print(cst_data) return large_data pipeline.setup() ``` -------------------------------- ### AsyncDAGExecution.setup Source: https://mindee.github.io/tawazi/DAGExecution Asynchronously performs the same setup as DAG.setup but uses the target_nodes and exclude_nodes provided during DAGExecution initialization. ```APIDOC ## AsyncDAGExecution.setup ### Description Same thing as DAG.setup but `target_nodes` and `exclude_nodes` come from the DAGExecution's init. ### Method setup (async) ### Parameters None ### Returns None ``` -------------------------------- ### Re-running SetupExecNode with deepcopy Source: https://mindee.github.io/tawazi To re-run a setup ExecNode, redeclare the DAG or use deepcopy on the instance. ```python from copy import deepcopy @dag def pipeline(): cst_data = setup_node() large_data = my_print(cst_data) return large_data setup_counter = 0 assert LARGE_DATA == deepcopy(pipeline)() assert setup_counter == 1 assert LARGE_DATA == deepcopy(pipeline)() assert setup_counter == 2 ``` -------------------------------- ### Executing Subgraphs with root_nodes Source: https://mindee.github.io/tawazi Shows how to define the starting point of a subgraph execution using root_nodes. ```python pipe_exec = pipeline.executor(root_nodes=["b"]) # will select all nodes depending on "b" pipe_exec() ``` -------------------------------- ### Compose a DAG with pipe.compose Source: https://mindee.github.io/tawazi/dag Shows how to use the compose method to create a new DAG from existing ExecNodes. The example defines several ExecNodes (unwanted_xn, x, y, z) and a DAG (pipe) that uses them, then composes a new DAG with inputs [x, y] and output z. The composed DAG is called with arguments (1, 1) and asserts the result equals 2.0. The example also includes commented-out lines showing optional typing of the returned DAG. ```python >>> from tawazi import dag, xn, DAG >>> from typing import Tuple, Any >>> @xn ... def unwanted_xn() -> int: return 42 >>> @xn ... def x(v: Any) -> int: return int(v) >>> @xn ... def y(v: Any) -> str: return str(v) >>> @xn ... def z(x: int, y: str) -> float: return float(x) + float(y) >>> @dag ... def pipe() -> Tuple[int, float, int]: ... a = unwanted_xn() ... res = z(x(1), y(1)) ... b = unwanted_xn() ... return a, res, b >>> composed_dag = pipe.compose("twinkle", [x, y], z) >>> assert composed_dag(1, 1) == 2.0 >>> # composed_dag: DAG[[int, str], float] = pipe.compose([x, y], [z]) # optional typing of the returned DAG! >>> # assert composed_dag(1, 1) == 2.0 # type checked! ``` -------------------------------- ### DAG.setup Source: https://mindee.github.io/tawazi/dag Sets up the DAG for execution. ```APIDOC ## DAG.setup ### Description Sets up the DAG for execution. ``` -------------------------------- ### AsyncDAG.setup Source: https://mindee.github.io/tawazi/dag Sets up the DAG for asynchronous execution. ```APIDOC ## AsyncDAG.setup ### Description Sets up the DAG for asynchronous execution. ``` -------------------------------- ### Executing subgraphs with SetupExecNode Source: https://mindee.github.io/tawazi SetupExecNodes allow loading only necessary resources when executing specific subgraphs. ```python from pprint import PrettyPrinter @xn(setup=True) def setup_node_1(): return "large data 1" @xn(setup=True) def setup_node_2(): return "large data 2" @xn def print_xn(val): print(val) @xn def pprint_xn(val): PrettyPrinter(indent=4).pprint(val) @dag def pipeline(): data1 = setup_node_1() data2 = setup_node_2() print_xn(data1) pprint_xn(data2) return data1, data2 exec_ = pipeline.executor(target_nodes=["print_xn"]) # Notice how the execution of the subgraph doesn't run the setop1 `ExecNode`. # This makes development of your complex pipeline faster by loading only the necessary resources. assert ("large data 1", None) == exec_() ``` -------------------------------- ### BaseDAG.get_nodes_by_tag(tag) Source: https://mindee.github.io/tawazi/dag Gets the ExecNodes with the given tag. The returned ExecNodes are not modified by any execution, so their results are not available via this method. ```APIDOC ## BaseDAG.get_nodes_by_tag(tag) ### Description Gets the ExecNodes with the given tag. The returned ExecNodes are not modified by any execution, so their results are not available via this method. ### Method get_nodes_by_tag ### Parameters - tag (Any) - Required - tag of the ExecNodes ### Returns - List[ExecNode] - corresponding ExecNodes ``` -------------------------------- ### BaseDAG.get_node_by_id(node_id) Source: https://mindee.github.io/tawazi/dag Gets the ExecNode with the given id. The returned ExecNode is not modified by any execution, so its result is not available via this method. ```APIDOC ## BaseDAG.get_node_by_id(node_id) ### Description Gets the ExecNode with the given id. The returned ExecNode is not modified by any execution, so its result is not available via this method. ### Method get_node_by_id ### Parameters - node_id (Identifier) - Required - id of the ExecNode ### Returns - ExecNode - corresponding ExecNode ``` -------------------------------- ### Caching results with SetupExecNode Source: https://mindee.github.io/tawazi SetupExecNode results are cached per DAG instance, useful for loading large resources once. ```python LARGE_DATA = "Long algorithm to generate Constant Data" @xn(setup=True) def setup_node(): global setup_counter setup_counter += 1 return LARGE_DATA @xn def my_print(arg): print(arg) return arg @dag def pipeline(): cst_data = setup_node() large_data = my_print(cst_data) return large_data setup_counter = 0 # create another instance because setup ExecNode result is cached inside the instance assert LARGE_DATA == pipeline() assert setup_counter == 1 assert LARGE_DATA == pipeline() assert setup_counter == 1 # setop_counter is skipped the second time pipe1 is invoked ``` -------------------------------- ### Constructing DAGExecution Source: https://mindee.github.io/tawazi Shows how to initialize a DAGExecution instance from a DAG to manage execution and access results. ```python from tawazi import DAGExecution # construct a DAGExecution from a DAG by doing dx = DAGExecution(pipeline) # or dx = pipeline.executor() ``` -------------------------------- ### Using Qualname for Target Nodes Source: https://mindee.github.io/tawazi Shows how to specify target nodes using the __qualname__ string identifier. ```python pipe_exec = pipeline.executor(target_nodes=["b"]) pipe_exec() ``` -------------------------------- ### Tagging ExecNodes Source: https://mindee.github.io/tawazi Shows how to assign single or multiple tags to ExecNodes and retrieve them using get_nodes_by_tag. ```python @xn(tag=("twinkle toes", "hot bird")) def a(): print("I am tough") @xn(tag=("yummy jam")) def c(): print("I am normal") @dag def pipeline(): a() c() pipeline() xn_a, = pipeline.get_nodes_by_tag("twinkle toes") xn_b, = pipeline.get_nodes_by_tag("hot bird") xn_c, = pipeline.get_nodes_by_tag("yummy jam") assert xn_a == xn_b ``` -------------------------------- ### Using Tags for Target Nodes Source: https://mindee.github.io/tawazi Shows how to specify target nodes using a tag string. ```python pipe_exec = pipeline.executor(target_nodes=["c_node"]) pipe_exec() ``` -------------------------------- ### Basic usage of @xn and @dag decorators Source: https://mindee.github.io/tawazi Demonstrates how to wrap functions with @xn to create LazyExecNodes and @dag to define a pipeline. The decorated functions behave as DAG objects when called. ```python from tawazi import xn, dag @xn def incr(x): return x + 1 # incr is no longer a function # incr became a `LazyExecNode` which is a subclass of `ExecNode`. print(type(incr)) ## @xn def decr(x): return x - 1 @xn def display(x): print(x) @dag def pipeline(x): x_lo = decr(x) x_hi = incr(x) display(x_hi) display(x_lo) # pipeline is no longer a function # pipeline became a `DAG` print(type(pipeline)) ## # `DAG` can be executed, they behave the same as the original function without decorators. pipeline(0) ``` -------------------------------- ### Using Calling Tags for Target Nodes Source: https://mindee.github.io/tawazi Shows how to specify target nodes using a specific calling tag. ```python pipe_exec = pipeline.executor(target_nodes=["byebye"]) pipe_exec() ``` -------------------------------- ### Tagging Specific ExecNode Calls Source: https://mindee.github.io/tawazi Demonstrates tagging a specific invocation of an ExecNode within a pipeline. ```python @xn def stub_xn(i): return i @xn(tag="c_node") def print_xn(i): print(i) @dag def pipeline(): b() hello = stub_xn("hello") print_xn(hello) goodbye = stub_xn("goodbye") print_xn(goodbye, twz_tag="byebye") pipeline() # multiple nodes can have the same tag! xns_bye = pipeline.get_nodes_by_tag("byebye") ``` -------------------------------- ### Multiple Tags for ExecNodes Source: https://mindee.github.io/tawazi Illustrates that multiple ExecNodes can share the same tag and a single ExecNode can have multiple tags. ```python @xn(tag=("twinkle", "toes")) def a(): print("I am tough") @xn(tag="twinkle") def b(): print("I am light") @dag def pipeline(): a() b() xn_a, xn_b = pipeline.get_nodes_by_tag("twinkle") ``` -------------------------------- ### Executing Subgraphs with target_nodes Source: https://mindee.github.io/tawazi Demonstrates running a subset of the pipeline by specifying target nodes. ```python pipe_exec = pipeline.executor(target_nodes=[b]) pipe_exec() ``` -------------------------------- ### Handling MyPy typing with kwargs Source: https://mindee.github.io/tawazi Demonstrates how to use **kwargs as a workaround for MyPy typing limitations with Tawazi-specific parameters like twz_tag. ```python @xn def f(x: int): return x f(2) # works try: # f() got an unexpected keyword argument 'twz_tag' f(2, twz_tag="twinkle") # fails (mypy error) except TypeError: ... @xn def f_with_kwargs(x: int, **kwargs): return x f_with_kwargs(2, twz_tag="toes") # works ``` -------------------------------- ### Basic Arithmetic Operations in DAGs Source: https://mindee.github.io/tawazi Demonstrates applying arithmetic operations directly to ExecNode results within a DAG. ```python @xn def gen_data(x): return (x + 1) ** 2 @xn(debug=True) def my_print(x): print(f"x={x}") @dag def pipe(x, y): x,y = gen_data(x), gen_data(y) # notice the +,-,* operations are applied directly to ExecNode's results return -x, -y, x+y, x*y assert pipe(1, 2) == (-4, -9, 13, 36) ``` -------------------------------- ### compose(qualname, inputs, outputs, is_async=None, **kwargs) Source: https://mindee.github.io/tawazi/dag Composes a new DAG using the provided inputs and outputs ExecNodes. The inputs and outputs are specified as Alias objects or sequences of Alias objects, and the method returns a new DAG instance. This is an experimental feature. ```APIDOC ## compose(qualname, inputs, outputs, is_async=None, **kwargs) ### Description Compose a new DAG using inputs and outputs ExecNodes (Experimental). All provided Alias objects must point to unique ExecNodes, otherwise a ValueError is raised. The user is responsible for correctly specifying the inputs and outputs signature of the DAG. ### Method compose ### Parameters #### Path Parameters None #### Query Parameters None #### Request Body - **qualname** (str) - Required - The name of the composed DAG. - **inputs** ((ellipsis, Alias | List[Alias])) - Required - The Inputs nodes whose results are provided. Provide ... to specify that you will provide every argument of the original DAG. - **outputs** (Alias | List[Alias]) - Required - The Output nodes that must execute last, the ones that will generate results. - **is_async** (bool | None) - Optional - If True, the composed DAG will be an AsyncDAG; if False, it will be a DAG. Defaults to whatever the original DAG is. - **kwargs** (Dict[str, Any]) - Optional - Additional arguments to be passed to the DAG's constructor. ### Request Example ```python from tawazi import dag, xn, DAG from typing import Tuple, Any @xn def unwanted_xn() -> int: return 42 @xn def x(v: Any) -> int: return int(v) @xn def y(v: Any) -> str: return str(v) @xn def z(x: int, y: str) -> float: return float(x) + float(y) @dag def pipe() -> Tuple[int, float, int]: a = unwanted_xn() res = z(x(1), y(1)) b = unwanted_xn() return a, res, b composed_dag = pipe.compose("twinkle", [x, y], z) ``` ### Response #### Success Response (200) - **composed_dag** (DAG) - The composed DAG instance. #### Response Example ```python assert composed_dag(1, 1) == 2.0 ``` ### Error Handling - ValueError is raised if any provided Alias does not point to a unique ExecNode. ``` -------------------------------- ### DAGExecution.__call__ Source: https://mindee.github.io/tawazi/DAGExecution Synchronously calls the DAG with the provided arguments and returns the DAG's return value. Raises TawaziUsageError if the DAGExecution has already been executed. ```APIDOC ## DAGExecution.__call__ ### Description Call the DAG synchronously. Passes the given positional and keyword arguments to the DAG and returns the DAG's return value. ### Method __call__ ### Parameters - **\*args** (args) - Required - Positional arguments to pass in to the DAG. - **\*\*kwargs** (kwargs) - Required - Keyword arguments to pass in to the DAG. ### Raises - **TawaziUsageError** - If the DAGExecution has already been executed. ### Returns - **RVDAG** - The return value of the DAG's execution. ``` -------------------------------- ### Parallel execution with DAG Source: https://mindee.github.io/tawazi Shows how to use max_concurrency in @dag to execute independent ExecNodes in parallel. ```python from time import sleep, time @xn def a(): print("Function 'a' is running", flush=True) sleep(1) return "A" @xn def b(): print("Function 'b' is running", flush=True) sleep(1) return "B" @xn def c(a, b): print("Function 'c' is running", flush=True) print(f"Function 'c' received {a} from 'a' & {b} from 'b'", flush=True) return f"{a} + {b} = C" @dag(max_concurrency=2) def pipeline(): res_a = a() res_b = b() res_c = c(res_a, res_b) return res_c t0 = time() # executing the dag takes a single line of code res = pipeline() execution_time = time() - t0 assert execution_time < 1.5 # a() and b() are executed in parallel print(f"Graph execution took {execution_time:.2f} seconds") print(f"res = {res}") ## Graph execution took 1.00 seconds ## 'A + B = C' ``` -------------------------------- ### Debug ExecNode Configuration Source: https://mindee.github.io/tawazi Demonstrates how to define an ExecNode that only executes when the RUN_DEBUG_NODES environment variable is set to True. ```python @xn def a(i): return i + 1 @xn(debug=True) def print_debug(i): global debug_has_run debug_has_run = True print("DEBUG: ", i) @dag def pipe(): i = a(0) print_debug(i) return i debug_has_run = False pipe() assert debug_has_run == False ``` -------------------------------- ### DAG.__call__(*args, **kwargs) Source: https://mindee.github.io/tawazi/dag Executes the DAG scheduler using a similar interface to the function that describes the dependencies. Note that kwargs are not supported and will raise a TawaziUsageError. ```APIDOC ## DAG.__call__(*args, **kwargs) ### Description Execute the DAG scheduler via a similar interface to the function that describes the dependencies. Note: Currently kwargs are not supported. ### Method __call__ ### Parameters - *args (args) - Required - arguments to be passed to the call of the DAG - **kwargs (kwargs) - Optional - keyword arguments to be passed to the call of the DAG ### Returns - RVDAG - return value of the DAG's execution ### Raises - TawaziUsageError - if kwargs are passed ``` -------------------------------- ### Compose a sub-DAG using DAG.compose Source: https://mindee.github.io/tawazi Demonstrates how to reuse a portion of a DAG by defining inputs and outputs using tags or ExecNode references. ```python @xn def add(x, y): return x + y @xn def mul(x, y): return x * y @dag def pipe(x, y, z, w): v1 = add(1, x, twz_tag="add_v1") v2 = add(v1, y) v3 = add(v2, z, twz_tag="add_v3") v4 = mul(v3, w) return v4 assert pipe(2,3,4,5) == 50 # declare a sub-dag that only depends on v1, y, z and produces v3 sub_dag = pipe.compose(qualname="my_composed_dag", inputs=["add_v1", "pipe>!>y", "pipe>!>z"], outputs="add_v3") assert sub_dag(2,3,4) == 9 # notice that for inputs, we provide the return value of the ExecNode (return value of ExecNode tagged "add_v1") # but for the outputs, we indicate the the ExecNode whose return value must return. ``` -------------------------------- ### Mixing Alias Types for Target Nodes Source: https://mindee.github.io/tawazi Demonstrates mixing different alias types (names, tags, references) when defining target nodes. ```python pipe_exec = pipeline.executor(target_nodes=["b", xns_bye[0]]) pipe_exec() ``` -------------------------------- ### AsyncDAG.__call__(*args, **kwargs) Source: https://mindee.github.io/tawazi/dag Executes the DAG scheduler via a similar interface to the function that describes the dependencies. Currently kwargs are not supported. ```APIDOC ## AsyncDAG.__call__(*args, **kwargs) ### Description Executes the DAG scheduler via a similar interface to the function that describes the dependencies. Currently kwargs are not supported. ### Method async call ### Parameters #### Arguments - *args (args) - Required - arguments to be passed to the call of the DAG - **kwargs (kwargs) - Optional - keyword arguments to be passed to the call of the DAG ### Returns - RVDAG - return value of the DAG's execution ### Raises - TawaziUsageError - if kwargs are passed ``` -------------------------------- ### AsyncDAG.executor(target_nodes=None, exclude_nodes=None, root_nodes=None, cache_deps_of=None, cache_in='', from_cache='') Source: https://mindee.github.io/tawazi/dag Generates an AsyncDAGExecution for the current AsyncDAG. Allows specifying which nodes to execute, exclude, or use as roots, as well as caching options. ```APIDOC ## AsyncDAG.executor(target_nodes=None, exclude_nodes=None, root_nodes=None, cache_deps_of=None, cache_in='', from_cache='') ### Description Generates an AsyncDAGExecution for the current AsyncDAG. Allows specifying which nodes to execute, exclude, or use as roots, as well as caching options. ### Method executor ### Parameters #### Arguments - target_nodes (Optional[Sequence[Alias]]) - Optional - the nodes to execute, excluding all nodes that can be excluded - exclude_nodes (Optional[Sequence[Alias]]) - Optional - the nodes to exclude from the execution - root_nodes (Optional[Sequence[Alias]]) - Optional - these nodes and their children will be included in the execution - cache_deps_of (Optional[Sequence[Alias]]) - Optional - which nodes to cache the dependencies of - cache_in (str) - Optional - the path to the file where to cache - from_cache (str) - Optional - the cache ### Returns - AsyncDAGExecution[P, RVDAG] - the DAGExecution object associated with the dag ``` -------------------------------- ### Control ExecNode resource usage Source: https://mindee.github.io/tawazi Configures where an ExecNode runs by setting the resource parameter to main-thread, thread, or async-thread. ```python from tawazi import Resource import threading @xn(resource=Resource.main_thread) def run_in_main_thread(main_thread_id): assert main_thread_id == threading.get_ident() print(f"I am running in the main thread with thread id {threading.get_ident()}") @xn(resource=Resource.thread) def run_in_thread(main_thread_id): assert main_thread_id != threading.get_ident() print(f"I am running in a thread with thread id {threading.get_ident()}") @xn(resource=Resource.async_thread) def run_in_async_thread(main_thread_id): assert main_thread_id != threading.get_ident() print(f"I am running in an async thread with thread id {threading.get_ident()}") @dag def dag_with_resource(main_thread_id): run_in_main_thread(main_thread_id) run_in_thread(main_thread_id) run_in_async_thread(main_thread_id) dag_with_resource(threading.get_ident()) ``` -------------------------------- ### Debugging DAG pipelines Source: https://mindee.github.io/tawazi Demonstrates debugging a pipeline by commenting out the @dag decorator to execute the function as standard Python code. ```python @dag def pipeline(x): x_lo = decr(x) x_hi = incr(x) display(x_hi) display(x_lo) return x assert pipeline(10) == 10 #@dag # comment out the decorator def pipeline(x): x_lo = decr(x) x_hi = incr(x) # put breakpoint here and debug! display(x_hi) display(x_lo) return x assert pipeline(10) == 10 ``` -------------------------------- ### BaseDAG.config_from_dict(config) Source: https://mindee.github.io/tawazi/dag Reconfigures the parameters of the nodes from a dictionary. Raises ValueError if two nodes are configured by the provided config. ```APIDOC ## BaseDAG.config_from_dict(config) ### Description Reconfigures the parameters of the nodes from a dictionary. ### Method config_from_dict ### Parameters - config (Dict[str, Any]) - Required - The dictionary containing the config, e.g., {"nodes": {"a": {"priority": 3, "is_sequential": True}}, "max_concurrency": 3} ### Raises - ValueError - if two nodes are configured by the provided config (which is ambiguous) ``` -------------------------------- ### BaseDAG.config_from_json(config_path) Source: https://mindee.github.io/tawazi/dag Reconfigures the parameters of the nodes from a JSON file. ```APIDOC ## BaseDAG.config_from_json(config_path) ### Description Reconfigures the parameters of the nodes from a JSON file. ### Method config_from_json ### Parameters - config_path (str) - Required - The path to the JSON file ``` -------------------------------- ### Execute a sub-DAG within a main DAG Source: https://mindee.github.io/tawazi Shows how to nest a DAG within another for logical code separation. ```python @xn def add(x, y): return x + y @dag def sub_dag(x): return add(x, 1) @dag def main_dag(x): return sub_dag(x) assert main_dag(2) == 3 ``` -------------------------------- ### AsyncDAGExecution.__call__ Source: https://mindee.github.io/tawazi/DAGExecution Asynchronously calls the DAG with the provided arguments and returns the DAG's return value. Raises TawaziUsageError if the DAGExecution has already been executed. ```APIDOC ## AsyncDAGExecution.__call__ ### Description Call the DAG asynchronously. Passes the given positional and keyword arguments to the DAG and returns the DAG's return value. ### Method __call__ (async) ### Parameters - **\*args** (args) - Required - Positional arguments to pass in to the DAG. - **\*\*kwargs** (kwargs) - Required - Keyword arguments to pass in to the DAG. ### Raises - **TawaziUsageError** - If the DAGExecution has already been executed. ### Returns - **RVDAG** - The return value of the DAG's execution. ``` -------------------------------- ### Indexing return values in ExecNode Source: https://mindee.github.io/tawazi ExecNodes can return Dict or List types, which can be accessed via indexing within the pipeline. ```python @xn def gen_dict(val): return {"k1": val, "k2": "2", "nested_list": [1 ,11, 3]} @xn def gen_list(val): return [val, val + 1, val + 2, val + 3] @xn def incr(val): return val + 1 @dag def pipeline(val): d = gen_dict(val) l = gen_list(d["k1"]) # indexing a dict inc_val = incr(l[0]) # indexing a list / tuple inc_val_2 = incr(d["nested_list"][1]) # indexing a list / tuple return d, l, inc_val, inc_val_2 d, l, inc_val, inc_val_2 = pipeline(123) assert d == {"k1": 123, "k2": "2", "nested_list": [1 ,11, 3]} assert l == [123, 124, 125, 126] assert inc_val == 124 assert inc_val_2 == 12 ``` -------------------------------- ### Unpacking multiple return values in ExecNode Source: https://mindee.github.io/tawazi For Tuple and List types, specify the unpacking number using the unpack_to parameter. ```python @xn(unpack_to=4) def replicate_tuple(val): return (val, val + 1, val + 2, val + 3) @xn(unpack_to=4) def replicate_list(val): return [val, val + 1, val + 2, val + 3] @dag def pipeline(): v1, v2, v3, v4 = replicate_tuple(1) # notice unpacked values here v5, v6, v7, v8 = replicate_list(v4) # and here return v1, v2, v3, v4, v5, v6, v7, v8 assert pipeline() == (1, 2, 3, 4, 4, 5, 6, 7) ``` -------------------------------- ### Conditional Execution with twz_active Source: https://mindee.github.io/tawazi Uses the twz_active parameter to conditionally execute nodes, recommending the use of and_, or_, and not_ nodes for boolean logic. ```python from tawazi import and_ @xn def f1(x): return x ** 2 + 1 @xn def f2(x): return x ** 2 - 1 @xn def f3(x): return x ** 3 @xn def wrap_dict(x): return {"value": x} @dag def pipe(x): v1 = f1(x, twz_active=x > 0) # equivalent to if x > 0: v1 = f1(x) v2 = f2(x, twz_active=x < 0) # equivalent to if x < 0: v2 = f2(x) # you can also write `v3 = f3(x, twz_active=(x > 1) & (x > 0))` v3 = f3(x, twz_active=and_(x > 1, x > 0)) # When twz_active is False, the ExecNode is not executed and returns None return v1, v2, v3 assert pipe(-1) == (None, 0, None) assert pipe(2) == (5, None, 8) assert pipe(0) == (None, None, None) ``` -------------------------------- ### BaseDAG.config_from_yaml(config_path) Source: https://mindee.github.io/tawazi/dag Reconfigures the parameters of the nodes from a YAML file. ```APIDOC ## BaseDAG.config_from_yaml(config_path) ### Description Reconfigures the parameters of the nodes from a YAML file. ### Method config_from_yaml ### Parameters - config_path (str) - Required - The path to the YAML file ``` -------------------------------- ### Run an AsyncDAG Source: https://mindee.github.io/tawazi Executes a DAG in an async context, requiring the async-thread resource to avoid blocking the event loop. ```python # will be awaited in the AsyncDAG's scheduler. @xn(resource=Resource.async_thread) def add_async(x, y): return x + y @dag(is_async=True) def my_async_dag(x): return add(x, 1) import asyncio import numpy as np # using numpy in the example to show that the event loop is not blocked! # because numpy releases the GIL. res = asyncio.run(my_async_dag(np.zeros(1000))) assert (res == np.ones(1000)).all() ``` -------------------------------- ### Controlling Parallel Execution with Priority and is_sequential Source: https://mindee.github.io/tawazi Configures node execution order using priority and ensures thread-safety by setting is_sequential to True. ```python from time import sleep, time from tawazi import xn, dag @xn def a(): print("Function 'a' is running", flush=True) sleep(1) return "A" # optionally configure each ExecNode using the decorator: # is_sequential = True to prevent ExecNode from running in parallel with other ExecNodes # priority to choose the ExecNode in the next execution phase @xn(is_sequential=True, priority=10) def b(): print("Function 'b' is running", flush=True) sleep(1) return "B" @xn def c(a, arg_b): print("Function 'c' is running", flush=True) print(f"Function 'c' received {a} from 'a' & {arg_b} from 'b'", flush=True) return f"{a} + {arg_b} = C" # optionally customize the DAG @dag(max_concurrency=2) def deps_describer(): res_a = a() res_b = b() res_c = c(res_a, res_b) return res_a, res_b, res_c t0 = time() # the dag instance is reusable. # This is recommended if you want to do the same computation multiple times res_a, res_b, res_c = deps_describer() execution_time = time() - t0 print(f"Graph execution took {execution_time:.2f} seconds") assert res_a == "A" assert res_b == "B" assert res_c == "A + B = C" ``` -------------------------------- ### Configuring ExecNode behavior outside DAGs Source: https://mindee.github.io/tawazi Shows how to set TAWAZI_EXECNODE_OUTSIDE_DAG_BEHAVIOR to warning to allow calling ExecNodes outside of a DAG for debugging purposes. ```python # set environment variable to warning from tawazi import cfg cfg.TAWAZI_EXECNODE_OUTSIDE_DAG_BEHAVIOR = "warning" display('Hello World!') # :1: UserWarning: Invoking LazyExecNode display ~ | <0x7fdc03d4ebb0> outside of a `DAG`. Executing wrapped function instead of describing dependency. # prints Hello World! ``` -------------------------------- ### AsyncDAG.__call__ Source: https://mindee.github.io/tawazi/dag Executes the DAG asynchronously. This method is part of the AsyncDAG class. ```APIDOC ## AsyncDAG.__call__ ### Description Executes the DAG asynchronously. This method is part of the AsyncDAG class. ```