stroblme
93e45274d0
Docs / docs (push) Successful in 23s
Playwright Tests / test-playwright (1, 2) (push) Successful in 3m12s
Playwright Tests / test-playwright (2, 2) (push) Successful in 1m49s
pre-commit / pre-commit (push) Failing after 2m16s
Test Backend / test-backend (push) Successful in 2m38s
Compose Smoke Test / test-compose (push) Successful in 32s
Playwright Tests / merge-reports (push) Successful in 1m15s
Node stop paths cancelled their background task and then caught CancelledError around the await, which swallows a cancellation aimed at the caller — the trap Supervisor._cancel already documents. One shared Node._cancel_task now waits the way the supervisor does; mqtt's publisher and subscription and delay's cron call it. The api container also collected zombie python workers: orphaned when --reload replaces the process holding their handle, they reparent onto a PID 1 that reaps nothing but its own. `init: true` on the backend service.
462 lines
18 KiB
Python
462 lines
18 KiB
Python
"""The node base class: identity, ports, lifecycle and execution.
|
|
|
|
Every node type in this package derives from :class:`Node`; the type-specific
|
|
modules beside this one add what talking to a particular outside world means.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import inspect
|
|
import logging
|
|
from collections.abc import Callable, Coroutine, Iterable, Iterator
|
|
from typing import TYPE_CHECKING, Any
|
|
|
|
from fluksio.flow import logs
|
|
from fluksio.flow.messages import MessageSpec, qualify
|
|
from fluksio.flow.supervision import Supervisor
|
|
|
|
if TYPE_CHECKING:
|
|
from fastapi import FastAPI
|
|
|
|
from fluksio.flow.pipeline import Pipeline
|
|
from fluksio.flow.state import StateBackend
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
# What a node hands back: the pipeline's state once it is bound, since a
|
|
# trigger runs the graph, and its own outputs when it is not.
|
|
type NodeResult = StateBackend | dict[str, Any] | None
|
|
|
|
|
|
class NodeOutputError(TypeError):
|
|
"""A node function returned something that cannot be mapped onto ports."""
|
|
|
|
|
|
#: Settings the engine reads itself rather than handing to the node's function.
|
|
RESERVED_SETTINGS = frozenset({"synchronous"})
|
|
|
|
|
|
class Node:
|
|
"""
|
|
A pipeline node that wraps a function with typed inputs/outputs.
|
|
|
|
Nodes are the fundamental building blocks of a pipeline. Each node encapsulates
|
|
a function that processes data, with explicit message-based inputs (requires)
|
|
and outputs (provides). Nodes can be connected into a directed acyclic graph (DAG)
|
|
where data flows from upstream to downstream nodes.
|
|
|
|
:param f: The function to execute when the node runs.
|
|
:type f: Callable
|
|
:param requires: Input messages this node consumes. Can be a single Message
|
|
or list of Messages. Empty for source nodes.
|
|
:type requires: MessageSpec | list[MessageSpec]
|
|
:param provides: Output messages this node produces. Can be a single Message
|
|
or list of Messages.
|
|
:type provides: MessageSpec | list[MessageSpec]
|
|
:param params: This node's settings — constants of its function, stored
|
|
with the flow. A function node reads them as keyword arguments beside
|
|
its ports; a built-in type validates them against its own ``Params``.
|
|
:type params: dict
|
|
:param name: Optional name for the node. Defaults to function name.
|
|
:type name: str | None
|
|
|
|
:ivar synchronous: If True, node only executes when all required inputs have
|
|
new versions since last execution. Useful for synchronizing multiple streams.
|
|
:vartype synchronous: bool
|
|
|
|
:example:
|
|
>>> def process_temp(temperature, offset):
|
|
... return {"celsius": temperature * 0.5 + offset}
|
|
>>>
|
|
>>> temp_node = Node(
|
|
... f=process_temp,
|
|
... requires=MessageSpec(name="temperature", dtype=DType.FLOAT),
|
|
... provides=MessageSpec(name="celsius", dtype=DType.FLOAT),
|
|
... params={"offset": 32},
|
|
... )
|
|
"""
|
|
|
|
__slots__ = (
|
|
"f",
|
|
"id",
|
|
"flow",
|
|
"name",
|
|
"input_ports",
|
|
"output_ports",
|
|
"requires",
|
|
"provides",
|
|
"params",
|
|
"_pipeline",
|
|
"synchronous",
|
|
"_on_health",
|
|
"supervisor",
|
|
"fingerprint",
|
|
)
|
|
|
|
def __init__(
|
|
self,
|
|
f: Callable[..., Any],
|
|
requires: MessageSpec | Iterable[MessageSpec] = (),
|
|
provides: MessageSpec | Iterable[MessageSpec] = (),
|
|
params: dict[str, Any] | None = None,
|
|
name: str | None = None,
|
|
):
|
|
self.f = f
|
|
self._pipeline: Pipeline | None = None
|
|
self._on_health: Callable[[Node, str, str | None], None] | None = None
|
|
# Set by the controller before start(); a node built on its own (tests,
|
|
# previews) runs its loops unsupervised.
|
|
self.supervisor: Supervisor | None = None
|
|
# Everything about this node a cached result would depend on, hashed.
|
|
# Empty means it is not cacheable — which is every node type that is
|
|
# not code, and every node built for anything but a run.
|
|
self.fingerprint: str = ""
|
|
self.params = dict(params) if params else {}
|
|
self.synchronous = bool(self.params.get("synchronous", False))
|
|
|
|
self.input_ports = self._normalize_ports(requires)
|
|
self.output_ports = self._normalize_ports(provides)
|
|
|
|
self.flow = ""
|
|
self.name: str = name or str(getattr(f, "__name__", "node"))
|
|
self.id: str = self.name
|
|
self._index_ports()
|
|
|
|
@staticmethod
|
|
def _normalize_ports(
|
|
msgs: MessageSpec | Iterable[MessageSpec],
|
|
) -> list[MessageSpec]:
|
|
"""Accept a single spec or any iterable of them."""
|
|
if isinstance(msgs, MessageSpec):
|
|
return [msgs]
|
|
return list(msgs or ())
|
|
|
|
def _index_ports(self) -> None:
|
|
"""Index bound ports by message name; unbound ports have no wiring."""
|
|
self.requires: dict[str, MessageSpec] = {
|
|
s.name: s for s in self.input_ports if s.name
|
|
}
|
|
self.provides: dict[str, MessageSpec] = {
|
|
s.name: s for s in self.output_ports if s.name
|
|
}
|
|
|
|
def assign_flow(self, flow: str, node_id: str) -> None:
|
|
"""Place this node in a flow, qualifying its identity and messages.
|
|
|
|
Called once by the loader, before the pipeline is built.
|
|
"""
|
|
self.flow = flow
|
|
self.name = node_id
|
|
self.id = f"{flow}.{node_id}"
|
|
self.input_ports = [
|
|
s.model_copy(update={"name": qualify(flow, s.name)})
|
|
for s in self.input_ports
|
|
]
|
|
self.output_ports = [
|
|
s.model_copy(update={"name": qualify(flow, s.name)})
|
|
for s in self.output_ports
|
|
]
|
|
self._index_ports()
|
|
|
|
@property
|
|
def local_id(self) -> str:
|
|
"""The node's id within its flow."""
|
|
return self.name
|
|
|
|
# -------------------------------------------------------------------------
|
|
# Lifecycle
|
|
#
|
|
# A plain logic node has nothing to start or stop. The ones that talk to the
|
|
# outside — subscriptions, schedules, webhooks — override these, which is
|
|
# how the controller can bring a flow up or down without knowing what any
|
|
# particular node type is.
|
|
# -------------------------------------------------------------------------
|
|
|
|
# Whether running this node twice is the same as running it once. False
|
|
# for anything that reaches outside — a second publish is a second command
|
|
# to the device, which at-least-once delivery must not cause.
|
|
idempotent: bool = True
|
|
|
|
@classmethod
|
|
def instance_key(cls, params: dict[str, Any]) -> str | None:
|
|
"""Which outside thing these parameters point at, if any.
|
|
|
|
Two nodes with the same key talk to the same broker topic, URL or
|
|
bucket, however many flows they sit in — which is what the brain graph
|
|
draws as one neuron. ``None``, the default, means a node of this type
|
|
is only ever itself.
|
|
|
|
Called with the *stored* parameters, so a credential is still an
|
|
unresolved ``{"$secret": ...}`` reference. No key may be built from one:
|
|
an id is not a place for a password.
|
|
"""
|
|
return None
|
|
|
|
async def start(self, app: FastAPI | None = None) -> None:
|
|
"""Begin whatever this node listens to. Called when its flow starts."""
|
|
|
|
async def stop(self, app: FastAPI | None = None) -> None:
|
|
"""Undo :meth:`start`. Called before a rebuild, and must be idempotent."""
|
|
|
|
def _run_supervised(
|
|
self, name: str, factory: Callable[[], Coroutine[Any, Any, None]]
|
|
) -> asyncio.Task[None] | None:
|
|
"""Run a background loop under supervision where there is one.
|
|
|
|
Returns the task only when unsupervised, since that is the one case
|
|
where the caller has to cancel it itself.
|
|
"""
|
|
if self.supervisor is not None:
|
|
self.supervisor.spawn(f"{self.id}:{name}", self.flow, factory)
|
|
return None
|
|
return asyncio.create_task(factory())
|
|
|
|
@staticmethod
|
|
async def _cancel_task(task: asyncio.Task[None]) -> None:
|
|
"""Stop an unsupervised loop and wait for it to be gone.
|
|
|
|
`wait` keeps whatever the task raises on its way out to itself, and
|
|
lets a cancellation aimed at *this* coroutine through — the
|
|
`except CancelledError` around `await task` it replaces swallowed that,
|
|
which left whoever asked for the teardown unkillable. The same trap
|
|
`Supervisor._cancel` documents.
|
|
"""
|
|
task.cancel()
|
|
await asyncio.wait([task])
|
|
if not task.cancelled():
|
|
# Retrieved so a crash on the way out is not reported at exit.
|
|
task.exception()
|
|
|
|
def report_health(self, status: str, detail: str | None = None) -> None:
|
|
"""Say how this node's connection is doing: ok, degraded or down."""
|
|
if self._on_health is not None:
|
|
self._on_health(self, status, detail)
|
|
|
|
def bind(self, pipeline: Pipeline) -> None:
|
|
"""
|
|
Bind this node to a pipeline for external triggering.
|
|
|
|
Once bound, the node can trigger downstream execution when called.
|
|
This is typically done automatically during pipeline construction.
|
|
|
|
:param pipeline: The pipeline to bind this node to.
|
|
:type pipeline: Pipeline
|
|
"""
|
|
self._pipeline = pipeline
|
|
|
|
# -------------------------------------------------------------------------
|
|
# Node-private state
|
|
#
|
|
# Code you write in a `python` node has no state handle: it is a pure
|
|
# function of its messages, and remembers things by reading a message it
|
|
# also writes. Node *types* the engine ships are a different matter — a
|
|
# rate limiter or a filter-on-change is about what happened last time, and
|
|
# that belongs to the engine rather than to any flow. It lives under a
|
|
# reserved key prefix, beside the timestamps and versions the pipeline
|
|
# already keeps there, so it survives a rebuild and never shows up as a
|
|
# message.
|
|
# -------------------------------------------------------------------------
|
|
|
|
def _state_key(self, key: str) -> str:
|
|
return f"__node__:{self.id}:{key}"
|
|
|
|
def remember(self, key: str, value: Any) -> None:
|
|
"""Keep a value for the next run of this node."""
|
|
if self._pipeline is not None:
|
|
self._pipeline.state[self._state_key(key)] = value
|
|
|
|
def recall(self, key: str, default: Any = None) -> Any:
|
|
"""What this node last remembered under ``key``."""
|
|
if self._pipeline is None:
|
|
return default
|
|
return self._pipeline.state.get(self._state_key(key), default)
|
|
|
|
def forget(self, key: str) -> None:
|
|
if self._pipeline is not None:
|
|
self._pipeline.state.delete(self._state_key(key))
|
|
|
|
def __repr__(self) -> str:
|
|
return self.id
|
|
|
|
def __hash__(self) -> int:
|
|
return hash(self.id)
|
|
|
|
# -------------------------------------------------------------------------
|
|
# Port translation
|
|
#
|
|
# The graph speaks qualified message names; node functions speak ports.
|
|
# -------------------------------------------------------------------------
|
|
|
|
def _to_kwargs(self, inputs: dict[str, Any]) -> dict[str, Any]:
|
|
kwargs = {}
|
|
for msg_name, value in inputs.items():
|
|
spec = self.requires.get(msg_name)
|
|
if spec is None:
|
|
continue
|
|
spec.check(value)
|
|
kwargs[spec.port] = value
|
|
return kwargs
|
|
|
|
def _to_messages(self, retval: Any) -> dict[str, Any] | None:
|
|
"""Map a function's port-keyed return value onto message names.
|
|
|
|
Only ``None`` means "nothing to publish". A falsy value of the wrong
|
|
shape — ``0``, ``""``, an empty list — is a mistake worth naming rather
|
|
than silence someone has to debug from an empty canvas. So is a key no
|
|
port declares: a mistyped metric name is how a training curve goes
|
|
missing, and it costs nothing to say so at the first yield.
|
|
"""
|
|
if retval is None:
|
|
return None
|
|
if not isinstance(retval, dict):
|
|
raise NodeOutputError(
|
|
f"'{self.local_id}' returned {type(retval).__name__}. Outputs are "
|
|
"keyed by port, so return a dict like {'out': value}, or None to "
|
|
"publish nothing."
|
|
)
|
|
by_port = {s.port: s for s in self.output_ports if s.name}
|
|
outputs = {}
|
|
for key, value in retval.items():
|
|
spec = by_port.get(key) or self.provides.get(key)
|
|
if spec is None:
|
|
declared = sorted(by_port) or ["none"]
|
|
raise NodeOutputError(
|
|
f"'{self.local_id}' produced '{key}', which no port "
|
|
f"declares. Its ports are: {', '.join(declared)}."
|
|
)
|
|
try:
|
|
spec.check(value)
|
|
except TypeError as exc:
|
|
# Same category as an undeclared key: what the node produced is
|
|
# wrong. Naming it as one is what lets an emission fail the
|
|
# call rather than being logged where nobody reads it.
|
|
raise NodeOutputError(f"'{self.local_id}' {exc}") from exc
|
|
outputs[spec.name] = value
|
|
return outputs or None
|
|
|
|
def execute(self, inputs: dict[str, Any] | None = None) -> dict[str, Any] | None:
|
|
"""Run the node function and return its outputs by message name.
|
|
|
|
Downstream nodes are not triggered — the pipeline schedules those.
|
|
"""
|
|
kwargs = self._to_kwargs(inputs or {})
|
|
result = self.f(**kwargs, params=self.params)
|
|
if inspect.isgenerator(result):
|
|
# Only when the node runs in this process; out of process the
|
|
# worker has already drained it and sent each yield on ahead.
|
|
result = self._drain(result)
|
|
return self._to_messages(result)
|
|
|
|
def _drain(self, generator: Iterator[Any]) -> Any:
|
|
"""Publish each yield as it happens; the end of it is the result.
|
|
|
|
The last yield is the result when the generator returns nothing of its
|
|
own, and which one is last is only known once it ends — so a value is
|
|
published one behind but *checked* the moment it arrives, which is what
|
|
fails a mistyped port at the yield that produced it.
|
|
"""
|
|
pending: Any = None
|
|
have_pending = False
|
|
try:
|
|
while True:
|
|
value = next(generator)
|
|
self._to_messages(value)
|
|
if have_pending:
|
|
self.emit(pending)
|
|
pending, have_pending = value, True
|
|
except StopIteration as stop:
|
|
if stop.value is not None:
|
|
if have_pending:
|
|
self.emit(pending)
|
|
return stop.value
|
|
return pending if have_pending else None
|
|
|
|
def emit(self, values: dict[str, Any] | None) -> None:
|
|
"""Publish on this node's output ports mid-execution.
|
|
|
|
A value produced while a node is still working is a value like any
|
|
other: same ports, same type checking, same place on the canvas. What
|
|
it is *not* is a log — nothing leaves a node except through a port it
|
|
declared, so a training curve is an output of the graph rather than a
|
|
side effect beside it.
|
|
"""
|
|
outputs = self._to_messages(values)
|
|
if outputs and self._pipeline is not None:
|
|
self._pipeline.publish_emission(self, outputs)
|
|
|
|
def trigger(
|
|
self, inputs: dict[str, Any] | None = None, durable: bool | None = None
|
|
) -> NodeResult:
|
|
"""
|
|
Trigger this node externally, executing downstream nodes if dependencies are met.
|
|
|
|
This method is for nodes that receive data from upstream dependencies.
|
|
For trigger/subscriber nodes that inject data into the pipeline, use :meth:`inject`.
|
|
|
|
:param inputs: Input values matching this node's ``requires``.
|
|
:type inputs: dict | None
|
|
:returns: Result of the node execution and downstream propagation.
|
|
:rtype: dict | None
|
|
:raises RuntimeError: If node is not bound to a pipeline.
|
|
"""
|
|
if self._pipeline is None:
|
|
raise RuntimeError("Node must be bound to a pipeline to trigger")
|
|
return self(inputs, durable=durable)
|
|
|
|
def inject(
|
|
self, outputs: dict[str, Any] | None = None, durable: bool | None = None
|
|
) -> NodeResult:
|
|
"""
|
|
Inject data into the pipeline as if this node produced it.
|
|
|
|
This method is for trigger/subscriber nodes that receive external data
|
|
(e.g., HTTP requests, MQTT messages) and need to inject it into the pipeline.
|
|
The data is validated against this node's ``provides`` specification.
|
|
|
|
For source nodes (nodes with no ``requires``), if no outputs are provided,
|
|
the node's function will be executed to generate outputs.
|
|
|
|
:param outputs: Output values matching this node's ``provides``.
|
|
:type outputs: dict | None
|
|
:returns: Result of downstream propagation.
|
|
:rtype: dict | None
|
|
:raises RuntimeError: If node is not bound to a pipeline.
|
|
:raises TypeError: If output values don't match ``provides`` types.
|
|
:raises KeyError: If required output keys are missing.
|
|
"""
|
|
if self._pipeline is None:
|
|
raise RuntimeError("Node must be bound to a pipeline to inject")
|
|
|
|
outputs = outputs or {}
|
|
|
|
# A source node asked to inject nothing produces its own data.
|
|
if not outputs and not self.requires:
|
|
collected = logs.Collector()
|
|
with logs.capture(collected):
|
|
outputs = self.f(params=self.params) or {}
|
|
self._pipeline.publish_log(self, collected, "")
|
|
|
|
return self._pipeline.trigger(self, self._to_messages(outputs), durable=durable)
|
|
|
|
def __call__(
|
|
self, inputs: dict[str, Any] | None = None, durable: bool | None = None
|
|
) -> NodeResult:
|
|
"""
|
|
Execute the node and trigger downstream nodes if bound to a pipeline.
|
|
|
|
Validates inputs against the node's ``requires`` specification, executes
|
|
the wrapped function, validates outputs, and triggers downstream execution
|
|
if the node is bound to a pipeline.
|
|
|
|
:param inputs: Input values keyed by message name. Must match the node's
|
|
``requires`` specification.
|
|
:type inputs: dict | None
|
|
:returns: Node outputs if successful, or pipeline execution results if bound.
|
|
:rtype: dict | None
|
|
"""
|
|
outputs = self.execute(inputs)
|
|
if self._pipeline is None:
|
|
return outputs
|
|
return self._pipeline.trigger(self, outputs, durable=durable)
|