fitz

import os # NOTE: This library is in an extremely early development stage (0.0.1.dev2) # and appears to be abandoned. The following example is derived from internal tests # and may not be functional or representative of a stable API. # It is NOT recommended for production use. from fitz.workflow import Workflow from fitz.node import Node from fitz.dag import DAG # Define simple nodes (tasks) class SimpleNode(Node): def __init__(self, name, func=None): super().__init__(name) self.func = func or (lambda: print(f"Executing {name}")) def execute(self): self.func() # Create nodes node_a = SimpleNode('A') node_b = SimpleNode('B') node_c = SimpleNode('C') # Create a workflow and add nodes workflow = Workflow('MyExampleWorkflow') workflow.add_node(node_a) workflow.add_node(node_b) workflow.add_node(node_c) # Define dependencies (A -> B, A -> C) workflow.add_edge(node_a, node_b) workflow.add_edge(node_a, node_c) print(f"Workflow '{workflow.name}' created with nodes: {[n.name for n in workflow.nodes]}") print(f"Edges: {workflow.dag.edges}") # Attempt to run the workflow (functionality is limited and untested for general use) # For a real run, you'd likely need more complex node logic and an executor. # The project's tests suggest running tasks, but no clear 'workflow.run()' method exists at this level. # This is purely illustrative of structure. print("\n--- Attempting to show node execution order (conceptual) ---") for node in workflow.dag.topological_sort(): print(f"Ready to execute: {node.name}") # node.execute() # Actual execution requires more setup than this quickstart provides