Skip to content

Nodes

Node

Bases: ABC

Represents a node in the graph.

The generic types define the type of state and shared state that the node expects. Due to variance or the usage of protocols, the node can also take any subtype of the state and shared state.

The node must implement the __call__ method to run the node.

Source code in src/edgygraph/nodes.py 
 10
 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
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
class Node[T: StateProtocol = StateProtocol, S: SharedProtocol = SharedProtocol](ABC):
    """
    Represents a node in the graph.

    The generic types define the type of state and shared state that the node expects.
    Due to variance or the usage of protocols, the node can also take any subtype of the state and shared state.

    The node must implement the `__call__` method to run the node.
    """

    dependencies: set[str] = set()
    """
    The pip dependencies of the node (python packages). On initialization of the node a check is performed if the dependencies are installed with importlib. If not an error is raised.

    The dependencies are collected from all parent classes. That means if a parent class has dependencies, the child class will also have those, but the child class can add more by setting the `dependencies` attribute without the need to repeat the parent classes dependencies.
    """

    @classmethod
    def check_dependencies(cls) -> None:
        """
        Checks if the dependencies of the node are installed.

        Raises:
            ImportError: If a dependency is not installed.
        """
        for dependency in cls.dependencies:
            try:
                metadata.version(dependency)
            except metadata.PackageNotFoundError:
                raise ImportError(f"Dependency {dependency} not found but required by node {cls.__name__}. Please install it with pip install {dependency}")

    def __init_subclass__(cls) -> None:
        """
        Called when a subclass is created. This is used to collect the dependencies of the node from parent classes.

        Works recursively because it is called on each inheritance level. Therefore the dependencies of all parent classes are collected and stored in the class variable `dependencies` of the child class.
        """
        super().__init_subclass__()

        cls.dependencies = cls.dependencies.copy() # To not modify the class variable of the parent

        for base in cls.__bases__: # Called on each inheritance level, therefore its recursive
            if not issubclass(base, Node):
                continue
            cls.dependencies.update(base.dependencies)

    def __init__(self) -> None:
        self.check_dependencies()


    @abstractmethod
    async def __call__(self, state: T, shared: S) -> None:
        """
        Runs the node with the given state and shared state from the graph.

        Operations on the state are merged in the graphs state after the node has finished.
        Therefore, the state is not shared between nodes and operations are safe.

        Operations on the shared state are reflected in all parallel running nodes.
        Therefore, the shared state is shared between nodes and operations are not safe without using the Lock.
        The lock can be accessed via `shared.lock`.

        Args:
            state: The state of the graph.
            shared: The shared state of the graph.

        Returns:
            None. The instance references of the arguments are used in the graph to enable variance.
        """
        pass


    def copy(self) -> Node[T, S]:
        """
        Creates a copy of the node. This can be used when the same node is used multiple times in the graph.

        Returns:
            A copy of the node.
        """
        return copy(self)


    def __neg__(self) -> tuple[Node[T, S], NodeConfig]:
        """
        Negates the node. This can be used to indicate that the node should only be used as a next and not as a source.
        """
        return (self, NodeConfig(operator="neg"))

    def __pos__(self) -> tuple[Node[T, S], NodeConfig]:
        """
        Positivizes the node. This can be used to indicate that the node should only be used as a source and not as a next.
        """
        return (self, NodeConfig(operator="pos"))

dependencies = set() class-attribute instance-attribute

The pip dependencies of the node (python packages). On initialization of the node a check is performed if the dependencies are installed with importlib. If not an error is raised.

The dependencies are collected from all parent classes. That means if a parent class has dependencies, the child class will also have those, but the child class can add more by setting the dependencies attribute without the need to repeat the parent classes dependencies.

check_dependencies() classmethod

Checks if the dependencies of the node are installed.

Raises:

Type Description
ImportError

If a dependency is not installed.
Source code in src/edgygraph/nodes.py 
27
28
29
30
31
32
33
34
35
36
37
38
39
@classmethod
def check_dependencies(cls) -> None:
    """
    Checks if the dependencies of the node are installed.

    Raises:
        ImportError: If a dependency is not installed.
    """
    for dependency in cls.dependencies:
        try:
            metadata.version(dependency)
        except metadata.PackageNotFoundError:
            raise ImportError(f"Dependency {dependency} not found but required by node {cls.__name__}. Please install it with pip install {dependency}")

__init_subclass__()

Called when a subclass is created. This is used to collect the dependencies of the node from parent classes.

Works recursively because it is called on each inheritance level. Therefore the dependencies of all parent classes are collected and stored in the class variable dependencies of the child class.

Source code in src/edgygraph/nodes.py 
41
42
43
44
45
46
47
48
49
50
51
52
53
54
def __init_subclass__(cls) -> None:
    """
    Called when a subclass is created. This is used to collect the dependencies of the node from parent classes.

    Works recursively because it is called on each inheritance level. Therefore the dependencies of all parent classes are collected and stored in the class variable `dependencies` of the child class.
    """
    super().__init_subclass__()

    cls.dependencies = cls.dependencies.copy() # To not modify the class variable of the parent

    for base in cls.__bases__: # Called on each inheritance level, therefore its recursive
        if not issubclass(base, Node):
            continue
        cls.dependencies.update(base.dependencies)

__call__(state, shared) abstractmethod async

Runs the node with the given state and shared state from the graph.

Operations on the state are merged in the graphs state after the node has finished. Therefore, the state is not shared between nodes and operations are safe.

Operations on the shared state are reflected in all parallel running nodes. Therefore, the shared state is shared between nodes and operations are not safe without using the Lock. The lock can be accessed via shared.lock.

Parameters:

Name Type Description Default
state T

The state of the graph.

 required
shared S

The shared state of the graph.

 required

Returns:

Type Description
None

None. The instance references of the arguments are used in the graph to enable variance.
Source code in src/edgygraph/nodes.py 
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
@abstractmethod
async def __call__(self, state: T, shared: S) -> None:
    """
    Runs the node with the given state and shared state from the graph.

    Operations on the state are merged in the graphs state after the node has finished.
    Therefore, the state is not shared between nodes and operations are safe.

    Operations on the shared state are reflected in all parallel running nodes.
    Therefore, the shared state is shared between nodes and operations are not safe without using the Lock.
    The lock can be accessed via `shared.lock`.

    Args:
        state: The state of the graph.
        shared: The shared state of the graph.

    Returns:
        None. The instance references of the arguments are used in the graph to enable variance.
    """
    pass

copy()

Creates a copy of the node. This can be used when the same node is used multiple times in the graph.

Returns:

Type Description
Node[T, S]

A copy of the node.
Source code in src/edgygraph/nodes.py 
82
83
84
85
86
87
88
89
def copy(self) -> Node[T, S]:
    """
    Creates a copy of the node. This can be used when the same node is used multiple times in the graph.

    Returns:
        A copy of the node.
    """
    return copy(self)

__neg__()

Negates the node. This can be used to indicate that the node should only be used as a next and not as a source.

Source code in src/edgygraph/nodes.py 
92
93
94
95
96
def __neg__(self) -> tuple[Node[T, S], NodeConfig]:
    """
    Negates the node. This can be used to indicate that the node should only be used as a next and not as a source.
    """
    return (self, NodeConfig(operator="neg"))

__pos__()

Positivizes the node. This can be used to indicate that the node should only be used as a source and not as a next.

Source code in src/edgygraph/nodes.py 
 98
 99
100
101
102
def __pos__(self) -> tuple[Node[T, S], NodeConfig]:
    """
    Positivizes the node. This can be used to indicate that the node should only be used as a source and not as a next.
    """
    return (self, NodeConfig(operator="pos"))

START

Represents a start node

Source code in src/edgygraph/nodes.py 
104
105
106
107
108
class START:
    """
    Represents a start node
    """
    pass

END

Represents an end node

Source code in src/edgygraph/nodes.py 
110
111
112
113
114
class END:
    """
    Represents an end node
    """
    pass

NodeConfig

Bases: BaseModel

Represents the configuration of a node.

Source code in src/edgygraph/nodes.py 
117
118
119
120
121
122
123
124
125
126
127
128
129
130
class NodeConfig(BaseModel):
    """
    Represents the configuration of a node.
    """

    operator: Literal["pos", "neg", None] = None

    @property
    def only_next(self) -> bool:
        return self.operator == "neg"

    @property
    def only_source(self) -> bool:
        return self.operator == "pos"