Tawazi

https://mindee.github.io/tawazi
Admin
Tawazi is a Python library for executing Directed Acyclic Graph (DAG) pipelines concurrently,...
Tokens:18,543
Snippets:133
Trust Score:9.7
Update:1 month ago
Show doc for...
Tokens:
Raw
### 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))
## <class 'tawazi.node.node.LazyExecNode'>

@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))
## <class 'tawazi._dag.dag.DAG'>

# `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!')
# <stdin>: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.
```
Tokens:
Raw