ewokscore.graph.taskgraph.TaskGraph#

class ewokscore.graph.taskgraph.TaskGraph(source=None, representation=None, root_dir=None, root_module=None)[source]#

Bases: object

The API for graph analysis is provided by networkx. Any directed graph is supported (cyclic or acyclic).

Loop over the dependencies of a task

for source in taskgraph.predecessors(target):
    link_attrs = taskgraph.graph[source][target]

Loop over the tasks dependent on a task

for target in taskgraph.successors(source):
    link_attrs = taskgraph.graph[source][target]

For acyclic graphs, sequential task execution can be done like this:

taskgraph.execute()
Parameters:

  • source (Union[str, Path, Mapping, Graph, None])

  • representation (Union[GraphRepresentation, str, None])

  • root_dir (Union[str, Path, None])

  • root_module (Optional[str])

dump(destination=None, representation=None, **save_options)[source]#
Parameters:
Return type:

Union[str, Path, dict, None]

execute(*args, **kw)[source]#
property graph_id: Hashable#
property graph_label: str#
property is_cyclic: bool#
load(source=None, representation=None, subgraph_representation=None, root_dir=None, root_module=None)[source]#
Parameters:

  • source (Union[str, Path, Mapping, Graph, None])

  • representation (Union[GraphRepresentation, str, None])

  • subgraph_representation (Union[GraphRepresentation, str, None])

  • root_dir (Union[str, Path, None])

  • root_module (Optional[str])

Return type:

None

property requirements: Sequence[str] | None#
serialize()[source]#
Return type:

str