from __future__ import annotations
from abc import ABC, abstractmethod
from collections.abc import Sequence
from typing import Any, Generic, TypeVar
from typing_extensions import Self
from langgraph._internal._typing import MISSING
from langgraph.errors import EmptyChannelError
Value = TypeVar("Value")
Update = TypeVar("Update")
Checkpoint = TypeVar("Checkpoint")
__all__ = ("BaseChannel",)
class BaseChannel(Generic[Value, Update, Checkpoint], ABC):
"""Base class for all channels."""
__slots__ = ("key", "typ")
def __init__(self, typ: Any, key: str = "") -> None:
self.typ = typ
self.key = key
@property
@abstractmethod
def ValueType(self) -> Any:
"""The type of the value stored in the channel."""
@property
@abstractmethod
def UpdateType(self) -> Any:
"""The type of the update received by the channel."""
# serialize/deserialize methods
def copy(self) -> Self:
"""Return a copy of the channel.
By default, delegates to checkpoint() and from_checkpoint().
Subclasses can override this method with a more efficient implementation."""
return self.from_checkpoint(self.checkpoint())
def checkpoint(self) -> Checkpoint | Any:
"""Return a serializable representation of the channel's current state.
Raises EmptyChannelError if the channel is empty (never updated yet),
or doesn't support checkpoints."""
try:
return self.get()
except EmptyChannelError:
return MISSING
@abstractmethod
def from_checkpoint(self, checkpoint: Checkpoint | Any) -> Self:
"""Return a new identical channel, optionally initialized from a checkpoint.
If the checkpoint contains complex data structures, they should be copied."""
# read methods
@abstractmethod
def get(self) -> Value:
"""Return the current value of the channel.
Raises EmptyChannelError if the channel is empty (never updated yet)."""
def is_available(self) -> bool:
"""Return True if the channel is available (not empty), False otherwise.
Subclasses should override this method to provide a more efficient
implementation than calling get() and catching EmptyChannelError.
"""
try:
self.get()
return True
except EmptyChannelError:
return False
# write methods
@abstractmethod
def update(self, values: Sequence[Update]) -> bool:
"""Update the channel's value with the given sequence of updates.
The order of the updates in the sequence is arbitrary.
This method is called by Pregel for all channels at the end of each step.
If there are no updates, it is called with an empty sequence.
Raises InvalidUpdateError if the sequence of updates is invalid.
Returns True if the channel was updated, False otherwise."""
def consume(self) -> bool:
"""Notify the channel that a subscribed task ran. By default, no-op.
A channel can use this method to modify its state, preventing the value
from being consumed again.
Returns True if the channel was updated, False otherwise.
"""
return False
def finish(self) -> bool:
"""Notify the channel that the Pregel run is finishing. By default, no-op.
A channel can use this method to modify its state, preventing finish.
Returns True if the channel was updated, False otherwise.
"""
return False