Skip to content

Types

Types

Typeguards for runtime typechecking.

Source code in src/edgygraph/graph/types.py 
 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
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
class Types[T: StateProtocol, S: SharedProtocol]:
    """
    Typeguards for runtime typechecking.
    """

    @classmethod
    def is_node_with_config(cls, x: Any) -> TypeGuard[NodeWithConfig[T, S]]:
        return (
            isinstance(x, tuple) and
            len(cast(tuple[Any], x)) == 2 and
            isinstance(x[0], Node) and
            isinstance(x[1], NodeConfig)
        )


    @classmethod
    def is_single_next(cls, x: Any) -> TypeGuard[SingleNext[T, S]]:
        return (
            x is None or
            isinstance(x, Node)
        )

    @classmethod
    def is_single_next_with_config(cls, x: Any) -> TypeGuard[SingleNextWithConfig[T, S]]:
        return (
            cls.is_single_next(x) or
            cls.is_node_with_config(x)
        )

    @classmethod
    def is_single_next_list(cls, x: Any) -> TypeGuard[list[SingleNext[T, S]]]:
        return isinstance(x, list) and all(cls.is_single_next(n) for n in cast(list[Any], x))

    @classmethod
    def is_single_next_with_config_list(cls, x: Any) -> TypeGuard[list[SingleNextWithConfig[T, S]]]:
        return isinstance(x, list) and all(cls.is_single_next_with_config(n) for n in cast(list[Any], x))

    @classmethod
    def is_resolved_next(cls, x: Any) -> TypeGuard[ResolvedNext[T, S]]:
        return (
            cls.is_single_next(x) or
            cls.is_single_next_list(x)
        )

    @classmethod
    def is_resolved_next_with_config(cls, x: Any) -> TypeGuard[ResolvedNextWithConfig[T, S]]:
        return (
            cls.is_single_next_with_config(x) or
            cls.is_single_next_with_config_list(x)
        )

    @classmethod
    def is_next_callable(cls, x: Any) -> TypeGuard[Callable[[T, S], ResolvedNext[T, S]] | Callable[[T, S], Awaitable[ResolvedNext[T, S]]]]:
        return callable(x) and not (
            cls.is_any_source(x) or # includes Node, START, Exceptions
            x is END
        )


    @classmethod
    def is_next(cls, x: Any) -> TypeGuard[Next[T, S]]:
        return (
            cls.is_resolved_next(x) or
            cls.is_next_callable(x)
        )

    @classmethod
    def is_next_with_config(cls, x: Any) -> TypeGuard[NextWithConfig[T, S]]:
        return (
            cls.is_resolved_next_with_config(x) or
            cls.is_next_callable(x)
        )

    @classmethod
    def is_single_source(cls, x: Any) -> TypeGuard[SingleSource[T, S]]:
        return (
            isinstance(x, Node) or
            x is START
        )

    @classmethod
    def is_single_source_with_config(cls, x: Any) -> TypeGuard[SingleSourceWithConfig[T, S]]:
        return cls.is_node_with_config(x) or cls.is_single_source(x)

    @classmethod
    def is_single_source_list(cls, x: Any) -> TypeGuard[list[SingleSource[T, S]]]:
        return isinstance(x, list) and all(cls.is_single_source(n) for n in cast(list[Any], x))

    @classmethod
    def is_single_source_with_config_list(cls, x: Any) -> TypeGuard[list[SingleSourceWithConfig[T, S]]]:
        return isinstance(x, list) and all(cls.is_single_source_with_config(n) for n in cast(list[Any], x))

    @classmethod
    def is_source(cls, x: Any) -> TypeGuard[Source[T, S]]:
        return (
            cls.is_single_source(x) or
            cls.is_single_source_list(x)
        )

    @classmethod
    def is_source_with_config(cls, x: Any) -> TypeGuard[SourceWithConfig[T, S]]:
        return (
            cls.is_single_source_with_config(x) or
            cls.is_single_source_with_config_list(x)
        )

    @classmethod
    def is_single_error_source(cls, x: Any) -> TypeGuard[SingleErrorSource[T, S]]:
        return (
            (isinstance(x, type) and issubclass(x, Exception)) or
            (isinstance(x, tuple) and len(cast(tuple[Any], x)) == 2 and isinstance(x[0], Node) and isinstance(x[1], type) and issubclass(x[1], Exception))
        )

    @classmethod
    def is_single_error_source_list(cls, x: Any) -> TypeGuard[list[SingleErrorSource[T, S]]]:
        return isinstance(x, list) and all(cls.is_single_error_source(n) for n in cast(list[Any], x))

    @classmethod
    def is_error_source(cls, x: Any) -> TypeGuard[ErrorSource[T, S]]:
        return (
            cls.is_single_error_source(x) or
            cls.is_single_error_source_list(x)
        )

    # @classmethod
    # def is_single_branch_source(cls, x: Any) -> TypeGuard[SingleBranchSource[T, S]]:
    #     return x is START or cls.is_single_source(x)

    # @classmethod
    # def is_single_branch_source_list(cls, x: Any) -> TypeGuard[list[SingleBranchSource[T, S]]]:
    #     return isinstance(x, list) and all(cls.is_single_branch_source(n) for n in cast(list[Any], x))

    # @classmethod
    # def is_branch_source(cls, x: Any) -> TypeGuard[BranchSource[T, S]]:
    #     return (
    #         cls.is_single_branch_source(x) or
    #         cls.is_single_branch_source_list(x)
    #     )

    @classmethod
    def is_any_single_source(cls, x: Any) -> TypeGuard[SingleSourceWithConfig[T, S] | SingleErrorSource[T, S]]:
        return cls.is_single_source_with_config(x) or cls.is_single_error_source(x)

    @classmethod
    def is_any_single_source_list(cls, x: Any) -> TypeGuard[list[SingleSourceWithConfig[T, S] | SingleErrorSource[T, S]]]:
        return isinstance(x, list) and all(cls.is_any_single_source(n) for n in cast(list[Any], x))

    @classmethod
    def is_any_source(cls, x: Any) -> TypeGuard[SourceWithConfig[T, S] | ErrorSource[T, S]]:
        return cls.is_source_with_config(x) or cls.is_error_source(x)

    @classmethod
    def is_branch_join(cls, x: Any) -> TypeGuard[BranchJoin[T, S]]:
        return x is END or cls.is_next(x)

ErrorConfig

Bases: BaseModel

Configuration for the error edge.

Attributes:

Name Type Description
propagate bool

If the error should be propagated to the next error edge. If False, the error is caught and the graph continues.
Source code in src/edgygraph/graph/types.py 
201
202
203
204
205
206
207
208
209
class ErrorConfig(BaseModel):
    """
    Configuration for the error edge.

    Attributes:
        propagate: If the error should be propagated to the next error edge. If False, the error is caught and the graph continues.
    """

    propagate: bool = False

Edge

Bases: BaseModel

An edge in a branch.

Attributes:

Name Type Description
source SingleSource[T, S]

The source of the edge.
next Next[T, S]

The unresolved targets of the edge.
Source code in src/edgygraph/graph/types.py 
211
212
213
214
215
216
217
218
219
220
221
222
class Edge[T: StateProtocol, S: SharedProtocol](BaseModel):
    """
    An edge in a branch.

    Attributes:
        source: The source of the edge.
        next: The unresolved targets of the edge.
    """

    source: SingleSource[T, S]
    next: Next[T, S]
    model_config = ConfigDict(arbitrary_types_allowed=True)

ErrorEdge

Bases: BaseModel

An error edge in a branch.

Attributes:

Name Type Description
source SingleErrorSource[T, S]

The source of the error edge.
next Next[T, S]

The unresolved targets of the error edge.
config ErrorConfig

The configuration of the error edge.
Source code in src/edgygraph/graph/types.py 
224
225
226
227
228
229
230
231
232
233
234
235
236
237
class ErrorEdge[T: StateProtocol, S: SharedProtocol](BaseModel):
    """
    An error edge in a branch.

    Attributes:
        source: The source of the error edge.
        next: The unresolved targets of the error edge.
        config: The configuration of the error edge.
    """

    source: SingleErrorSource[T, S]
    next: Next[T, S]
    config: ErrorConfig = Field(default_factory=ErrorConfig)
    model_config = ConfigDict(arbitrary_types_allowed=True)

BaseEntry

Bases: BaseModel

Base class for the values of edge indexing dictionaries of a branch.

Do not instantiate directly.

Attributes:

Name Type Description
next Next[T, S]

The unresolved targets of the edge.
index int

The original index of the entry in the list of edges of the branch.
Source code in src/edgygraph/graph/types.py 
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
class BaseEntry[T: StateProtocol, S: SharedProtocol](BaseModel):
    """
    Base class for the values of edge indexing dictionaries of a branch.

    Do not instantiate directly.

    Attributes:
        next: The unresolved targets of the edge.
        index: The original index of the entry in the list of edges of the branch.
    """

    next: Next[T, S]
    index: int
    model_config = ConfigDict(arbitrary_types_allowed=True)

    def model_post_init(self, __context: Any):
        if type(self) is BaseEntry:
            raise Exception("BaseEntry is not meant to be instantiated directly.") # Safeguard

Entry

Bases: BaseEntry[T, S]

A value of the edge indexing dictionary of a branch.

Attributes:

Name Type Description
next Next[T, S]

The unresolved targets of the edge.
index int

The original index of the entry in the list of edges.
Source code in src/edgygraph/graph/types.py 
260
261
262
263
264
265
266
267
class Entry[T: StateProtocol, S: SharedProtocol](BaseEntry[T, S]):
    """
    A value of the edge indexing dictionary of a branch.

    Attributes:
        next: The unresolved targets of the edge.
        index: The original index of the entry in the list of edges.
    """

ErrorEntry

Bases: BaseEntry[T, S]

A value of the error edge indexing dictionary of a branch.

Attributes:

Name Type Description
next Next[T, S]

The unresolved targets of the edge.
index int

The original index of the entry in the list of edges.
propagate bool

If the error should be reraised. If False, the error is caught and the graph continues.
Source code in src/edgygraph/graph/types.py 
269
270
271
272
273
274
275
276
277
278
279
class ErrorEntry[T: StateProtocol, S: SharedProtocol](BaseEntry[T, S]):
    """
    A value of the error edge indexing dictionary of a branch.

    Attributes:
        next: The unresolved targets of the edge.
        index: The original index of the entry in the list of edges.
        propagate: If the error should be reraised. If False, the error is caught and the graph continues.
    """

    propagate: bool = Field(default=False)

NextNode

Bases: BaseModel

A node that is the target of an edge.

Attributes:

Name Type Description
node Node[T, S]

The node.
reached_by Entries[T, S]

The edge that targeted this node.
Source code in src/edgygraph/graph/types.py 
286
287
288
289
290
291
292
293
294
295
296
297
class NextNode[T: StateProtocol, S: SharedProtocol](BaseModel):
    """
    A node that is the target of an edge.

    Attributes:
        node: The node.
        reached_by: The edge that targeted this node.
    """

    node: Node[T, S]
    reached_by: Entries[T, S]
    model_config = ConfigDict(arbitrary_types_allowed=True)