pydantic_graph.basenode

StateT module-attribute

StateT = TypeVar('StateT', default=object)

Type variable for the state in a graph.

GraphRunContext dataclass

Bases: Generic[StateT, DepsT]

Context for a graph.

Source code in pydantic_graph/pydantic_graph/basenode.py

@dataclass(kw_only=True)
class GraphRunContext(Generic[StateT, DepsT]):
    """Context for a graph."""

    state: StateT
    """The state of the graph."""
    deps: DepsT
    """Dependencies for the graph."""

state instance-attribute

state: StateT

The state of the graph.

deps instance-attribute

deps: DepsT

Dependencies for the graph.

BaseNode

Bases: ABC, Generic[StateT, DepsT, NodeRunEndT]

Base class for a node.

Source code in pydantic_graph/pydantic_graph/basenode.py

class BaseNode(ABC, Generic[StateT, DepsT, NodeRunEndT]):
    """Base class for a node."""

    @abstractmethod
    async def run(self, ctx: GraphRunContext[StateT, DepsT]) -> BaseNode[StateT, DepsT, Any] | End[NodeRunEndT]:
        """Run the node.

        This is an abstract method that must be implemented by subclasses.

        !!! note "Return types used at runtime"
            The return type of this method are read by `pydantic_graph` at runtime and used to define which
            nodes can be called next in the graph, and enforced when running the graph.

        Args:
            ctx: The graph context.

        Returns:
            The next node to run or [`End`][pydantic_graph.basenode.End] to signal the end of the graph.
        """
        ...

    @classmethod
    @cache
    def get_node_id(cls) -> str:
        """Get the ID of the node."""
        return cls.__name__

run abstractmethod async

run(
    ctx: GraphRunContext[StateT, DepsT],
) -> BaseNode[StateT, DepsT, Any] | End[NodeRunEndT]

Run the node.

This is an abstract method that must be implemented by subclasses.

Return types used at runtime

The return type of this method are read by pydantic_graph at runtime and used to define which nodes can be called next in the graph, and enforced when running the graph.

Parameters:

Name	Type	Description	Default
ctx	GraphRunContext[StateT, DepsT]	The graph context.	required

Returns:

Type	Description
BaseNode[StateT, DepsT, Any] | End[NodeRunEndT]	The next node to run or End to signal the end of the graph.

Source code in pydantic_graph/pydantic_graph/basenode.py

@abstractmethod
async def run(self, ctx: GraphRunContext[StateT, DepsT]) -> BaseNode[StateT, DepsT, Any] | End[NodeRunEndT]:
    """Run the node.

    This is an abstract method that must be implemented by subclasses.

    !!! note "Return types used at runtime"
        The return type of this method are read by `pydantic_graph` at runtime and used to define which
        nodes can be called next in the graph, and enforced when running the graph.

    Args:
        ctx: The graph context.

    Returns:
        The next node to run or [`End`][pydantic_graph.basenode.End] to signal the end of the graph.
    """
    ...

get_node_id cached classmethod

get_node_id() -> str

Get the ID of the node.

Source code in pydantic_graph/pydantic_graph/basenode.py

@classmethod
@cache
def get_node_id(cls) -> str:
    """Get the ID of the node."""
    return cls.__name__

End dataclass

Bases: Generic[RunEndT]

Type to return from a node to signal the end of the graph.

Source code in pydantic_graph/pydantic_graph/basenode.py

@dataclass
class End(Generic[RunEndT]):
    """Type to return from a node to signal the end of the graph."""

    data: RunEndT
    """Data to return from the graph."""

data instance-attribute

data: RunEndT

Data to return from the graph.

Edge dataclass

Annotation to apply a label to an edge in a graph.

Source code in pydantic_graph/pydantic_graph/basenode.py

@dataclass(frozen=True)
class Edge:
    """Annotation to apply a label to an edge in a graph."""

    label: str | None
    """Label for the edge."""

label instance-attribute

label: str | None

Label for the edge.

DepsT module-attribute

DepsT = TypeVar("DepsT", default=object, contravariant=True)

Type variable for the dependencies of a graph and node.

RunEndT module-attribute

RunEndT = TypeVar("RunEndT", covariant=True, default=object)

Covariant type variable for the return type of a graph run.

NodeRunEndT module-attribute

NodeRunEndT = TypeVar(
    "NodeRunEndT", covariant=True, default=Never
)

Covariant type variable for the return type of a node run.