stroblme
a96f6dd015
Moving a dashboard slider lit up an edge between two nodes that had done nothing. The canvas pulsed on the message's timestamp alone, and a message has no idea who published it — so it credited whichever node happened to be drawn as a producer. That was never only about dashboards. Two nodes producing one message pulsed both their edges whichever fired, and a message produced in another flow changed with nothing on screen to account for it at all. Values now carry their cause: a node, a dashboard widget, another flow, an agent or an API caller. An edge pulses only for the producer that actually published, and the edge inspector says where a value came from when it did not come from a node. What is not a node in this flow is now drawn as one — a label rather than a card, because a dashboard with twenty tiles would otherwise bury the logic the canvas exists to show. That covers cross-flow wiring too, which is the link in/out affordance that has been missing. They are never part of the document. They join at render, after everything that reads or writes the canvas nodes, so an autosave, an undo or a delete cannot reach them — with a Playwright test that drags a node and asserts the stored flow still holds exactly what it did.
835 lines
30 KiB
Python
835 lines
30 KiB
Python
"""Turns stored flows into a running pipeline.
|
|
|
|
The controller is the only thing that builds nodes: it reads flow definitions
|
|
from the store, instantiates each node from its type, and rebuilds the shared
|
|
pipeline. A node that fails to load is reported and skipped — one broken node
|
|
never stops the rest.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import hashlib
|
|
import logging
|
|
import sys
|
|
import traceback
|
|
from collections.abc import Callable
|
|
from dataclasses import dataclass, field
|
|
from enum import Enum
|
|
from types import ModuleType
|
|
from typing import Any, cast
|
|
|
|
from fastapi import FastAPI
|
|
from fastapi.concurrency import run_in_threadpool
|
|
|
|
from app.flow.alerts import AlertManager
|
|
from app.flow.events import EventBus
|
|
from app.flow.executor import ExecutionService
|
|
from app.flow.messages import MessageSpec, flow_of, qualify
|
|
from app.flow.nodes import (
|
|
ChangeNode,
|
|
DelayNode,
|
|
ExecNode,
|
|
FileNode,
|
|
HttpNode,
|
|
InfluxDbNode,
|
|
InjectNode,
|
|
JoinNode,
|
|
MLPNode,
|
|
MqttNode,
|
|
Node,
|
|
NtfyNode,
|
|
RbeNode,
|
|
SwitchNode,
|
|
TriggerNode,
|
|
)
|
|
from app.flow.pipeline import Pipeline, ValidationIssue, ValueSource
|
|
from app.flow.schemas import (
|
|
FlowDef,
|
|
Health,
|
|
NodeDef,
|
|
NodeStatusPublic,
|
|
NodeTypeInfo,
|
|
)
|
|
from app.flow.secrets import SecretNotFound, resolve_params
|
|
from app.flow.state import MemoryState, StateBackend
|
|
from app.flow.store import LIB_DIR, FlowNotFound, FlowStore, LibNotFound
|
|
from app.flow.supervision import Supervisor
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
HOOK_PREFIX = "/hooks"
|
|
|
|
|
|
class NodeStatus(str, Enum):
|
|
ACTIVE = "active"
|
|
ERROR = "error"
|
|
|
|
|
|
@dataclass
|
|
class LoadedNode:
|
|
"""A node as the editor sees it: built, or failed with a reason.
|
|
|
|
``status`` says whether it loaded; ``health`` says how its connection is
|
|
doing once it is running. A node can be perfectly loaded and unable to
|
|
reach its broker.
|
|
"""
|
|
|
|
id: str
|
|
flow: str
|
|
status: NodeStatus = NodeStatus.ACTIVE
|
|
node: Node | None = None
|
|
error: str | None = None
|
|
health: Health = "ok"
|
|
health_detail: str | None = None
|
|
|
|
|
|
@dataclass
|
|
class Preview:
|
|
"""A draft as it would run, without running it."""
|
|
|
|
nodes: list[NodeStatusPublic]
|
|
issues: list[ValidationIssue]
|
|
|
|
|
|
@dataclass
|
|
class NodeType:
|
|
"""A node type the editor can place on a canvas."""
|
|
|
|
title: str
|
|
description: str
|
|
# Constructor signatures differ per node type.
|
|
cls: Any
|
|
has_source: bool = False
|
|
free_params: bool = False
|
|
params_schema: dict[str, Any] = field(default_factory=dict)
|
|
#: Which installed package supplied this type, for the ones that are not
|
|
#: built in.
|
|
plugin: str | None = None
|
|
|
|
|
|
def _schema_of(cls: Any) -> dict[str, Any]:
|
|
params = getattr(cls, "Params", None)
|
|
return params.model_json_schema() if params is not None else {}
|
|
|
|
|
|
NODE_TYPES: dict[str, NodeType] = {
|
|
"python": NodeType(
|
|
title="Function",
|
|
description="Your own Python code, run on every incoming message.",
|
|
cls=Node,
|
|
has_source=True,
|
|
free_params=True,
|
|
),
|
|
"mqtt": NodeType(
|
|
title="MQTT",
|
|
description="Subscribe to topics, or publish what arrives on its inputs.",
|
|
cls=MqttNode,
|
|
params_schema=_schema_of(MqttNode),
|
|
),
|
|
"http": NodeType(
|
|
title="HTTP",
|
|
description="Receive data on a webhook, or send it to a URL.",
|
|
cls=HttpNode,
|
|
params_schema=_schema_of(HttpNode),
|
|
),
|
|
"influxdb": NodeType(
|
|
title="InfluxDB",
|
|
description="Write measurements to a bucket, or read them back.",
|
|
cls=InfluxDbNode,
|
|
params_schema=_schema_of(InfluxDbNode),
|
|
),
|
|
"delay": NodeType(
|
|
title="Delay & schedule",
|
|
description="Hold messages back, limit their rate, or emit on a schedule.",
|
|
cls=DelayNode,
|
|
params_schema=_schema_of(DelayNode),
|
|
),
|
|
"mlp": NodeType(
|
|
title="Perceptron",
|
|
description="A small neural layer over its numeric inputs.",
|
|
cls=MLPNode,
|
|
params_schema=_schema_of(MLPNode),
|
|
),
|
|
"inject": NodeType(
|
|
title="Inject",
|
|
description="Emit a value on request, on a timer, or when the flow starts.",
|
|
cls=InjectNode,
|
|
params_schema=_schema_of(InjectNode),
|
|
),
|
|
"switch": NodeType(
|
|
title="Switch",
|
|
description="Send a value down one branch or another, by rule.",
|
|
cls=SwitchNode,
|
|
params_schema=_schema_of(SwitchNode),
|
|
),
|
|
"change": NodeType(
|
|
title="Change",
|
|
description="Scale, offset, round or map a value on its way past.",
|
|
cls=ChangeNode,
|
|
params_schema=_schema_of(ChangeNode),
|
|
),
|
|
"rbe": NodeType(
|
|
title="Filter unchanged",
|
|
description="Pass a value on only when it has actually changed.",
|
|
cls=RbeNode,
|
|
params_schema=_schema_of(RbeNode),
|
|
),
|
|
"join": NodeType(
|
|
title="Join",
|
|
description="Gather several inputs into one object or list.",
|
|
cls=JoinNode,
|
|
params_schema=_schema_of(JoinNode),
|
|
),
|
|
"trigger": NodeType(
|
|
title="Trigger",
|
|
description="Send one value now and another once things go quiet.",
|
|
cls=TriggerNode,
|
|
params_schema=_schema_of(TriggerNode),
|
|
),
|
|
"exec": NodeType(
|
|
title="Command",
|
|
description="Run a command in the engine's container and read its output.",
|
|
cls=ExecNode,
|
|
params_schema=_schema_of(ExecNode),
|
|
),
|
|
"file": NodeType(
|
|
title="File",
|
|
description="Read a file into the flow, or write one out of it.",
|
|
cls=FileNode,
|
|
params_schema=_schema_of(FileNode),
|
|
),
|
|
"ntfy": NodeType(
|
|
title="Notification",
|
|
description="Push an incoming value to a phone through ntfy.",
|
|
cls=NtfyNode,
|
|
params_schema=_schema_of(NtfyNode),
|
|
),
|
|
}
|
|
|
|
|
|
def node_type_info() -> list[NodeTypeInfo]:
|
|
return [
|
|
NodeTypeInfo(
|
|
type=key,
|
|
title=spec.title,
|
|
description=spec.description,
|
|
params_schema=spec.params_schema,
|
|
has_source=spec.has_source,
|
|
free_params=spec.free_params,
|
|
plugin=spec.plugin,
|
|
)
|
|
for key, spec in NODE_TYPES.items()
|
|
]
|
|
|
|
|
|
def _load_function(flow: str, node_id: str, code: str) -> Callable[..., Any]:
|
|
"""Compile a node's source and return the function to run.
|
|
|
|
A node file defines ``process(...)``; if it defines exactly one public
|
|
function under another name, that one is used.
|
|
"""
|
|
digest = hashlib.md5(code.encode()).hexdigest()[:8]
|
|
module_name = f"_fluksio_node_{flow}_{node_id}_{digest}"
|
|
|
|
module = ModuleType(module_name)
|
|
module.__dict__["__name__"] = module_name
|
|
sys.modules[module_name] = module
|
|
exec(compile(code, f"<node {flow}.{node_id}>", "exec"), module.__dict__)
|
|
|
|
if callable(getattr(module, "process", None)):
|
|
return cast(Callable[..., Any], module.process)
|
|
|
|
functions = [
|
|
value
|
|
for name, value in vars(module).items()
|
|
if callable(value)
|
|
and not name.startswith("_")
|
|
and getattr(value, "__module__", None) == module_name
|
|
]
|
|
if len(functions) == 1:
|
|
return cast(Callable[..., Any], functions[0])
|
|
raise ValueError(
|
|
"Define a function named 'process' — this file has "
|
|
f"{len(functions)} functions to choose from."
|
|
)
|
|
|
|
|
|
class FlowController:
|
|
"""Owns the running pipeline and keeps it in step with the store."""
|
|
|
|
def __init__(
|
|
self,
|
|
store: FlowStore,
|
|
state: StateBackend | None = None,
|
|
events: EventBus | None = None,
|
|
max_workers: int | None = None,
|
|
fastapi_app: FastAPI | None = None,
|
|
execution: ExecutionService | None = None,
|
|
alerts: AlertManager | None = None,
|
|
) -> None:
|
|
self.store = store
|
|
self.state = state if state is not None else MemoryState()
|
|
self.events = events
|
|
self.max_workers = max_workers
|
|
self.app = fastapi_app
|
|
# Without one, every trigger runs inline where it was raised.
|
|
self.execution = execution
|
|
self.alerts = alerts
|
|
# Set by the lifespan once the store exists; the canvas asks the
|
|
# controller which dashboards are wired into a flow.
|
|
self.dashboards: Any = None
|
|
|
|
self.pipeline: Pipeline | None = None
|
|
self.loaded: dict[str, LoadedNode] = {}
|
|
self.issues: list[ValidationIssue] = []
|
|
self.disabled: set[str] = set()
|
|
self.supervisor = Supervisor(events)
|
|
self.history_limits: dict[str, int] = {}
|
|
self._lock = asyncio.Lock()
|
|
|
|
# -------------------------------------------------------------------------
|
|
# Lifecycle
|
|
# -------------------------------------------------------------------------
|
|
|
|
async def start(self) -> None:
|
|
# Build first: the consumer must have a pipeline to execute against
|
|
# before it claims anything, or work waiting from the last run would be
|
|
# taken and dropped — which is the very case the queue exists for.
|
|
await self.reload()
|
|
if self.execution is not None:
|
|
self.execution.start()
|
|
|
|
async def stop(self) -> None:
|
|
await self._teardown()
|
|
if self.execution is not None:
|
|
await run_in_threadpool(self.execution.stop)
|
|
|
|
async def set_enabled(self, flow: str, enabled: bool) -> None:
|
|
"""Stop or start one flow. Rebuilding is what applies it."""
|
|
await run_in_threadpool(self.store.write_enabled, flow, enabled)
|
|
await self.reload()
|
|
|
|
async def reload(self) -> None:
|
|
"""Rebuild the whole pipeline from what is currently stored."""
|
|
async with self._lock:
|
|
# Work already claimed belongs to the pipeline it was claimed
|
|
# against; let it finish there before swapping the graph out.
|
|
if self.execution is not None:
|
|
await run_in_threadpool(self.execution.pause_intake)
|
|
await self._teardown()
|
|
# A fresh supervisor per build, so a flow quarantined by the last
|
|
# one gets another chance once its author has changed something.
|
|
self.supervisor = Supervisor(self.events)
|
|
|
|
published = self.store.read_all()
|
|
self.disabled = {
|
|
flow.name
|
|
for flow in published
|
|
if not self.store.read_enabled(flow.name)
|
|
}
|
|
nodes, loaded, initial_values, flow_inputs = self._build_flows(
|
|
[(flow, False) for flow in published]
|
|
)
|
|
|
|
self.loaded = loaded
|
|
self.pipeline = Pipeline(
|
|
nodes=nodes,
|
|
state=self.state,
|
|
events=self.events,
|
|
max_workers=self.max_workers,
|
|
initial_values=initial_values,
|
|
disabled_flows=self.disabled,
|
|
work_queue=self.execution.queue if self.execution else None,
|
|
node_pool=self.execution.node_pool if self.execution else None,
|
|
)
|
|
self.pipeline.history_limits = self.history_limits
|
|
if self.execution is not None:
|
|
self.execution.bind(self.pipeline)
|
|
self.issues = _collect_issues(loaded, self.pipeline, flow_inputs)
|
|
|
|
await self._activate()
|
|
|
|
if self.execution is not None:
|
|
self.execution.resume_intake()
|
|
|
|
self._publish(
|
|
{
|
|
"type": "pipeline_rebuilt",
|
|
"issues": [issue.model_dump() for issue in self.issues],
|
|
"nodes": [status.model_dump() for status in self.node_statuses()],
|
|
# A rebuild is a fresh pipeline, so nothing is paused any more.
|
|
"paused": self.paused_flows(),
|
|
}
|
|
)
|
|
|
|
async def _teardown(self) -> None:
|
|
"""Stop everything the previous pipeline started."""
|
|
for entry in self.loaded.values():
|
|
node = entry.node
|
|
if node is None:
|
|
continue
|
|
try:
|
|
await node.stop(self.app)
|
|
except Exception:
|
|
logger.exception("Error stopping node '%s'", entry.id)
|
|
# After the nodes, so a loop still winding down is not restarted.
|
|
await self.supervisor.cancel_all()
|
|
|
|
async def _activate(self) -> None:
|
|
"""Start subscriptions, schedules and webhooks of the new pipeline.
|
|
|
|
What that means per node type is the node's own business — a connector
|
|
written against the contract starts here the same way the built-ins do.
|
|
"""
|
|
for entry in self.loaded.values():
|
|
node = entry.node
|
|
if node is None:
|
|
continue
|
|
# A stopped flow gets no subscriptions, schedules or webhooks —
|
|
# that is what stopping it means.
|
|
if entry.flow in self.disabled:
|
|
continue
|
|
node.supervisor = self.supervisor
|
|
try:
|
|
await node.start(self.app)
|
|
except Exception as exc:
|
|
logger.exception("Error starting node '%s'", entry.id)
|
|
entry.status = NodeStatus.ERROR
|
|
entry.error = f"{type(exc).__name__}: {exc}"
|
|
|
|
# -------------------------------------------------------------------------
|
|
# Building
|
|
# -------------------------------------------------------------------------
|
|
|
|
def _build_flows(
|
|
self, flows: list[tuple[FlowDef, bool]]
|
|
) -> tuple[list[Node], dict[str, LoadedNode], dict[str, Any], dict[str, bool]]:
|
|
"""Instantiate the nodes of several flows, each published or draft."""
|
|
nodes: list[Node] = []
|
|
loaded: dict[str, LoadedNode] = {}
|
|
initial_values: dict[str, Any] = {}
|
|
# Declared flow inputs, mapped to whether they start with a value.
|
|
flow_inputs: dict[str, bool] = {}
|
|
|
|
for flow, draft in flows:
|
|
for node_def in flow.nodes:
|
|
entry = self._build_node(flow.name, node_def, draft=draft)
|
|
loaded[entry.id] = entry
|
|
if entry.node is not None:
|
|
nodes.append(entry.node)
|
|
for flow_input in flow.inputs:
|
|
name = qualify(flow.name, flow_input.spec.name)
|
|
if not name:
|
|
continue
|
|
flow_inputs[name] = flow_input.initial is not None
|
|
if flow_input.initial is not None:
|
|
initial_values[name] = flow_input.initial
|
|
|
|
return nodes, loaded, initial_values, flow_inputs
|
|
|
|
def _build_node(
|
|
self, flow: str, node_def: NodeDef, draft: bool = False
|
|
) -> LoadedNode:
|
|
node_id = f"{flow}.{node_def.id}"
|
|
entry = LoadedNode(id=node_id, flow=flow)
|
|
try:
|
|
node_type = NODE_TYPES.get(node_def.type)
|
|
if node_type is None:
|
|
raise ValueError(f"Unknown node type '{node_def.type}'")
|
|
|
|
params = resolve_params(node_def.params)
|
|
|
|
if node_type.has_source:
|
|
# A shared node runs the library's copy, compiled once under
|
|
# the library's own name so every flow using it agrees.
|
|
if node_def.source_ref:
|
|
code = self.store.read_lib_source(node_def.source_ref)
|
|
function = _load_function(LIB_DIR, node_def.source_ref, code)
|
|
else:
|
|
code = self.store.read_node_source(flow, node_def.id, draft=draft)
|
|
function = _load_function(flow, node_def.id, code)
|
|
node = Node(
|
|
f=function,
|
|
requires=_bound(node_def.requires),
|
|
provides=_bound(node_def.provides),
|
|
params=params,
|
|
name=node_def.id,
|
|
)
|
|
else:
|
|
node = node_type.cls(
|
|
requires=_bound(node_def.requires),
|
|
provides=_bound(node_def.provides),
|
|
params=params,
|
|
name=node_def.id,
|
|
)
|
|
|
|
node.assign_flow(flow, node_def.id)
|
|
node._on_health = self._health_changed
|
|
|
|
if isinstance(node, HttpNode) and node.mode == HttpNode.Mode.TRIGGER:
|
|
# Webhooks live under their flow, away from the JSON API.
|
|
node.url = f"{HOOK_PREFIX}/{flow}/{node.url.lstrip('/')}"
|
|
|
|
entry.node = node
|
|
except (SecretNotFound, LibNotFound) as exc:
|
|
entry.status = NodeStatus.ERROR
|
|
entry.error = str(exc)
|
|
except Exception as exc:
|
|
logger.warning("Node '%s' failed to load: %s", node_id, exc)
|
|
entry.status = NodeStatus.ERROR
|
|
entry.error = _short_error(exc)
|
|
return entry
|
|
|
|
# -------------------------------------------------------------------------
|
|
# Queries
|
|
# -------------------------------------------------------------------------
|
|
|
|
def node_statuses(self, flow: str | None = None) -> list[NodeStatusPublic]:
|
|
return [
|
|
NodeStatusPublic(
|
|
id=entry.id,
|
|
status=entry.status.value,
|
|
error=entry.error,
|
|
health=entry.health,
|
|
health_detail=entry.health_detail,
|
|
)
|
|
for entry in self.loaded.values()
|
|
if flow is None or entry.flow == flow
|
|
]
|
|
|
|
def _health_changed(self, node: Node, status: str, detail: str | None) -> None:
|
|
"""A node saying how its connection is doing, from any thread."""
|
|
entry = self.loaded.get(node.id)
|
|
if entry is None or (entry.health == status and entry.health_detail == detail):
|
|
return
|
|
entry.health = cast(Health, status)
|
|
entry.health_detail = detail
|
|
self._publish(
|
|
{
|
|
"type": "node_health",
|
|
"flow": node.flow,
|
|
"node": node.id,
|
|
"health": status,
|
|
"detail": detail,
|
|
}
|
|
)
|
|
|
|
def flow_issues(self, flow: str) -> list[ValidationIssue]:
|
|
return [issue for issue in self.issues if not issue.flow or issue.flow == flow]
|
|
|
|
def preview(self, name: str) -> Preview:
|
|
"""Build a flow's unpublished draft without deploying it.
|
|
|
|
The draft stands in for its published self among the other flows, so
|
|
cross-flow messages resolve the way they would after publishing. The
|
|
graph is thrown away afterwards and seeded into a throwaway state, so
|
|
nothing here touches what is running.
|
|
"""
|
|
flows: list[tuple[FlowDef, bool]] = [
|
|
(flow, False) for flow in self.store.read_all() if flow.name != name
|
|
]
|
|
try:
|
|
flows.append((self.store.read_flow(name, draft=True), True))
|
|
except FlowNotFound:
|
|
return Preview(nodes=[], issues=[])
|
|
|
|
nodes, loaded, initial_values, flow_inputs = self._build_flows(flows)
|
|
pipeline = Pipeline(
|
|
nodes=nodes,
|
|
state=MemoryState(),
|
|
max_workers=self.max_workers,
|
|
initial_values=initial_values,
|
|
)
|
|
issues = _collect_issues(loaded, pipeline, flow_inputs)
|
|
return Preview(
|
|
nodes=[
|
|
NodeStatusPublic(
|
|
id=entry.id, status=entry.status.value, error=entry.error
|
|
)
|
|
for entry in loaded.values()
|
|
if entry.flow == name
|
|
],
|
|
issues=[issue for issue in issues if not issue.flow or issue.flow == name],
|
|
)
|
|
|
|
def compile_check(self, flow: str, node_id: str, code: str) -> str | None:
|
|
"""Does this source load? Returns what to show the author, or None."""
|
|
try:
|
|
_load_function(flow, node_id, code)
|
|
except Exception as exc:
|
|
return _short_error(exc)
|
|
return None
|
|
|
|
def values(self, flow: str | None = None) -> dict[str, dict[str, Any]]:
|
|
return self.pipeline.values(flow) if self.pipeline else {}
|
|
|
|
def get_node(self, node_id: str) -> Node | None:
|
|
entry = self.loaded.get(node_id)
|
|
return entry.node if entry else None
|
|
|
|
# -------------------------------------------------------------------------
|
|
# Execution
|
|
# -------------------------------------------------------------------------
|
|
|
|
def is_enabled(self, flow: str) -> bool:
|
|
return flow not in self.disabled
|
|
|
|
def is_paused(self, flow: str) -> bool:
|
|
return self.pipeline is not None and flow in self.pipeline.paused_flows()
|
|
|
|
def is_quarantined(self, flow: str) -> bool:
|
|
return flow in self.supervisor.quarantined
|
|
|
|
@property
|
|
def quarantined(self) -> set[str]:
|
|
return self.supervisor.quarantined
|
|
|
|
def paused_flows(self) -> list[str]:
|
|
return self.pipeline.paused_flows() if self.pipeline else []
|
|
|
|
def pause_flow(self, flow: str) -> None:
|
|
if self.pipeline is not None:
|
|
self.pipeline.pause(flow)
|
|
|
|
def resume_flow(self, flow: str) -> None:
|
|
"""Blocking — call from a worker thread: held-back nodes run on resume."""
|
|
if self.pipeline is None:
|
|
return
|
|
self.pipeline.resume(flow)
|
|
if self.execution is not None:
|
|
# Whatever arrived while the flow was held is queued again, oldest
|
|
# first, so a pause loses nothing.
|
|
for item in self.execution.queue.unpark(flow):
|
|
self.execution.queue.add(item)
|
|
|
|
def set_history_limits(self, limits: dict[str, int]) -> None:
|
|
"""How deep to keep each charted message's series. Applies at once."""
|
|
self.history_limits = limits
|
|
if self.pipeline is not None:
|
|
self.pipeline.history_limits = limits
|
|
|
|
def message_catalog(self) -> list[Any]:
|
|
"""Every message the published flows declare, with its last value.
|
|
|
|
What a dashboard picks from, so it spans flows rather than sitting
|
|
inside one.
|
|
"""
|
|
from app.api.routes.messages import MessageInfo
|
|
|
|
specs: dict[str, MessageSpec] = {}
|
|
providers: dict[str, list[str]] = {}
|
|
for flow in self.store.read_all():
|
|
for node in flow.nodes:
|
|
for spec in _bound(node.provides):
|
|
name = qualify(flow.name, spec.name)
|
|
specs.setdefault(name, spec)
|
|
providers.setdefault(name, []).append(f"{flow.name}.{node.id}")
|
|
for spec in _bound(node.requires):
|
|
specs.setdefault(qualify(flow.name, spec.name), spec)
|
|
for declared in flow.inputs:
|
|
if declared.spec.name:
|
|
specs.setdefault(
|
|
qualify(flow.name, declared.spec.name), declared.spec
|
|
)
|
|
|
|
values = self.values()
|
|
infos = []
|
|
for name, spec in sorted(specs.items()):
|
|
current = values.get(name, {})
|
|
infos.append(
|
|
MessageInfo(
|
|
name=name,
|
|
flow=flow_of(name),
|
|
dtype=spec.dtype.value,
|
|
providers=sorted(providers.get(name, [])),
|
|
writable=True,
|
|
numeric=spec.dtype.value in ("float", "int"),
|
|
value=current.get("value"),
|
|
ts=current.get("ts"),
|
|
)
|
|
)
|
|
return infos
|
|
|
|
def cross_flow_nodes(
|
|
self, flow: str
|
|
) -> list[tuple[str, str, list[str], list[str]]]:
|
|
"""Nodes in other flows on the far side of a dotted message name.
|
|
|
|
Returns ``(flow, node id, messages of `flow` it provides, ones it
|
|
requires)``. This is what link-in and link-out mean: the wiring is
|
|
already real — the canvas simply had no way to draw the other end.
|
|
"""
|
|
prefix = f"{flow}."
|
|
found = []
|
|
for other in self.store.read_all():
|
|
if other.name == flow:
|
|
continue
|
|
for node in other.nodes:
|
|
provides = sorted(
|
|
qualify(other.name, spec.name)
|
|
for spec in _bound(node.provides)
|
|
if qualify(other.name, spec.name).startswith(prefix)
|
|
)
|
|
requires = sorted(
|
|
qualify(other.name, spec.name)
|
|
for spec in _bound(node.requires)
|
|
if qualify(other.name, spec.name).startswith(prefix)
|
|
)
|
|
if provides or requires:
|
|
found.append((other.name, node.id, provides, requires))
|
|
return found
|
|
|
|
def publish_message(
|
|
self, name: str, value: Any, source: ValueSource | None = None
|
|
) -> None:
|
|
"""Put a value into the graph from outside. Blocking.
|
|
|
|
Refuses a name no flow declares: the flows own the namespace, and a
|
|
message nothing reads or writes would just be a key nobody sees.
|
|
"""
|
|
if self.pipeline is None:
|
|
raise KeyError(name)
|
|
|
|
spec = None
|
|
for flow in self.store.read_all():
|
|
for node in flow.nodes:
|
|
for candidate in [*_bound(node.provides), *_bound(node.requires)]:
|
|
if qualify(flow.name, candidate.name) == name:
|
|
spec = candidate
|
|
break
|
|
for declared in flow.inputs:
|
|
if qualify(flow.name, declared.spec.name) == name:
|
|
spec = declared.spec
|
|
if spec is None:
|
|
raise KeyError(name)
|
|
|
|
# The same check a node's output gets; a dashboard is not looser.
|
|
spec.check(value)
|
|
self.pipeline.publish({name: value}, source)
|
|
|
|
def queue_stats(self) -> dict[str, Any]:
|
|
return self.execution.stats() if self.execution is not None else {}
|
|
|
|
def forget_flow(self, flow: str) -> None:
|
|
"""Drop what a deleted flow left behind. Blocking."""
|
|
if self.execution is not None:
|
|
self.execution.queue.clear_flow(flow)
|
|
prefix = f"{flow}."
|
|
with self.state.lock():
|
|
stale = [
|
|
key
|
|
for key in self.state.keys()
|
|
# Both the messages themselves and the engine's own bookkeeping
|
|
# about them, which is keyed by message name too.
|
|
if key.startswith(prefix) or f":{prefix}" in key
|
|
]
|
|
for key in stale:
|
|
self.state.delete(key)
|
|
|
|
def run_flow(self, flow: str, inputs: dict[str, Any] | None = None) -> None:
|
|
"""Run every node of one flow. Blocking — call from a worker thread."""
|
|
if self.pipeline is None:
|
|
return
|
|
self.pipeline.run(inputs or {}, nodes=self.pipeline.flow_nodes(flow))
|
|
|
|
def run_preview(self, flow_name: str, inputs: dict[str, Any] | None = None) -> None:
|
|
"""Run a flow's unpublished draft once, as the editor shows it.
|
|
|
|
The nodes are built and run but never activated, so a draft never
|
|
acquires subscriptions, schedules or webhooks — only publishing does
|
|
that. State and events are the live ones, so values land on the canvas
|
|
exactly as they do for a published run.
|
|
"""
|
|
flow = self.store.read_flow(flow_name, draft=True)
|
|
nodes, _, initial_values, _ = self._build_flows([(flow, True)])
|
|
pipeline = Pipeline(
|
|
nodes=nodes,
|
|
state=self.state,
|
|
events=self.events,
|
|
max_workers=self.max_workers,
|
|
initial_values=initial_values,
|
|
)
|
|
pipeline.run(inputs or {})
|
|
|
|
def trigger_node(self, node_id: str, values: dict[str, Any] | None = None) -> None:
|
|
"""Feed values into one node. Blocking — call from a worker thread.
|
|
|
|
Runs here rather than through the queue: the caller is a person waiting
|
|
on the response, and wants the state it produced.
|
|
"""
|
|
node = self.get_node(node_id)
|
|
if node is None:
|
|
raise KeyError(node_id)
|
|
if node.requires and values:
|
|
node.trigger(values, durable=False)
|
|
else:
|
|
node.inject(values or {}, durable=False)
|
|
|
|
def _publish(self, event: dict[str, Any]) -> None:
|
|
if self.events is not None:
|
|
self.events.publish(event)
|
|
|
|
|
|
def _bound(specs: list[MessageSpec]) -> list[MessageSpec]:
|
|
"""Ports without a message name are not wired into the graph."""
|
|
return [spec for spec in specs if spec.name]
|
|
|
|
|
|
def _collect_issues(
|
|
loaded: dict[str, LoadedNode],
|
|
pipeline: Pipeline,
|
|
flow_inputs: dict[str, bool],
|
|
) -> list[ValidationIssue]:
|
|
"""Everything wrong with a built graph, whether it is running or not."""
|
|
issues = pipeline.validate(flow_inputs)
|
|
issues += [
|
|
ValidationIssue(
|
|
code="node_error",
|
|
message=entry.error or "This node could not be loaded.",
|
|
flow=entry.flow,
|
|
node=entry.id,
|
|
)
|
|
for entry in loaded.values()
|
|
if entry.status is NodeStatus.ERROR
|
|
]
|
|
# Webhooks are mounted without authentication, so a hook without a secret is
|
|
# open to anyone who guesses its URL. Flows written before the parameter
|
|
# existed keep running, but say so.
|
|
issues += [
|
|
ValidationIssue(
|
|
code="unauthenticated_hook",
|
|
message=(
|
|
f"Webhook '{entry.node.local_id}' has no secret — "
|
|
"anyone who knows its URL can trigger it."
|
|
),
|
|
flow=entry.flow,
|
|
node=entry.id,
|
|
)
|
|
for entry in loaded.values()
|
|
if isinstance(entry.node, HttpNode)
|
|
and entry.node.mode == HttpNode.Mode.TRIGGER
|
|
and not entry.node.secret
|
|
]
|
|
return issues
|
|
|
|
|
|
def _short_error(exc: Exception) -> str:
|
|
"""One line a node author can act on: what broke, and where in their code.
|
|
|
|
The frames in between belong to the loader rather than to the node, so only
|
|
the line in the node's own source is kept. The full traceback stays in the
|
|
server log.
|
|
"""
|
|
if isinstance(exc, SyntaxError):
|
|
# Its own message already names the compiled file, which is noise here.
|
|
return f"{type(exc).__name__}: {exc.msg} (line {exc.lineno})"
|
|
|
|
frames = [
|
|
frame
|
|
for frame in traceback.extract_tb(exc.__traceback__)
|
|
if frame.filename.startswith("<node ")
|
|
]
|
|
where = f" (line {frames[-1].lineno})" if frames else ""
|
|
return f"{type(exc).__name__}: {exc}{where}"
|