context

DAGContext dataclass

DAGContext(
    *,
    node: Node,
    nodes: tuple[Node, ...],
    streams: tuple[Stream, ...]
)

A context for a directed acyclic graph (DAG).

node instance-attribute

node: Node

The root node (the destination) of the DAG.

nodes instance-attribute

nodes: tuple[Node, ...]

All nodes in the graph.

streams instance-attribute

streams: tuple[Stream, ...]

All streams in the graph.

all_nodes cached property

all_nodes: list[Node]

All nodes in the graph sorted by the number of upstream nodes.

all_streams cached property

all_streams: list[Stream]

All streams in the graph sorted by the number of upstream nodes and the index of the stream.

outgoing_nodes cached property

outgoing_nodes: dict[Stream, list[tuple[Node, int]]]

A dictionary of outgoing nodes for each stream.

outgoing_streams cached property

outgoing_streams: dict[Node, list[Stream]]

A dictionary of outgoing streams for each node.

node_labels cached property

node_labels: dict[Node, str]

A dictionary of outgoing streams for each node.

build classmethod

build(node: Node) -> DAGContext

create a DAG context based on the given node

Parameters:

Name	Type	Description	Default
node	Node	The root node of the DAG.	required

Returns:

Type	Description
DAGContext	A DAG context based on the given node.

get_outgoing_nodes

get_outgoing_nodes(
    stream: Stream,
) -> list[tuple[Node, int]]

Get all outgoing nodes of the stream.

Parameters:

Name	Type	Description	Default
stream	Stream	The stream to get the outgoing nodes of.	required

Returns:

Type	Description
list[tuple[Node, int]]	The outgoing nodes of the stream.

get_node_label

get_node_label(node: Node) -> str

Get the label of the node.

Parameters:

Name	Type	Description	Default
node	Node	The node to get the label of.	required

Returns:

Type	Description
str	The label of the node.

get_outgoing_streams

get_outgoing_streams(node: Node) -> list[Stream]

Extract all node's outgoing streams from the given set of streams, Because a node only know its incoming streams.

Parameters:

Name	Type	Description	Default
node	Node	The node to get the outgoing streams of.	required

Returns:

Type	Description
list[Stream]	The outgoing streams of the node.

render

render(obj: Any) -> Any

Render the object to a string.

Parameters:

Name	Type	Description	Default
obj	Any	The object to render.	required

Returns:

Type	Description
Any	The rendered object.