event

class Event

Container for a single value returned from a single Node.process() call

id: int

An integer event ID that combines some random prefix with a monotonic sequence counter.

Constructed like:

(base << 32) ^ sequence

The base and sequence should both be integers < 2**32, but for performance/YAGNI reasons bounds are not checked (for now).

EventIDs should be treated as opaque identifiers for now, they bear no inherent information about an event source (except that events were produced within the same context). In the future the ID system may be unified such that events, nodes, runners, etc. have a unified hierarchical identifier system.

See also

event_id() , event_id_parts()

timestamp: Annotated[datetime, BeforeValidator(func=_from_isoformat, json_schema_input_type=PydanticUndefined), PlainSerializer(func=_to_isoformat, return_type=PydanticUndefined, when_used=json)]

Timestamp of when the event was received by the TubeRunner

node_id: str

ID of node that emitted the event

signal: str

name of the signal that emitted the event

epoch: Epoch

Epoch number the event was emitted in

value: Annotated[_TEvent, BeforeValidator(func=_from_jsonable_pickle, json_schema_input_type=PydanticUndefined), WrapSerializer(func=_to_jsonable_pickle, return_type=PydanticUndefined, when_used=json)]

Value emitted by the processing node

is_event(instance: Any) → bool

TypedDicts don’t support instancechecks by default, we want to use typed dicts for perf’s sake, but we still also would like to check if something is an event

class MetaEventType(*values)

Types of meta events emitted by tubes, schedulers, stores, and runners.

NodeReady = 'NodeReady'

A node was made ready in the toplogical sorting graph. The value of the event is the node_id of the node that is ready.

EpochEnded = 'EpochEnded'

An epoch has ended, the value of the event contains which epoch has ended

class MetaSignal(*values)

NoEvent = '__META__NoEvent'

class MetaEvent

All events generated by internal processes rather than nodes.

Used to coordinate the tube as well as allow code to hook into tube execution.

These are not stored in the EventStore, but emitted by callbacks and consumed internally.

See MetaEventType for descriptions of the types of MetaEvents.

node_id: Literal['meta']

signal: MetaEventType

id: int

timestamp: Annotated[datetime, BeforeValidator(func=_from_isoformat, json_schema_input_type=PydanticUndefined), PlainSerializer(func=_to_isoformat, return_type=PydanticUndefined, when_used=json)]

epoch: Epoch

value: Annotated[_TEvent, BeforeValidator(func=_from_jsonable_pickle, json_schema_input_type=PydanticUndefined), WrapSerializer(func=_to_jsonable_pickle, return_type=PydanticUndefined, when_used=json)]

event_id(prefix: int, sequence: int, shifted: bool = False) → int

Construct an event id from a prefix and a sequence like:

(prefix << _ID_SEQUENCE_BITS) ^ sequence

Parameters:

• prefix (int) – A 32-bit prefix

• sequence (int) – A 32-bit monotonically increasing sequence

• shifted (bool) – If True , prefix is already shifted (for performance’s sake, avoid an unnecessary operation on a high-call path)

Returns:

A 64-bit event ID

event_id_parts(event_id: int) → tuple[int, int]

Decompose an event ID into its prefix and sequence identifier

Parameters:

event_id (int) – A 64-bit event ID

Returns:

tuple[int, int] - a tuple of the (prefix, sequence)

class EventMaker(prefix: int | None = None, node_id: str | None = None)

Convenience constructor for Event s / MetaEvent s for a single execution context (a scheduler, store, node, …), stamping each with a (locally) unique id and timestamp.

Prefixes are created with random rather than secrets , because they are only used for local disambiguation for now. IDs will be made secure for signing events when we need to handle networked event processing

prefix

node_id

new_event(value: _TEvent, epoch: Epoch, node_id: str | None = None, signal: str = 'value', timestamp: Annotated[datetime, BeforeValidator(func=_from_isoformat, json_schema_input_type=PydanticUndefined), PlainSerializer(func=_to_isoformat, return_type=PydanticUndefined, when_used=json)] | None = None) → Event[_TEvent]

Make an Event, filling in the id and (UTC) timestamp. Pass timestamp to share one across a batch of events emitted together.

new_meta_event(epoch: Epoch, signal: MetaEventType, value: Any = None, timestamp: Annotated[datetime, BeforeValidator(func=_from_isoformat, json_schema_input_type=PydanticUndefined), PlainSerializer(func=_to_isoformat, return_type=PydanticUndefined, when_used=json)] | None = None) → MetaEvent

Make a MetaEvent (node_id is always "meta"). See new_event().

NoEventable

Convenience generic type to indicate that some signal can be a NoEvent

alias of _NoEventableInner | Event[_NoEventableInner] | Literal[NoEvent]