Channels

Classes:

Name	Description
BaseChannel	Base class for all channels.

BaseChannel

Bases: Generic[Value, Update, C], ABC

Base class for all channels.

Methods:

Name	Description
copy	Return a copy of the channel.
checkpoint	Return a serializable representation of the channel's current state.
from_checkpoint	Return a new identical channel, optionally initialized from a checkpoint.
update	Update the channel's value with the given sequence of updates.
get	Return the current value of the channel.
consume	Mark the current value of the channel as consumed. By default, no-op.
is_available	Return True if the channel is available (not empty), False otherwise.

Attributes:

Name	Type	Description
ValueType	Any	The type of the value stored in the channel.
UpdateType	Any	The type of the update received by the channel.

ValueType abstractmethod property

ValueType: Any

The type of the value stored in the channel.

UpdateType abstractmethod property

UpdateType: Any

The type of the update received by the channel.

copy

copy() -> 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.

checkpoint

checkpoint() -> C

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.

from_checkpoint abstractmethod

from_checkpoint(checkpoint: C) -> Self

Return a new identical channel, optionally initialized from a checkpoint. If the checkpoint contains complex data structures, they should be copied.

update abstractmethod

update(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.

get abstractmethod

get() -> Value

Return the current value of the channel.

Raises EmptyChannelError if the channel is empty (never updated yet).

consume

consume() -> bool

Mark the current value of the channel as consumed. By default, no-op. This is called by Pregel before the start of the next step, for all channels that triggered a node. If the channel was updated, return True.

is_available

is_available() -> 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.

Classes:

Name	Description
Topic	A configurable PubSub Topic.
LastValue	Stores the last value received, can receive at most one value per step.
EphemeralValue	Stores the value received in the step immediately preceding, clears after.
BinaryOperatorAggregate	Stores the result of applying a binary operator to the current value and each new value.
AnyValue	Stores the last value received, assumes that if multiple values are

Topic

Bases: Generic[Value], BaseChannel[Sequence[Value], Union[Value, list[Value]], list[Value]]

A configurable PubSub Topic.

Parameters:

Name	Type	Description	Default
typ	type[Value]	The type of the value stored in the channel.	required
accumulate	bool	Whether to accumulate values across steps. If False, the channel will be emptied after each step.	False

Methods:

Name	Description
consume	Mark the current value of the channel as consumed. By default, no-op.
copy	Return a copy of the channel.

Attributes:

Name	Type	Description
ValueType	Any	The type of the value stored in the channel.
UpdateType	Any	The type of the update received by the channel.

ValueType property

ValueType: Any

The type of the value stored in the channel.

UpdateType property

UpdateType: Any

The type of the update received by the channel.

consume

consume() -> bool

Mark the current value of the channel as consumed. By default, no-op. This is called by Pregel before the start of the next step, for all channels that triggered a node. If the channel was updated, return True.

copy

copy() -> Self

Return a copy of the channel.

LastValue

Bases: Generic[Value], BaseChannel[Value, Value, Value]

Stores the last value received, can receive at most one value per step.

Methods:

Name	Description
consume	Mark the current value of the channel as consumed. By default, no-op.
copy	Return a copy of the channel.

Attributes:

Name	Type	Description
ValueType	type[Value]	The type of the value stored in the channel.
UpdateType	type[Value]	The type of the update received by the channel.

ValueType property

ValueType: type[Value]

The type of the value stored in the channel.

UpdateType property

UpdateType: type[Value]

The type of the update received by the channel.

consume

consume() -> bool

Mark the current value of the channel as consumed. By default, no-op. This is called by Pregel before the start of the next step, for all channels that triggered a node. If the channel was updated, return True.

copy

copy() -> Self

Return a copy of the channel.

EphemeralValue

Bases: Generic[Value], BaseChannel[Value, Value, Value]

Stores the value received in the step immediately preceding, clears after.

Methods:

Name	Description
consume	Mark the current value of the channel as consumed. By default, no-op.
copy	Return a copy of the channel.

Attributes:

Name	Type	Description
ValueType	type[Value]	The type of the value stored in the channel.
UpdateType	type[Value]	The type of the update received by the channel.

ValueType property

ValueType: type[Value]

The type of the value stored in the channel.

UpdateType property

UpdateType: type[Value]

The type of the update received by the channel.

consume

consume() -> bool

Mark the current value of the channel as consumed. By default, no-op. This is called by Pregel before the start of the next step, for all channels that triggered a node. If the channel was updated, return True.

copy

copy() -> Self

Return a copy of the channel.

BinaryOperatorAggregate

Bases: Generic[Value], BaseChannel[Value, Value, Value]

Stores the result of applying a binary operator to the current value and each new value.

import operator

total = Channels.BinaryOperatorAggregate(int, operator.add)

Methods:

Name	Description
consume	Mark the current value of the channel as consumed. By default, no-op.
copy	Return a copy of the channel.

Attributes:

Name	Type	Description
ValueType	type[Value]	The type of the value stored in the channel.
UpdateType	type[Value]	The type of the update received by the channel.

ValueType property

ValueType: type[Value]

The type of the value stored in the channel.

UpdateType property

UpdateType: type[Value]

The type of the update received by the channel.

consume

consume() -> bool

Mark the current value of the channel as consumed. By default, no-op. This is called by Pregel before the start of the next step, for all channels that triggered a node. If the channel was updated, return True.

copy

copy() -> Self

Return a copy of the channel.

AnyValue

Bases: Generic[Value], BaseChannel[Value, Value, Value]

Stores the last value received, assumes that if multiple values are received, they are all equal.

Methods:

Name	Description
consume	Mark the current value of the channel as consumed. By default, no-op.
copy	Return a copy of the channel.

Attributes:

Name	Type	Description
ValueType	type[Value]	The type of the value stored in the channel.
UpdateType	type[Value]	The type of the update received by the channel.

ValueType property

ValueType: type[Value]

The type of the value stored in the channel.

UpdateType property

UpdateType: type[Value]

The type of the update received by the channel.

consume

consume() -> bool

Mark the current value of the channel as consumed. By default, no-op. This is called by Pregel before the start of the next step, for all channels that triggered a node. If the channel was updated, return True.

copy

copy() -> Self

Return a copy of the channel.

Comments