Skip to content

Types

Types

Typeguards for runtime typechecking.

Source code in src/edgygraph/graph/types.py
 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
189
190
191
192
193
194
195
196
197
198
199
200
201
class Types[T: StateProtocol, S: SharedProtocol]:
    """
    Typeguards for runtime typechecking.
    """

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

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

    @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], NodeConfig) and
            isinstance(x[1], (Node, NavigationNode))
        )


    @classmethod
    def is_error_source(cls, x: Any) -> TypeGuard[ErrorSource[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], type) and issubclass(x[0], Exception) and isinstance(x[1], Node)
        )

    @classmethod
    def is_unresolved_source(cls, x: Any) -> TypeGuard[UnresolvedSource[T, S]]:
        return (
            cls.is_source(x)
            or cls.is_error_source(x) 
            or cls.is_node_with_config(x)
            or isinstance(x, list) and all(
                cls.is_source(n)
                or cls.is_node_with_config(n)
            for n in cast(list[Any], x)
            )
        )

    @classmethod
    def is_unresolved_source_only_flexible(cls, x: Any) -> TypeGuard[Flexible[Source[T, S] | NodeWithConfig[T, S]]]:
        return (
            cls.is_source(x)
            or cls.is_node_with_config(x)
            or isinstance(x, list) and all(
                cls.is_source(n)
                or cls.is_node_with_config(n)
            for n in cast(list[Any], x)
            )
        )

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

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

    @classmethod
    def is_next_callable(cls, x: Any) -> TypeGuard[NextCallable[T, S]]:
        """Limited to determine weither x is a callable and not another known class or type.
        Therefore it can be false positive for unknown classes or types, but it should not be false negative for callables.
        """
        return callable(x) and not (
            cls.is_source(x)
            or cls.is_next(x)
            or x is START
            or x is END
            or x is Exception
        )

    @classmethod
    def is_unresolved_next(cls, x: Any) -> TypeGuard[UnresolvedNext[T, S]]:
        return (
            cls.is_next(x) 
            or cls.is_node_with_config(x)
            or cls.is_next_callable(x) 
            or isinstance(x, list) and all(
                cls.is_next(n) 
                or cls.is_node_with_config(n) 
            for n in cast(list[Any], x)
            )
        )

    @classmethod
    def is_unresolved_next_only_flexible(cls, x: Any) -> TypeGuard[Flexible[Next[T, S] | NodeWithConfig[T, S]]]:
        return (
            cls.is_next(x) 
            or cls.is_node_with_config(x) 
            or isinstance(x, list) and all(
                cls.is_next(n) 
                or cls.is_node_with_config(n) 
            for n in cast(list[Any], x)
            )
        )

is_next_callable(x) classmethod

Limited to determine weither x is a callable and not another known class or type. Therefore it can be false positive for unknown classes or types, but it should not be false negative for callables.

Source code in src/edgygraph/graph/types.py
165
166
167
168
169
170
171
172
173
174
175
176
@classmethod
def is_next_callable(cls, x: Any) -> TypeGuard[NextCallable[T, S]]:
    """Limited to determine weither x is a callable and not another known class or type.
    Therefore it can be false positive for unknown classes or types, but it should not be false negative for callables.
    """
    return callable(x) and not (
        cls.is_source(x)
        or cls.is_next(x)
        or x is START
        or x is END
        or x is Exception
    )

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
340
341
342
343
344
345
346
347
348
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

BaseEdge

Bases: BaseModel

Base class for edges.

Attributes:

Name Type Description
next Flexible[Next[T, S]] | NextCallable[T, S]

The unresolved targets of the edge.

Source code in src/edgygraph/graph/types.py
350
351
352
353
354
355
356
357
358
359
class BaseEdge[T: StateProtocol, S: SharedProtocol](BaseModel):
    """
    Base class for edges.

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

    next: Flexible[Next[T, S]] | NextCallable[T, S]
    model_config = ConfigDict(arbitrary_types_allowed=True)

Edge

Bases: BaseEdge[T, S]

An edge in a branch.

Attributes:

Name Type Description
source Source[T, S]

The source of the edge.

Source code in src/edgygraph/graph/types.py
361
362
363
364
365
366
367
368
369
class Edge[T: StateProtocol, S: SharedProtocol](BaseEdge[T, S]):
    """
    An edge in a branch.

    Attributes:
        source: The source of the edge.
    """

    source: Source[T, S]

ErrorEdge

Bases: BaseEdge[T, S]

An error edge in a branch.

Attributes:

Name Type Description
source ErrorSource[T, S]

The source of the error edge.

config ErrorConfig

The configuration of the error edge.

Source code in src/edgygraph/graph/types.py
371
372
373
374
375
376
377
378
379
380
381
class ErrorEdge[T: StateProtocol, S: SharedProtocol](BaseEdge[T, S]):
    """
    An error edge in a branch.

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

    source: ErrorSource[T, S]
    config: ErrorConfig = Field(default_factory=ErrorConfig)

BaseEntry

Bases: BaseModel

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

Do not instantiate directly.

Attributes:

Name Type Description
next Flexible[Next[T, S]] | NextCallable[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
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
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: Flexible[Next[T, S]] | NextCallable[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 Flexible[Next[T, S]] | NextCallable[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
403
404
405
406
407
408
409
410
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 Flexible[Next[T, S]] | NextCallable[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
412
413
414
415
416
417
418
419
420
421
422
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] | NavigationNode

The node.

reached_by Entries[T, S]

The edge that targeted this node.

Source code in src/edgygraph/graph/types.py
429
430
431
432
433
434
435
436
437
438
439
440
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] | NavigationNode
    reached_by: Entries[T, S]
    model_config = ConfigDict(arbitrary_types_allowed=True)