base

process_method(func: _TProcess) → _TProcess

Decorator to mark a method as the designated ‘process’ method for a class.

pydantic model Node

A node within a processing tube

Config:

• extra: str = forbid

Fields:

• enabled (bool)

• id (str)

• spec (noob.node.spec.NodeSpecification | None)

• stateful (bool)

field enabled: bool = True

Starting state for a node being enabled. When a node is disabled, it will be deinitialized and removed from the processing graph, but the node object will still be kept by the Tube. Nodes can be disabled and enabled during a tube’s operation without recreating the tube. When a node is disabled, other nodes that depend on it will not be disabled, but they may never be called since their dependencies will never be satisfied.

field id: str [Required]

Unique identifier of the node

field spec: NodeSpecification | None = None

field stateful: bool = True

Whether this node is stateful (True), or stateless (False). Stateful nodes are assumed to care about the order in which they receive events - i.e. for a given set of inputs, the values returned by process are different when called in a different order.

This attribute has no effect in synchronous runners, but in concurrent runners where multiple epochs of events can be processed simultaneously, setting a node as stateless can improve performance as the node processes events as soon as it receives them rather than waiting for the next epoch in the sequence to arrive.

Defined as an instance, rather than a class attribute to allow it being overridden by a node specification. Subclasses should override the default value to be considered stateless by default.

By default, unless specified otherwise:

• Class nodes are considered stateful

• Generator nodes are considered stateful

• Function nodes are considered stateless

classmethod from_specification(spec: NodeSpecification, input_collection: InputCollection | None = None) → Node

Create a node from its spec

• resolve the node type

• if a function, wrap it in a node class

• if a class, just instantiate it

classmethod get_signals(spec: NodeSpecification | None = None) → dict[str, Signal]

Public class method for computing signals from a node class and a specification.

Exposed as a class method for reflection purposes - we don’t want to have to instantiate a node to tell what would come out of it, but signals can be dynamically computed by the node depending on its spec.

If signals can’t be computed (because they are not annotated, etc.), returns a generic “value” signal that refers to whatever the node produces.

Base implementation reads annotations from the process method and does not use the spec. If subclass overrides need a spec to compute their signals, they must raise a ValueError if none is provided.

classmethod get_slots(spec: NodeSpecification | None = None) → dict[str, Slot]

Similar to get_signals(), but for slots!

deinit() → None

Stop producing, processing, or receiving data

Default is a no-op. Subclasses do not need to override if they have no deinit logic.

init() → None

Start producing, processing, or receiving data.

Default is a no-op. Subclasses do not need to override if they have no initialization logic.

Subclasses MAY add a context: RunnerContext param to request information about the enclosing runner while initializing

model_post_init(_Node__context: Any) → None

See docstring of process() for description of post init wrapping of generators

process(*args: Any, **kwargs: Any) → Any | None

Process some input, emitting it. See subclasses for details.

If the process method is a generator, when the Node class is instantiated, this method is replaced by one that wraps creating and calling the generator.

something like this:

`python gen = self.process() self.process = lambda: next(gen) `

Note that send handling is not implemented for generators, so process methods that are generators cannot depend on events from any other nodes (i.e. behave like source nodes).

property edges: list[Edge]

The dependencies this node has declared, express as edges between another node’s signals and our slots

property injections: dict[str, str]

If a node’s process method requests a dependency to be injected, returns a map from the type of inejction to the kwargs to pass them as.

property is_coroutine: bool

is the process method a coroutine?

(checking this on every call proves to be surprisingly expensive)

property signals: dict[str, Signal]

Cached instance-level accessor for signals.

Uses get_signals() and stores the result for frequent access.

property slots: dict[str, Slot]

pydantic model WrapClassNode

Wrap a non-Node class that has annotated one of its methods as being the “process” method using the process_method() decorator.

Wrapping allows us to use arbitrary classes as Nodes within noob, which expects a process method, but avoids the problem of potentially breaking the class if it has its own attribute or method named process.

After instantiating the outer wrapping class, instantiate the inner wrapped class using the params given to the outer wrapping class during model_post_init() . Then dynamically assign the discovered process method on the inner class to the outer class as process.

Dynamic discovery at instantiation time, rather than statically defining an outer process method that then calls the inner method annotated with process_method does two things:

• Allows us to statically infer whether the method is a regular function that return`s or a generator using :func:`inspect.isgeneratorfunction , which relies on a flag set on a method at the time it is defined: e.g. a method that internally switches between return self._wrapped() and yield from self._wrapped() would not be correctly detected.

• Avoids modifying the signature of the wrapped process method with generic args and kwargs

Config:

• extra: str = forbid

Fields:

• cls (type)

• instance (type | None)

• params (dict[str, Any])

field cls: type [Required]

field instance: type | None = None

field params: dict[str, Any] [Optional]

classmethod get_signals(spec: NodeSpecification | None = None) → dict[str, Signal]

classmethod get_slots(spec: NodeSpecification | None = None) → dict[str, Slot]

deinit() → None

model_post_init(context: Any, /) → None

Get the method decorated with process_method(), assign it to process, see class docstring.

pydantic model WrapFuncNode

Config:

• extra: str = forbid

Fields:

• fn (collections.abc.Callable)

• params (dict)

• stateful (bool)

Validators:

• set_default_statefulness » all fields

field fn: Callable [Required]

Validated by:

• set_default_statefulness

field params: dict [Optional]

Validated by:

• set_default_statefulness

field stateful: bool = False

Function nodes are considered stateless by default, except if they are generators, which are typically stateful.

Validated by:

• set_default_statefulness

classmethod get_signals(spec: NodeSpecification | None = None) → dict[str, Signal]

classmethod get_slots(spec: NodeSpecification | None = None) → dict[str, Slot]

validator set_default_statefulness  »  all fields

If no stateful argument is provided explicitly, set stateful default False for functions and True for generators

model_post_init(_WrapFuncNode__context: Any) → None

Complete wrapping fn without calling super() because we need to pass params to the function if it is a generator, and create a functools.partial() of it if it is not.