context

Classes:

Name	Description
DAGContext	A context for a directed acyclic graph (DAG).

DAGContext dataclass

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

A context for a directed acyclic graph (DAG).

Methods:

Name	Description
build	create a DAG context based on the given node
get_outgoing_nodes	Get all outgoing nodes of the stream.
get_node_label	Get the label of the node.
get_outgoing_streams	Extract all node's outgoing streams from the given set of streams, Because a node only know its incoming streams.
render	Render the object to a string.

Attributes:

Name	Type	Description
node	Node	The root node (the destination) of the DAG.
nodes	tuple[Node, ...]	All nodes in the graph.
streams	tuple[Stream, ...]	All streams in the graph.
all_nodes	list[Node]	All nodes in the graph sorted by the number of upstream nodes.
all_streams	list[Stream]	All streams in the graph sorted by the number of upstream nodes and the index of the stream.
outgoing_nodes	dict[Stream, list[tuple[Node, int]]]	A dictionary of outgoing nodes for each stream.
outgoing_streams	dict[Node, list[Stream]]	A dictionary of outgoing streams for each node.
node_labels	dict[Node, str]	A dictionary of outgoing streams for each node.

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.