Skip to content

States

State

Bases: BaseModel

Holds variables for the nodes of serializable types. Arbitrary types are not allowed.

The state is copied for each branch and merged on the join event. The state is also copied for each parallel node inside a branch and merged after each execution step. Therefore, the state is not shared between nodes and operations are safe.

Parallel changes of the same variable will be detected as a conflict and raise an error.

Implements pydantic's BaseModel.

Source code in src/edgygraph/states.py 
 8
 9
10
11
12
13
14
15
16
17
18
19
20
class State(BaseModel):
    """
    Holds variables for the nodes of serializable types. Arbitrary types are not allowed.

    The state is copied for each branch and merged on the join event.
    The state is also copied for each parallel node inside a branch and merged after each execution step.
    Therefore, the state is not shared between nodes and operations are safe.

    Parallel changes of the same variable will be detected as a conflict and raise an error.

    Implements pydantic's BaseModel.
    """
    model_config = ConfigDict(arbitrary_types_allowed=False) # for deep copy

Shared

Bases: BaseModel

Holds shared variables for the nodes of any type.

The shared state is shared between all branches and parallel nodes and operations are not safe without using the Lock. The lock can be accessed via shared.lock.

Implements pydantic's BaseModel with arbitrary types allowed.

Source code in src/edgygraph/states.py 
23
24
25
26
27
28
29
30
31
32
33
class Shared(BaseModel):
    """
    Holds shared variables for the nodes of any type.

    The shared state is shared between *all* branches and parallel nodes and operations are not safe without using the Lock.
    The lock can be accessed via `shared.lock`.

    Implements pydantic's BaseModel with arbitrary types allowed.
    """
    lock: Lock = Field(default_factory=Lock)
    model_config = ConfigDict(arbitrary_types_allowed=True)

PydanticModel

Bases: Protocol

Minimal Protocol for Pydantic BaseModel Operations

Source code in src/edgygraph/states.py 
36
37
38
39
40
41
42
43
44
45
@runtime_checkable
class PydanticModel(Protocol):
    """Minimal Protocol for Pydantic BaseModel Operations"""

    def model_dump(self) -> dict[str, Any]: ...

    def model_copy(self, *, update: Mapping[str, Any] | None = None, deep: bool = False) -> Self: ...

    @classmethod
    def model_validate(cls, obj: Any) -> Self: ...

StateProtocol

Bases: PydanticModel, Protocol

Protocol of State.

The protocol is used to define the type of the state that the node expects. The usage of protocols allows a more flexible approach to generic typing.

Source code in src/edgygraph/states.py 
48
49
50
51
52
53
54
55
56
@runtime_checkable
class StateProtocol(PydanticModel, Protocol):
    """
    Protocol of State.

    The protocol is used to define the type of the state that the node expects.
    The usage of protocols allows a more flexible approach to generic typing.
    """
    pass

SharedProtocol

Bases: PydanticModel, Protocol

Protocol of Shared.

The protocol is used to define the type of the shared state that the node expects. The usage of protocols allows a more flexible approach to generic typing.

Source code in src/edgygraph/states.py 
59
60
61
62
63
64
65
66
67
@runtime_checkable
class SharedProtocol(PydanticModel, Protocol):
    """
    Protocol of Shared.

    The protocol is used to define the type of the shared state that the node expects.
    The usage of protocols allows a more flexible approach to generic typing.
    """
    lock: Lock

StateAttribute

Bases: BaseModel

Holds variables of serializable types for the nodes. Arbitrary types are not allowed.

Simplifies composition in states.

Source code in src/edgygraph/states.py 
70
71
72
73
74
75
76
class StateAttribute(BaseModel):
    """
    Holds variables of serializable types for the nodes. Arbitrary types are not allowed.

    Simplifies composition in states.
    """
    model_config = ConfigDict(arbitrary_types_allowed=False) # for deep copy

SharedAttribute

Bases: BaseModel

Holds shared variables of any type for the nodes.

Simplifies composition in shared states.

Source code in src/edgygraph/states.py 
78
79
80
81
82
83
84
class SharedAttribute(BaseModel):
    """
    Holds shared variables of any type for the nodes.

    Simplifies composition in shared states.
    """
    model_config = ConfigDict(arbitrary_types_allowed=True)

Stream

Bases: ABC, AsyncIterator[T]

Standardized wrapper interface for streams of data.

Set the type of the data that the stream will yield with the generic type parameter T.

Implements an async iterator and async context manager.

Source code in src/edgygraph/states.py 
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
class Stream[T: object](ABC, AsyncIterator[T]):
    """
    Standardized wrapper interface for streams of data.

    Set the type of the data that the stream will yield with the generic type parameter `T`.

    Implements an async iterator and async context manager.
    """

    @abstractmethod
    async def aclose(self) -> None:
        pass

    @abstractmethod
    async def __anext__(self) -> T:
        pass

    async def __aenter__(self) -> "Stream[T]":
        return self

    async def __aexit__(
            self, exc_type: type[BaseException] | None, 
            exc: BaseException | None, 
            tb: TracebackType | None
        ) -> None: # Not handling exceptions here -> returns None

        await self.aclose()