Skip to content

Base Node

graphorchestrator.nodes.base

Node

Bases: ABC

Abstract base class representing a node in a graph.

Nodes have unique IDs and can have incoming and outgoing edges.

Source code in graphorchestrator\nodes\base.py 
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
class Node(ABC):
    """
    Abstract base class representing a node in a graph.

    Nodes have unique IDs and can have incoming and outgoing edges.
    """

    def __init__(self, node_id: str) -> None:
        """
        Initializes a new Node instance.

        Args:
            node_id (str): The unique identifier for this node.
        """
        self.node_id: str = node_id
        self.incoming_edges = []
        self.outgoing_edges = []
        self.fallback_node_id: Optional[str] = None
        self.retry_policy: Optional[RetryPolicy] = None

        GraphLogger.get().info(
            **wrap_constants(
                message="Node initialized",
                **{
                    LC.EVENT_TYPE: "node",
                    LC.NODE_ID: self.node_id,
                    LC.NODE_TYPE: self.__class__.__name__,
                    LC.ACTION: "node_created",
                    LC.CUSTOM: {
                        "incoming_edges": 0,
                        "outgoing_edges": 0,
                        "fallback_node_id": None,
                        "retry_policy": None,
                    },
                }
            )
        )

    @abstractmethod
    def execute(self, state: State):
        """
        Abstract method to execute the node's logic.

        Args:
            state: The current state of the execution.
        """
        raise NotImplementedError

    def set_fallback(self, fallback_node_id: str) -> None:
        """
        Sets the fallback node ID for this node.

        Args:
            fallback_node_id (str): The ID of the fallback node.
        """
        self.fallback_node_id = fallback_node_id

    def set_retry_policy(self, retry_policy: RetryPolicy) -> None:
        """
        Sets the retry policy for this node.

        Args:
            retry_policy (RetryPolicy): The retry policy to apply.
        """
        self.retry_policy = retry_policy

__init__(node_id)

Initializes a new Node instance.

Parameters:

Name Type Description Default
node_id str

The unique identifier for this node.

 required
Source code in graphorchestrator\nodes\base.py 
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
def __init__(self, node_id: str) -> None:
    """
    Initializes a new Node instance.

    Args:
        node_id (str): The unique identifier for this node.
    """
    self.node_id: str = node_id
    self.incoming_edges = []
    self.outgoing_edges = []
    self.fallback_node_id: Optional[str] = None
    self.retry_policy: Optional[RetryPolicy] = None

    GraphLogger.get().info(
        **wrap_constants(
            message="Node initialized",
            **{
                LC.EVENT_TYPE: "node",
                LC.NODE_ID: self.node_id,
                LC.NODE_TYPE: self.__class__.__name__,
                LC.ACTION: "node_created",
                LC.CUSTOM: {
                    "incoming_edges": 0,
                    "outgoing_edges": 0,
                    "fallback_node_id": None,
                    "retry_policy": None,
                },
            }
        )
    )

execute(state) abstractmethod

Abstract method to execute the node's logic.

Parameters:

Name Type Description Default
state State

The current state of the execution.

 required
Source code in graphorchestrator\nodes\base.py 
49
50
51
52
53
54
55
56
57
@abstractmethod
def execute(self, state: State):
    """
    Abstract method to execute the node's logic.

    Args:
        state: The current state of the execution.
    """
    raise NotImplementedError

set_fallback(fallback_node_id)

Sets the fallback node ID for this node.

Parameters:

Name Type Description Default
fallback_node_id str

The ID of the fallback node.

 required
Source code in graphorchestrator\nodes\base.py 
59
60
61
62
63
64
65
66
def set_fallback(self, fallback_node_id: str) -> None:
    """
    Sets the fallback node ID for this node.

    Args:
        fallback_node_id (str): The ID of the fallback node.
    """
    self.fallback_node_id = fallback_node_id

set_retry_policy(retry_policy)

Sets the retry policy for this node.

Parameters:

Name Type Description Default
retry_policy RetryPolicy

The retry policy to apply.

 required
Source code in graphorchestrator\nodes\base.py 
68
69
70
71
72
73
74
75
def set_retry_policy(self, retry_policy: RetryPolicy) -> None:
    """
    Sets the retry policy for this node.

    Args:
        retry_policy (RetryPolicy): The retry policy to apply.
    """
    self.retry_policy = retry_policy