codekingpro/portable-devtools

1from __future__ import annotations2 3from abc import ABC, abstractmethod4from collections.abc import Sequence5from typing import Any, Generic, TypeVar6 7from typing_extensions import Self8 9from langgraph._internal._typing import MISSING10from langgraph.errors import EmptyChannelError11 12Value = TypeVar("Value")13Update = TypeVar("Update")14Checkpoint = TypeVar("Checkpoint")15 16__all__ = ("BaseChannel",)17 18 19class BaseChannel(Generic[Value, Update, Checkpoint], ABC):20    """Base class for all channels."""21 22    __slots__ = ("key", "typ")23 24    def __init__(self, typ: Any, key: str = "") -> None:25        self.typ = typ26        self.key = key27 28    @property29    @abstractmethod30    def ValueType(self) -> Any:31        """The type of the value stored in the channel."""32 33    @property34    @abstractmethod35    def UpdateType(self) -> Any:36        """The type of the update received by the channel."""37 38    # serialize/deserialize methods39 40    def copy(self) -> Self:41        """Return a copy of the channel.42 43        By default, delegates to `checkpoint()` and `from_checkpoint()`.44 45        Subclasses can override this method with a more efficient implementation.46        """47        return self.from_checkpoint(self.checkpoint())48 49    def checkpoint(self) -> Checkpoint | Any:50        """Return a serializable representation of the channel's current state.51 52        Raises `EmptyChannelError` if the channel is empty (never updated yet),53        or doesn't support checkpoints.54        """55        try:56            return self.get()57        except EmptyChannelError:58            return MISSING59 60    @abstractmethod61    def from_checkpoint(self, checkpoint: Checkpoint | Any) -> Self:62        """Return a new identical channel, optionally initialized from a checkpoint.63 64        If the checkpoint contains complex data structures, they should be copied.65        """66 67    # read methods68 69    @abstractmethod70    def get(self) -> Value:71        """Return the current value of the channel.72 73        Raises `EmptyChannelError` if the channel is empty (never updated yet)."""74 75    def is_available(self) -> bool:76        """Return `True` if the channel is available (not empty), `False` otherwise.77 78        Subclasses should override this method to provide a more efficient79        implementation than calling `get()` and catching `EmptyChannelError`.80        """81        try:82            self.get()83            return True84        except EmptyChannelError:85            return False86 87    # write methods88 89    @abstractmethod90    def update(self, values: Sequence[Update]) -> bool:91        """Update the channel's value with the given sequence of updates.92        The order of the updates in the sequence is arbitrary.93        This method is called by Pregel for all channels at the end of each step.94 95        If there are no updates, it is called with an empty sequence.96 97        Raises `InvalidUpdateError` if the sequence of updates is invalid.98 99        Returns `True` if the channel was updated, `False` otherwise."""100 101    def consume(self) -> bool:102        """Notify the channel that a subscribed task ran.103 104        By default, no-op.105 106        A channel can use this method to modify its state, preventing the value from being consumed again.107 108        Returns `True` if the channel was updated, `False` otherwise.109        """110        return False111 112    def finish(self) -> bool:113        """Notify the channel that the Pregel run is finishing.114 115        By default, no-op.116 117        A channel can use this method to modify its state, preventing finish.118 119        Returns `True` if the channel was updated, `False` otherwise.120        """121        return False122