Skip to content

Channels

BaseChannel

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

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.

Topic

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

A configurable PubSub Topic.

Parameters:

  • typ (Type[Value]) –

    The type of the value stored in the channel.

  • accumulate (bool, default: False ) –

    Whether to accumulate values across steps. If False, the channel will be emptied after each step.

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.

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.

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)

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.

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