flowno.decorators.node

Node decorator for Flowno.

This module provides the @node decorator, which transforms async functions or classes into DraftNode subclasses. These nodes can then be used within a FlowHDL context to define dataflow graphs.

Examples

Basic usage:

>>> from flowno import node
>>>
>>> @node
... async def Add(x: int, y: int) -> int:
...     return x + y
>>>
>>> add_node = Add(1, 2)
>>> print(add_node)  # DraftNode instance

With stream inputs:

>>> from flowno import node, Stream
>>>
>>> @node(stream_in=["a"])
... async def SumStream(x: int, a: Stream[int]) -> int:
...     total = x
...     async for value in a:
...         total += value
...     return total
>>>
>>> sum_stream_node = SumStream(1)
>>> print(sum_stream_node)  # DraftNode instance with stream input

With multiple outputs:

>>> from flowno import node
>>>
>>> @node(multiple_outputs=True)
... async def SumAndDiff(x: int, y: int) -> tuple[int, int]:
...     return x + y, x - y
>>>
>>> sum_and_diff_node = SumAndDiff(3, 1)
>>> print(sum_and_diff_node)  # DraftNode instance with multiple outputs

flowno.decorators.node.node(func_or_cls: Callable[[T1], Coroutine[Any, Any, _ReturnT_co]], /) → type[MonoNode1[T1, tuple[_ReturnT_co]]][source]

flowno.decorators.node.node(func_or_cls: Callable[[], Coroutine[Any, Any, None]], /) → type[MonoNode0_0]

flowno.decorators.node.node(func_or_cls: Callable[[], Coroutine[Any, Any, _ReturnT_co]], /) → type[MonoNode0_1[_ReturnT_co]]

flowno.decorators.node.node(func_or_cls: None = None, /, *, multiple_outputs: Literal[False] | None = None, stream_in: list[str] = EMPTY_LIST) → node_meta_single_dec

Decorator that transforms async functions or classes into DraftNode subclasses.

Parameters:

func_or_cls – The async function or class to transform
multiple_outputs – Whether the node has multiple outputs
stream_in – List of input streams

Returns:

A DraftNode subclass or a node_meta decorator

Examples

Basic usage:

>>> from flowno import node
>>>
>>> @node
... async def Add(x: int, y: int) -> int:
...     return x + y
>>>
>>> add_node = Add(1, 2)
>>> print(add_node)  # DraftNode instance

With stream inputs:

>>> from flowno import node, Stream
>>>
>>> @node(stream_in=["a"])
... async def SumStream(x: int, a: Stream[int]) -> int:
...     total = x
...     async for value in a:
...         total += value
...     return total
>>>
>>> sum_stream_node = SumStream(1)
>>> print(sum_stream_node)  # DraftNode instance with stream input

With multiple outputs:

>>> from flowno import node
>>>
>>> @node(multiple_outputs=True)
... async def SumAndDiff(x: int, y: int) -> tuple[int, int]:
...     return x + y, x - y
>>>
>>> sum_and_diff_node = SumAndDiff(3, 1)
>>> print(sum_and_diff_node)  # DraftNode instance with multiple outputs