Node settings arrive as keyword arguments, not a params dict

A python node's settings are constants of its own function, so they are passed
the way its ports are: by name. The controller binds them to the compiled
function, the `params` field is gone from the worker and remote protocols, and
a setting sharing a port's name is reported as a node error rather than
shadowing it. The panel's scaffold follows suit and keeps the header in step
with both ports and settings.

The demo's `pace` moves from a flow input to a setting of the training node,
which is what it always was.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01NUb8YpL2s3gmN9WTACTt4q
This commit is contained in:
2026-08-20 17:47:45 +02:00
co-authored by Claude Opus 5
parent 2552c92a45
commit 3508713e85
21 changed files with 202 additions and 106 deletions
+1 -3
View File
@@ -6,8 +6,7 @@ Where TYPE could be BUG, FEAT, PERF, CHORE and SCOPE could be UX, UI, FLOW, NODE
Don't write temporary reasons for deferring a task in the task description (only strategical reasons should be noted). Don't write temporary reasons for deferring a task in the task description (only strategical reasons should be noted).
Deferring because out of scope is fine, but don't mention deferring than. Deferring because out of scope is fine, but don't mention deferring than.
`Deferred` holds what stays open on purpose, each with the condition that `Deferred` holds what stays open on purpose, each with the condition that should reopen it.
should reopen it.
## Open ## Open
@@ -17,7 +16,6 @@ should reopen it.
- FEAT/UI add animation to widgets; i.e. status of bars, gauges etc. should fade from one state to another. Multi-buttons (like "Mode" in the "Home" dashboard of the demo) should transition from one state to another; use inspiration for animations based on the google material guidelines - FEAT/UI add animation to widgets; i.e. status of bars, gauges etc. should fade from one state to another. Multi-buttons (like "Mode" in the "Home" dashboard of the demo) should transition from one state to another; use inspiration for animations based on the google material guidelines
- BUG when clicking "edit" in the "Home" dashboard of the demo on hub.fluksio.com, most of the panels disappear (only a handfull is left for actual edit) - BUG when clicking "edit" in the "Home" dashboard of the demo on hub.fluksio.com, most of the panels disappear (only a handfull is left for actual edit)
- CHORE/UI: the edge popover shows the same value twice — `MessageSparkline` falls through to a collapsed `ValuePreview` for a non-numeric value, and `EdgeInspector` then renders its own `ValuePreview defaultOpen` below it. Cosmetic; one of the two is redundant. - CHORE/UI: the edge popover shows the same value twice — `MessageSparkline` falls through to a collapsed `ValuePreview` for a non-numeric value, and `EdgeInspector` then renders its own `ValuePreview defaultOpen` below it. Cosmetic; one of the two is redundant.
- BUG/UI in the e.g. the "PV yield model" demo are two nodes which takes inputs that were not produced by any other node (e.g. seed, noise, samples). Their value seem sto be hard coded but unchangable in the node panel. We should only allow for static parameters in the corresponding "Settings" section of a node. Debug first where these static values came from. Then develop a concept which matches the philosophy of the project. We could a) go for the same global/local parameter thing which node red uses (Push-back from my side; this becomes hardly manageable on scale) b) only restrict to node-level static parameters (and later implement an overview of parameters as dedicated page) c) drop the "params" feature entirely and make nodes entirely parameter free (would cause users to create their own static parameters). . My personal vote is b) as it forces the atomic flow-style we want to have in the app and it makes the app truly scalable. Reusing the same parameter across multiple nodes would essentially mean having one node where the parameter is set and then just returning this parameter as an output which other nodes can then consume. This would mirror the getter/setter pattern from python. When resolving this, we could also consider dissolving the obscure "params" input to nodes entirely; i.e. static node paramters would become input variables just as any other variable that goes into a function
- FEAT/UI we should introduce a sync between the header of the python function and the node configuration; i.e. adding an input/output or static paramter would change the header of the python function and changing the python function header and vice versa - FEAT/UI we should introduce a sync between the header of the python function and the node configuration; i.e. adding an input/output or static paramter would change the header of the python function and changing the python function header and vice versa
- BUG/UI on flows like "House history" where the widget sets the range for the "draw the window" node to generate some data, the edges overlap the nodes. We should adjust the flow visualization to account for these cyclic behaviors - BUG/UI on flows like "House history" where the widget sets the range for the "draw the window" node to generate some data, the edges overlap the nodes. We should adjust the flow visualization to account for these cyclic behaviors
- BUG/UI when enlarging the code editor of a node, the code editor should enlarge to the left (node settings remain on the right) so that the code editor fills the center of the screen with the node properties available next to it - BUG/UI when enlarging the code editor of a node, the code editor should enlarge to the left (node settings remain on the right) so that the code editor fills the center of the screen with the node properties available next to it
+34 -1
View File
@@ -25,6 +25,7 @@ from app.flow.events import EventBus
from app.flow.executor import ExecutionService from app.flow.executor import ExecutionService
from app.flow.messages import MessageSpec, flow_of, qualify from app.flow.messages import MessageSpec, flow_of, qualify
from app.flow.nodes import ( from app.flow.nodes import (
RESERVED_SETTINGS,
ChangeNode, ChangeNode,
DelayNode, DelayNode,
ExecNode, ExecNode,
@@ -479,6 +480,17 @@ class FlowController:
params = resolve_params(node_def.params) params = resolve_params(node_def.params)
if node_type.has_source: if node_type.has_source:
# Settings and ports are both keyword arguments of the same
# function, so one name cannot mean both.
ports = {spec.port for spec in _bound(node_def.requires)}
clash = sorted((set(params) - RESERVED_SETTINGS) & ports)
if clash:
raise ValueError(
f"'{clash[0]}' is both an input and a setting of "
f"'{node_def.id}'. A setting is an argument like a "
"port, so rename one of them."
)
# A shared node runs the library's copy, compiled once under # A shared node runs the library's copy, compiled once under
# the library's own name so every flow using it agrees. # the library's own name so every flow using it agrees.
if node_def.source_ref: if node_def.source_ref:
@@ -544,7 +556,7 @@ class FlowController:
), ),
) )
node = Node( node = Node(
f=function, f=with_settings(function, params),
requires=_bound(node_def.requires), requires=_bound(node_def.requires),
provides=_bound(node_def.provides), provides=_bound(node_def.provides),
params=params, params=params,
@@ -1019,6 +1031,27 @@ def _bound(specs: list[MessageSpec]) -> list[MessageSpec]:
return [spec for spec in specs if spec.name] return [spec for spec in specs if spec.name]
def with_settings(
function: Callable[..., Any], params: dict[str, Any]
) -> Callable[..., Any]:
"""A node's function with its settings bound as keyword arguments.
A setting is a constant of one node's function, so it is passed the way a
port is: by name. The engine's own settings never reach the code, and the
``params`` the pipeline offers is dropped here rather than travelling to a
worker that has nothing to do with it.
"""
settings = {k: v for k, v in params.items() if k not in RESERVED_SETTINGS}
def call(
params: dict[str, Any] | None = None, # noqa: ARG001 - absorbed here
**ports: Any,
) -> Any:
return function(**ports, **settings)
return call
def _collect_issues( def _collect_issues(
loaded: dict[str, LoadedNode], loaded: dict[str, LoadedNode],
pipeline: Pipeline, pipeline: Pipeline,
+2 -1
View File
@@ -4,7 +4,7 @@ Split by the outside world each one talks to. Importing from
``app.flow.nodes`` keeps working, which is what every caller does. ``app.flow.nodes`` keeps working, which is what every caller does.
""" """
from app.flow.nodes.base import Node from app.flow.nodes.base import RESERVED_SETTINGS, Node
from app.flow.nodes.delay import DelayNode from app.flow.nodes.delay import DelayNode
from app.flow.nodes.exec import ExecNode from app.flow.nodes.exec import ExecNode
from app.flow.nodes.file import FileNode from app.flow.nodes.file import FileNode
@@ -30,6 +30,7 @@ __all__ = [
"MqttNode", "MqttNode",
"Node", "Node",
"NtfyNode", "NtfyNode",
"RESERVED_SETTINGS",
"RbeNode", "RbeNode",
"SwitchNode", "SwitchNode",
"TriggerNode", "TriggerNode",
+10 -4
View File
@@ -33,6 +33,10 @@ class NodeOutputError(TypeError):
"""A node function returned something that cannot be mapped onto ports.""" """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: class Node:
""" """
A pipeline node that wraps a function with typed inputs/outputs. A pipeline node that wraps a function with typed inputs/outputs.
@@ -50,7 +54,9 @@ class Node:
:param provides: Output messages this node produces. Can be a single Message :param provides: Output messages this node produces. Can be a single Message
or list of Messages. or list of Messages.
:type provides: MessageSpec | list[MessageSpec] :type provides: MessageSpec | list[MessageSpec]
:param params: Additional parameters passed to the function during execution. :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 :type params: dict
:param name: Optional name for the node. Defaults to function name. :param name: Optional name for the node. Defaults to function name.
:type name: str | None :type name: str | None
@@ -60,14 +66,14 @@ class Node:
:vartype synchronous: bool :vartype synchronous: bool
:example: :example:
>>> def process_temp(temperature, params): >>> def process_temp(temperature, offset):
... return {"celsius": temperature * 0.5 + 32} ... return {"celsius": temperature * 0.5 + offset}
>>> >>>
>>> temp_node = Node( >>> temp_node = Node(
... f=process_temp, ... f=process_temp,
... requires=MessageSpec(name="temperature", dtype=DType.FLOAT), ... requires=MessageSpec(name="temperature", dtype=DType.FLOAT),
... provides=MessageSpec(name="celsius", dtype=DType.FLOAT), ... provides=MessageSpec(name="celsius", dtype=DType.FLOAT),
... params={}, ... params={"offset": 32},
... ) ... )
""" """
+2 -5
View File
@@ -255,7 +255,6 @@ class RemoteWorkerHub:
node: str, node: str,
source: str, source: str,
kwargs: dict[str, Any], kwargs: dict[str, Any],
params: dict[str, Any] | None,
node_id: str, node_id: str,
timeout: float, timeout: float,
run_id: str = "", run_id: str = "",
@@ -272,7 +271,6 @@ class RemoteWorkerHub:
"node": node, "node": node,
"source": source, "source": source,
"kwargs": kwargs, "kwargs": kwargs,
"params": params or {},
"run": {"id": run_id} if run_id else None, "run": {"id": run_id} if run_id else None,
"timeout": timeout, "timeout": timeout,
}, },
@@ -344,16 +342,15 @@ class RemoteWorkerHub:
the label rather than requiring it. the label rather than requiring it.
""" """
def call(params: dict[str, Any] | None = None, **kwargs: Any) -> Any: def call(**kwargs: Any) -> Any:
if fallback is not None and self.pick(label) is None: if fallback is not None and self.pick(label) is None:
return fallback(params=params, **kwargs) return fallback(**kwargs)
return self.run( return self.run(
label, label,
flow, flow,
node, node,
source, source,
kwargs, kwargs,
params,
node_id, node_id,
timeout, timeout,
run_id=run_id, run_id=run_id,
+5 -2
View File
@@ -35,6 +35,9 @@ class NodeDef(BaseModel):
id: str id: str
type: str = "python" type: str = "python"
title: str = "" title: str = ""
#: This node's settings: constants of its function, stored with the flow.
#: A function node reads them as keyword arguments beside its ports, so a
#: setting cannot share a name with one.
params: dict[str, Any] = Field(default_factory=dict) params: dict[str, Any] = Field(default_factory=dict)
requires: list[MessageSpec] = Field(default_factory=list) requires: list[MessageSpec] = Field(default_factory=list)
provides: list[MessageSpec] = Field(default_factory=list) provides: list[MessageSpec] = Field(default_factory=list)
@@ -264,8 +267,8 @@ class NodeTypeInfo(BaseModel):
params_schema: dict[str, Any] = Field(default_factory=dict) params_schema: dict[str, Any] = Field(default_factory=dict)
has_source: bool = False has_source: bool = False
#: Whether this type takes settings beyond the ones its schema declares. #: Whether this type takes settings beyond the ones its schema declares.
#: A function node's params are its author's to name, and reach `process` #: A function node's settings are its author's to name, and reach `process`
#: as whatever they put there. #: as keyword arguments beside its ports.
free_params: bool = False free_params: bool = False
#: The package a connector came from; empty for the built-in types. #: The package a connector came from; empty for the built-in types.
plugin: str | None = None plugin: str | None = None
+1 -1
View File
@@ -31,7 +31,7 @@ LIB_DIR = "_lib"
DEFAULT_SOURCE = '''"""A new node. Return a dict keyed by your output ports.""" DEFAULT_SOURCE = '''"""A new node. Return a dict keyed by your output ports."""
def process(params): def process():
return {} return {}
''' '''
+1 -3
View File
@@ -302,9 +302,7 @@ def _handle(request: dict[str, Any], cache: dict[tuple[str, str], Any]) -> Any:
if request["op"] == "compile": if request["op"] == "compile":
return None return None
result = function( result = function(**(request.get("kwargs") or {}))
**(request.get("kwargs") or {}), params=request.get("params") or {}
)
if inspect.isgenerator(result): if inspect.isgenerator(result):
result = _drain(result) result = _drain(result)
try: try:
+1 -4
View File
@@ -379,7 +379,6 @@ class PythonWorkerPool:
node: str, node: str,
source: str, source: str,
kwargs: dict[str, Any], kwargs: dict[str, Any],
params: dict[str, Any] | None,
node_id: str, node_id: str,
timeout: float, timeout: float,
run_id: str = "", run_id: str = "",
@@ -393,7 +392,6 @@ class PythonWorkerPool:
"node": node, "node": node,
"source": source, "source": source,
"kwargs": kwargs, "kwargs": kwargs,
"params": params or {},
"run": {"id": run_id} if run_id else None, "run": {"id": run_id} if run_id else None,
}, },
timeout=timeout, timeout=timeout,
@@ -432,13 +430,12 @@ class PythonWorkerPool:
node be told apart when one of them is cancelled. node be told apart when one of them is cancelled.
""" """
def call(params: dict[str, Any] | None = None, **kwargs: Any) -> Any: def call(**kwargs: Any) -> Any:
return self.run( return self.run(
flow, flow,
node, node,
source, source,
kwargs, kwargs,
params,
node_id, node_id,
timeout, timeout,
run_id=run_id, run_id=run_id,
+2 -2
View File
@@ -169,8 +169,8 @@ async def save_flow(name: str, definition: dict[str, Any]) -> Any:
async def save_node_source(name: str, node_id: str, code: str) -> Any: async def save_node_source(name: str, node_id: str, code: str) -> Any:
"""Save a node's Python source and report whether it compiles. """Save a node's Python source and report whether it compiles.
A node defines ``process(...)``, taking one argument per input port plus A node defines ``process(...)``, taking one argument per input port and
``params``, and returns a dict keyed by output port. one per setting, and returns a dict keyed by output port.
""" """
return await _call( return await _call(
"PUT", f"/flows/{name}/nodes/{node_id}/source", json={"code": code} "PUT", f"/flows/{name}/nodes/{node_id}/source", json={"code": code}
+4 -4
View File
@@ -5,12 +5,12 @@ from app.core.config import settings
PREFIX = f"{settings.API_V1_STR}/flows" PREFIX = f"{settings.API_V1_STR}/flows"
WORKING_NODE = """ WORKING_NODE = """
def process(params): def process():
return {"reading": 21.5} return {"reading": 21.5}
""" """
BROKEN_NODE = """ BROKEN_NODE = """
def process(params): def process():
raise RuntimeError("boom") raise RuntimeError("boom")
""" """
@@ -75,7 +75,7 @@ def test_broken_node_is_reported_and_siblings_stay_active(
response = client.put( response = client.put(
f"{PREFIX}/demo/nodes/logger/source", f"{PREFIX}/demo/nodes/logger/source",
headers=superuser_token_headers, headers=superuser_token_headers,
json={"code": "def process(reading, params:\n"}, json={"code": "def process(reading:\n"},
) )
assert response.status_code == 200 assert response.status_code == 200
assert response.json()["status"] == "error" assert response.json()["status"] == "error"
@@ -98,7 +98,7 @@ def test_running_a_flow_produces_values(
client.put( client.put(
f"{PREFIX}/demo/nodes/logger/source", f"{PREFIX}/demo/nodes/logger/source",
headers=superuser_token_headers, headers=superuser_token_headers,
json={"code": "def process(reading, params):\n return {}\n"}, json={"code": "def process(reading):\n return {}\n"},
) )
response = client.post( response = client.post(
+2 -2
View File
@@ -8,8 +8,8 @@ from app.flow.messages import MessageSpec
from app.flow.schemas import FlowDef, NodeDef from app.flow.schemas import FlowDef, NodeDef
from app.flow.store import FlowStore, StaleVersion from app.flow.store import FlowStore, StaleVersion
SOURCE = "def process(params):\n return {}\n" SOURCE = "def process():\n return {}\n"
EDITED = "def process(params):\n return {'temp': 1}\n" EDITED = "def process():\n return {'temp': 1}\n"
@pytest.fixture @pytest.fixture
+1 -1
View File
@@ -8,7 +8,7 @@ from app.flow.messages import MessageSpec
from app.flow.schemas import FlowDef, NodeDef from app.flow.schemas import FlowDef, NodeDef
from app.flow.store import FlowStore, LibExists, LibNotFound from app.flow.store import FlowStore, LibExists, LibNotFound
SOURCE = "def process(params):\n return {'temp': 1}\n" SOURCE = "def process():\n return {'temp': 1}\n"
@pytest.fixture @pytest.fixture
+7 -2
View File
@@ -4,6 +4,7 @@ import sys
from typing import Any from typing import Any
from app.flow import logs from app.flow import logs
from app.flow.controller import with_settings
from app.flow.messages import DType, MessageSpec from app.flow.messages import DType, MessageSpec
from app.flow.nodes import Node from app.flow.nodes import Node
from app.flow.pipeline import Pipeline from app.flow.pipeline import Pipeline
@@ -73,7 +74,7 @@ def test_a_failing_node_reports_its_traceback():
namespace: dict[str, Any] = {} namespace: dict[str, Any] = {}
exec( exec(
compile( compile(
'def process(params):\n print("about to fail")\n' 'def process():\n print("about to fail")\n'
' raise RuntimeError("boom")\n', ' raise RuntimeError("boom")\n',
"<node demo.broken>", "<node demo.broken>",
"exec", "exec",
@@ -82,7 +83,11 @@ def test_a_failing_node_reports_its_traceback():
) )
bus = RecordingBus() bus = RecordingBus()
run_with_capture([make_node("broken", namespace["process"])], bus) # Wrapped the way the controller wraps it, so the settings a node declares
# arrive as keyword arguments and the frames match the real call.
run_with_capture(
[make_node("broken", with_settings(namespace["process"], {}))], bus
)
captured = logs_of(bus) captured = logs_of(bus)
assert len(captured) == 1 assert len(captured) == 1
+63
View File
@@ -0,0 +1,63 @@
"""A node's settings are arguments of its function, like its ports.
What distinguishes them is where the value comes from: a port carries whatever
the graph last published, a setting is a constant stored with the flow. Both
arrive by name, so one name cannot mean both.
"""
from pathlib import Path
import pytest
from app.flow.controller import FlowController
from app.flow.messages import DType, MessageSpec
from app.flow.pipeline import Pipeline
from app.flow.schemas import FlowDef, NodeDef
from app.flow.state import MemoryState
from app.flow.store import FlowStore
SOURCE = "def process(reading, factor):\n return {'scaled': reading * factor}\n"
def a_flow(**params: object) -> FlowDef:
return FlowDef(
name="house",
nodes=[
NodeDef(
id="scale",
params=dict(params),
requires=[MessageSpec(name="reading", dtype=DType.FLOAT)],
provides=[MessageSpec(name="scaled", dtype=DType.FLOAT)],
)
],
)
@pytest.fixture
def store(tmp_path: Path) -> FlowStore:
return FlowStore(tmp_path / "flows")
def test_a_setting_reaches_the_function_as_a_keyword_argument(store: FlowStore):
store.write_flow(a_flow(factor=3))
store.write_node_source("house", "scale", SOURCE)
controller = FlowController(store)
nodes, _loaded, _initial, _inputs = controller._build_flows(
[(store.read_flow("house"), False)]
)
pipeline = Pipeline(nodes=nodes, state=MemoryState())
pipeline.run({"house.reading": 2.0})
assert pipeline.values()["house.scaled"]["value"] == 6.0
def test_a_setting_named_after_a_port_is_refused(store: FlowStore):
store.write_draft(a_flow(reading=3), 0)
store.write_node_source("house", "scale", SOURCE, draft=True)
controller = FlowController(store)
preview = controller.preview("house")
assert [node.status for node in preview.nodes] == ["error"]
assert "both an input and a setting" in (preview.nodes[0].error or "")
+6 -9
View File
@@ -64,7 +64,7 @@ def test_a_call_crosses_to_the_thread_and_the_answer_comes_back(loop):
thread = call_in_thread( thread = call_in_thread(
lambda: result.update( lambda: result.update(
value=hub.run( value=hub.run(
"gpu", "flow", "node", "src", {"x": 1}, {}, "flow.node", timeout=5 "gpu", "flow", "node", "src", {"x": 1}, "flow.node", timeout=5
) )
) )
) )
@@ -93,7 +93,6 @@ def test_reports_arrive_before_the_answer_and_a_heartbeat_is_not_one(loop):
"node", "node",
"src", "src",
{}, {},
{},
"flow.node", "flow.node",
timeout=5, timeout=5,
run_id="r1", run_id="r1",
@@ -125,7 +124,7 @@ def test_a_failure_keeps_its_class_across_the_socket(loop):
def call() -> None: def call() -> None:
try: try:
hub.run("gpu", "flow", "node", "src", {}, {}, "flow.node", timeout=5) hub.run("gpu", "flow", "node", "src", {}, "flow.node", timeout=5)
except Exception as exc: except Exception as exc:
caught.append(exc) caught.append(exc)
@@ -150,7 +149,7 @@ def test_a_worker_that_goes_away_fails_the_call_rather_than_hanging(loop):
def call() -> None: def call() -> None:
try: try:
hub.run("gpu", "flow", "node", "src", {}, {}, "flow.node", timeout=30) hub.run("gpu", "flow", "node", "src", {}, "flow.node", timeout=30)
except Exception as exc: except Exception as exc:
caught.append(exc) caught.append(exc)
@@ -173,7 +172,7 @@ def test_silence_past_the_deadline_is_a_timeout(loop):
def call() -> None: def call() -> None:
try: try:
hub.run("gpu", "flow", "node", "src", {}, {}, "flow.node", timeout=0.3) hub.run("gpu", "flow", "node", "src", {}, "flow.node", timeout=0.3)
except Exception as exc: except Exception as exc:
caught.append(exc) caught.append(exc)
@@ -186,7 +185,7 @@ def test_a_label_nothing_carries_is_named_rather_than_waited_on(loop):
attach(hub, loop) attach(hub, loop)
with pytest.raises(NoWorker, match="tpu"): with pytest.raises(NoWorker, match="tpu"):
hub.run("tpu", "flow", "node", "src", {}, {}, "flow.node", timeout=5) hub.run("tpu", "flow", "node", "src", {}, "flow.node", timeout=5)
# Compiling against a machine that is not attached is not a broken node — # Compiling against a machine that is not attached is not a broken node —
# a node importing torch is correct there and missing here. # a node importing torch is correct there and missing here.
assert hub.compile("tpu", "flow", "node", "src") is None assert hub.compile("tpu", "flow", "node", "src") is None
@@ -208,9 +207,7 @@ def test_cancelling_a_run_reaches_only_that_run(loop):
def call(run_id: str) -> None: def call(run_id: str) -> None:
try: try:
hub.run( hub.run("gpu", "flow", "node", "src", {}, "flow.node", 30, run_id=run_id)
"gpu", "flow", "node", "src", {}, {}, "flow.node", 30, run_id=run_id
)
except Exception: except Exception:
pass pass
+3 -9
View File
@@ -43,9 +43,7 @@ def test_every_change_is_committed(store: FlowStore):
store.write_flow(a_flow()) store.write_flow(a_flow())
assert commit_count(store) == before + 1 assert commit_count(store) == before + 1
store.write_node_source( store.write_node_source("heating", "sensor", "def process():\n return {}\n")
"heating", "sensor", "def process(params):\n return {}\n"
)
assert commit_count(store) == before + 2 assert commit_count(store) == before + 2
@@ -65,9 +63,7 @@ def test_missing_flow_is_reported(store: FlowStore):
def test_deleting_removes_flow_and_its_nodes(store: FlowStore): def test_deleting_removes_flow_and_its_nodes(store: FlowStore):
store.write_flow(a_flow()) store.write_flow(a_flow())
store.write_node_source( store.write_node_source("heating", "sensor", "def process():\n return {}\n")
"heating", "sensor", "def process(params):\n return {}\n"
)
store.delete_flow("heating") store.delete_flow("heating")
@@ -77,9 +73,7 @@ def test_deleting_removes_flow_and_its_nodes(store: FlowStore):
def test_renaming_a_flow_carries_its_nodes(store: FlowStore): def test_renaming_a_flow_carries_its_nodes(store: FlowStore):
store.write_flow(a_flow()) store.write_flow(a_flow())
store.write_node_source( store.write_node_source("heating", "sensor", "def process():\n return {}\n")
"heating", "sensor", "def process(params):\n return {}\n"
)
renamed = store.rename_flow("heating", "warmth") renamed = store.rename_flow("heating", "warmth")
+26 -38
View File
@@ -23,18 +23,17 @@ def pool() -> Iterator[PythonWorkerPool]:
def run(pool: PythonWorkerPool, code: str, node: str = "demo", **kwargs): def run(pool: PythonWorkerPool, code: str, node: str = "demo", **kwargs):
return pool.run( return pool.run("demo", node, code, kwargs, f"demo.{node}", timeout=5)
"demo", node, code, kwargs, {"factor": 2}, f"demo.{node}", timeout=5
)
def test_a_node_returns_its_value_and_what_it_printed(pool, capsys): def test_a_node_returns_its_value_and_what_it_printed(pool, capsys):
result = run( result = run(
pool, pool,
"def process(value, params):\n" "def process(value, factor):\n"
" print('seen', value)\n" " print('seen', value)\n"
" return {'out': value * params['factor']}\n", " return {'out': value * factor}\n",
value=21, value=21,
factor=2,
) )
assert result == {"out": 42} assert result == {"out": 42}
# The proxy writes them to stdout, which is where the engine's tee is. # The proxy writes them to stdout, which is where the engine's tee is.
@@ -43,7 +42,7 @@ def test_a_node_returns_its_value_and_what_it_printed(pool, capsys):
def test_a_failure_keeps_its_class_and_points_at_the_node(pool): def test_a_failure_keeps_its_class_and_points_at_the_node(pool):
with pytest.raises(Exception) as caught: with pytest.raises(Exception) as caught:
run(pool, "def process(params):\n raise ValueError('bad input')\n") run(pool, "def process():\n raise ValueError('bad input')\n")
# The engine renders a node error as "<class>: <message>", so both have to # The engine renders a node error as "<class>: <message>", so both have to
# survive the trip. # survive the trip.
@@ -55,9 +54,9 @@ def test_a_failure_keeps_its_class_and_points_at_the_node(pool):
def test_a_node_that_kills_its_worker_is_an_ordinary_error(pool): def test_a_node_that_kills_its_worker_is_an_ordinary_error(pool):
with pytest.raises(Exception, match="worker died"): with pytest.raises(Exception, match="worker died"):
run(pool, "import os\n\n\ndef process(params):\n os._exit(1)\n") run(pool, "import os\n\n\ndef process():\n os._exit(1)\n")
assert run(pool, "def process(params):\n return {'out': 1}\n") == {"out": 1} assert run(pool, "def process():\n return {'out': 1}\n") == {"out": 1}
def test_a_node_that_runs_too_long_is_killed_and_the_pool_recovers(pool): def test_a_node_that_runs_too_long_is_killed_and_the_pool_recovers(pool):
@@ -66,8 +65,7 @@ def test_a_node_that_runs_too_long_is_killed_and_the_pool_recovers(pool):
pool.run( pool.run(
"demo", "demo",
"slow", "slow",
"import time\n\n\ndef process(params):\n time.sleep(30)\n", "import time\n\n\ndef process():\n time.sleep(30)\n",
{},
{}, {},
"demo.slow", "demo.slow",
timeout=1, timeout=1,
@@ -75,7 +73,7 @@ def test_a_node_that_runs_too_long_is_killed_and_the_pool_recovers(pool):
assert time.monotonic() - started < 10 assert time.monotonic() - started < 10
# The killed worker's slot is refilled on the next call. # The killed worker's slot is refilled on the next call.
assert run(pool, "def process(params):\n return {'out': 2}\n") == {"out": 2} assert run(pool, "def process():\n return {'out': 2}\n") == {"out": 2}
def test_a_running_node_can_be_cancelled(pool): def test_a_running_node_can_be_cancelled(pool):
@@ -92,8 +90,7 @@ def test_a_running_node_can_be_cancelled(pool):
pool.run( pool.run(
"demo", "demo",
"slow", "slow",
"import time\n\n\ndef process(params):\n time.sleep(30)\n", "import time\n\n\ndef process():\n time.sleep(30)\n",
{},
{}, {},
"demo.slow", "demo.slow",
timeout=30, timeout=30,
@@ -104,12 +101,12 @@ def test_a_running_node_can_be_cancelled(pool):
def test_a_result_that_is_not_json_is_refused(pool): def test_a_result_that_is_not_json_is_refused(pool):
with pytest.raises(Exception, match="cannot be sent back as JSON"): with pytest.raises(Exception, match="cannot be sent back as JSON"):
run(pool, "def process(params):\n return {'out': {1, 2}}\n") run(pool, "def process():\n return {'out': {1, 2}}\n")
def test_compiling_reports_where_the_source_is_wrong(pool): def test_compiling_reports_where_the_source_is_wrong(pool):
assert pool.compile("demo", "broken", "def process(params)\n return {}\n") assert pool.compile("demo", "broken", "def process()\n return {}\n")
assert pool.compile("demo", "fine", "def process(params):\n return {}\n") is None assert pool.compile("demo", "fine", "def process():\n return {}\n") is None
def test_a_node_imports_the_standard_library_not_the_engines_own_modules(pool): def test_a_node_imports_the_standard_library_not_the_engines_own_modules(pool):
@@ -118,7 +115,7 @@ def test_a_node_imports_the_standard_library_not_the_engines_own_modules(pool):
result = run( result = run(
pool, pool,
"import queue\nimport secrets\n\n\n" "import queue\nimport secrets\n\n\n"
"def process(params):\n" "def process():\n"
" return {'out': [queue.Queue().qsize(), len(secrets.token_hex(4))]}\n", " return {'out': [queue.Queue().qsize(), len(secrets.token_hex(4))]}\n",
) )
assert result == {"out": [0, 8]} assert result == {"out": [0, 8]}
@@ -134,7 +131,7 @@ def test_the_engines_secrets_are_not_in_a_workers_environment(pool, monkeypatch)
result = run( result = run(
pool, pool,
"import os\n\n\n" "import os\n\n\n"
"def process(params):\n" "def process():\n"
" return {'out': [k for k in ('SECRET_KEY', 'POSTGRES_PASSWORD',\n" " return {'out': [k for k in ('SECRET_KEY', 'POSTGRES_PASSWORD',\n"
" 'FLUKSIO_HARMLESS') if k in os.environ]}\n", " 'FLUKSIO_HARMLESS') if k in os.environ]}\n",
) )
@@ -152,8 +149,7 @@ def test_a_pool_can_stop_while_a_node_is_running(pool):
pool.run( pool.run(
"demo", "demo",
node, node,
"import time\n\n\ndef process(params):\n time.sleep(60)\n", "import time\n\n\ndef process():\n time.sleep(60)\n",
{},
{}, {},
f"demo.{node}", f"demo.{node}",
timeout=60, timeout=60,
@@ -177,7 +173,7 @@ def test_a_pool_can_stop_while_a_node_is_running(pool):
assert len(outcomes) == 2 assert len(outcomes) == 2
with pytest.raises(Exception, match="shutting down"): with pytest.raises(Exception, match="shutting down"):
run(pool, "def process(params):\n return {'out': 1}\n") run(pool, "def process():\n return {'out': 1}\n")
# ----------------------------------------------------------------------------- # -----------------------------------------------------------------------------
@@ -190,12 +186,11 @@ def test_a_generator_node_publishes_each_yield_and_returns_the_end(pool):
result = pool.run( result = pool.run(
"demo", "demo",
"train", "train",
"def process(params):\n" "def process():\n"
" for step in range(3):\n" " for step in range(3):\n"
" yield {'loss': 1.0 / (step + 1)}\n" " yield {'loss': 1.0 / (step + 1)}\n"
" return {'weights': 'w', 'final_loss': 0.25}\n", " return {'weights': 'w', 'final_loss': 0.25}\n",
{}, {},
{},
"demo.train", "demo.train",
timeout=5, timeout=5,
run_id="r1", run_id="r1",
@@ -219,12 +214,11 @@ def test_without_a_return_the_last_yield_is_the_result(pool):
result = pool.run( result = pool.run(
"demo", "demo",
"count", "count",
"def process(params):\n" "def process():\n"
" yield {'out': 1}\n" " yield {'out': 1}\n"
" yield {'out': 2}\n" " yield {'out': 2}\n"
" yield {'out': 3}\n", " yield {'out': 3}\n",
{}, {},
{},
"demo.count", "demo.count",
timeout=5, timeout=5,
on_event=seen.append, on_event=seen.append,
@@ -242,14 +236,13 @@ def test_emit_reaches_the_same_ports_from_inside_a_callback(pool):
"demo", "demo",
"fit", "fit",
"import fluksio\n" "import fluksio\n"
"def process(params):\n" "def process():\n"
" def on_epoch(n):\n" " def on_epoch(n):\n"
" fluksio.emit(loss=1.0 / (n + 1))\n" " fluksio.emit(loss=1.0 / (n + 1))\n"
" for epoch in range(2):\n" " for epoch in range(2):\n"
" on_epoch(epoch)\n" " on_epoch(epoch)\n"
" return {'done': True}\n", " return {'done': True}\n",
{}, {},
{},
"demo.fit", "demo.fit",
timeout=5, timeout=5,
on_event=seen.append, on_event=seen.append,
@@ -261,7 +254,7 @@ def test_emit_reaches_the_same_ports_from_inside_a_callback(pool):
def test_a_plain_function_still_just_returns(pool): def test_a_plain_function_still_just_returns(pool):
seen = [] seen = []
assert run(pool, "def process(params):\n return {'out': 7}\n") == {"out": 7} assert run(pool, "def process():\n return {'out': 7}\n") == {"out": 7}
assert seen == [] assert seen == []
@@ -272,13 +265,12 @@ def test_events_hold_off_the_timeout_but_silence_does_not(pool):
"demo", "demo",
"slow", "slow",
"import time\n" "import time\n"
"def process(params):\n" "def process():\n"
" for step in range(12):\n" " for step in range(12):\n"
" time.sleep(0.05)\n" " time.sleep(0.05)\n"
" yield {'beat': step}\n" " yield {'beat': step}\n"
" return {'done': True}\n", " return {'done': True}\n",
{}, {},
{},
"demo.slow", "demo.slow",
timeout=0.3, timeout=0.3,
run_id="r2", run_id="r2",
@@ -290,8 +282,7 @@ def test_events_hold_off_the_timeout_but_silence_does_not(pool):
pool.run( pool.run(
"demo", "demo",
"quiet", "quiet",
"import time\ndef process(params):\n time.sleep(2)\n return {}\n", "import time\ndef process():\n time.sleep(2)\n return {}\n",
{},
{}, {},
"demo.quiet", "demo.quiet",
timeout=0.3, timeout=0.3,
@@ -309,8 +300,7 @@ def test_cancelling_one_run_leaves_the_same_node_in_another_alone(pool):
pool.run( pool.run(
"demo", "demo",
"hold", "hold",
"import time\ndef process(params):\n time.sleep(5)\n return {}\n", "import time\ndef process():\n time.sleep(5)\n return {}\n",
{},
{}, {},
"demo.hold", "demo.hold",
timeout=10, timeout=10,
@@ -344,10 +334,9 @@ def test_a_node_saves_and_loads_an_artifact(tmp_path):
"demo", "demo",
"save", "save",
"import fluksio\n" "import fluksio\n"
"def process(params):\n" "def process():\n"
" return {'weights': fluksio.save_artifact(b'x' * 2048, 'w.npz')}\n", " return {'weights': fluksio.save_artifact(b'x' * 2048, 'w.npz')}\n",
{}, {},
{},
"demo.save", "demo.save",
timeout=10, timeout=10,
)["weights"] )["weights"]
@@ -360,11 +349,10 @@ def test_a_node_saves_and_loads_an_artifact(tmp_path):
"demo", "demo",
"load", "load",
"import fluksio\n" "import fluksio\n"
"def process(weights, params):\n" "def process(weights):\n"
" with open(fluksio.load_artifact(weights), 'rb') as f:\n" " with open(fluksio.load_artifact(weights), 'rb') as f:\n"
" return {'size': len(f.read())}\n", " return {'size': len(f.read())}\n",
{"weights": ref}, {"weights": ref},
{},
"demo.load", "demo.load",
timeout=10, timeout=10,
) )
+27 -11
View File
@@ -73,6 +73,9 @@ const SECTION = PANEL_SECTION
/** Settings the engine reads itself, so they are not the author's to name. */ /** Settings the engine reads itself, so they are not the author's to name. */
const RESERVED_PARAMS = new Set(["synchronous"]) const RESERVED_PARAMS = new Set(["synchronous"])
/** A setting reaches the function by name, so the name has to be one. */
const IDENTIFIER = /^[A-Za-z_]\w*$/
/** /**
* A text field that offers what is already in use elsewhere. * A text field that offers what is already in use elsewhere.
* *
@@ -421,7 +424,9 @@ function FreeParamsForm({
const entries = Object.entries(params).filter(([key]) => !reserved.has(key)) const entries = Object.entries(params).filter(([key]) => !reserved.has(key))
const rename = (from: string, to: string) => { const rename = (from: string, to: string) => {
if (to === from) return // A setting is an argument of `process`, so a name it cannot take is not
// a rename — it is a typo on its way to a node that will not load.
if (to === from || !IDENTIFIER.test(to)) return
// Rebuilt rather than patched, so the settings keep the order they were // Rebuilt rather than patched, so the settings keep the order they were
// typed in instead of jumping around as one is renamed. // typed in instead of jumping around as one is renamed.
const next: Record<string, unknown> = {} const next: Record<string, unknown> = {}
@@ -453,7 +458,8 @@ function FreeParamsForm({
{entries.length === 0 ? ( {entries.length === 0 ? (
<p className="text-sm text-muted-foreground"> <p className="text-sm text-muted-foreground">
Values your code reads from <code>params</code>. Constants of this node, passed to <code>process</code> by name like
its inputs.
</p> </p>
) : null} ) : null}
@@ -855,7 +861,7 @@ const SCAFFOLD_SHAPE = (() => {
const value = Object.values(PLACEHOLDER).map(quote).join("|") const value = Object.values(PLACEHOLDER).map(quote).join("|")
const entry = `"[^"]+": (?:${value})` const entry = `"[^"]+": (?:${value})`
return new RegExp( return new RegExp(
`^${quote(SCAFFOLD_DOC)}\\n\\n\\ndef process\\(\\w+(?:, \\w+)*\\):\\n return \\{(?:${entry}(?:, ${entry})*)?\\}\\n$`, `^${quote(SCAFFOLD_DOC)}\\n\\n\\ndef process\\((?:\\w+(?:, \\w+)*)?\\):\\n return \\{(?:${entry}(?:, ${entry})*)?\\}\\n$`,
) )
})() })()
@@ -863,20 +869,30 @@ const SCAFFOLD_SHAPE = (() => {
const portName = (spec: MessageSpec) => const portName = (spec: MessageSpec) =>
spec.port || (spec.name ?? "").split(".").pop() || "" spec.port || (spec.name ?? "").split(".").pop() || ""
/** A `process` that takes this node's inputs and returns its outputs. */ /**
* A `process` that takes this node's inputs and settings, and returns its
* outputs.
*
* Both arrive by name, so both are arguments the ports first, in the order
* they are declared, then the settings.
*/
function scaffoldFor(node: NodeDef_Input): string { function scaffoldFor(node: NodeDef_Input): string {
const args = [ const args = [
...new Set( ...new Set(
(node.requires ?? []) [
.map(portName) ...(node.requires ?? []).map(portName),
// Anything else cannot be a keyword argument, so it cannot be a port. ...Object.keys(node.params ?? {}).filter(
.filter((port) => /^[A-Za-z_]\w*$/.test(port) && port !== "params"), (name) => !RESERVED_PARAMS.has(name),
),
]
// Anything else cannot be a keyword argument, so it cannot be one.
.filter((name) => IDENTIFIER.test(name)),
), ),
] ]
const returns = (node.provides ?? []) const returns = (node.provides ?? [])
.filter((spec) => portName(spec)) .filter((spec) => portName(spec))
.map((spec) => `"${portName(spec)}": ${PLACEHOLDER[spec.dtype ?? "float"]}`) .map((spec) => `"${portName(spec)}": ${PLACEHOLDER[spec.dtype ?? "float"]}`)
return `${SCAFFOLD_DOC}\n\n\ndef process(${[...args, "params"].join( return `${SCAFFOLD_DOC}\n\n\ndef process(${args.join(
", ", ", ",
)}):\n return {${returns.join(", ")}}\n` )}):\n return {${returns.join(", ")}}\n`
} }
@@ -998,13 +1014,13 @@ function PanelBody({
type={node.type} type={node.type}
schema={nodeType?.params_schema} schema={nodeType?.params_schema}
params={node.params ?? {}} params={node.params ?? {}}
onChange={(params) => onChange({ ...node, params })} onChange={(params) => editNode({ ...node, params })}
/> />
{nodeType?.free_params ? ( {nodeType?.free_params ? (
<FreeParamsForm <FreeParamsForm
params={node.params ?? {}} params={node.params ?? {}}
reserved={RESERVED_PARAMS} reserved={RESERVED_PARAMS}
onChange={(params) => onChange({ ...node, params })} onChange={(params) => editNode({ ...node, params })}
/> />
) : null} ) : null}
{hasSource ? ( {hasSource ? (
+2 -2
View File
@@ -101,12 +101,12 @@ test("running a flow puts values on its edges", async ({ page }) => {
await setNodeSource( await setNodeSource(
page, page,
"python", "python",
'def process(params):\n return {"reading": 42.0}\n', 'def process():\n return {"reading": 42.0}\n',
) )
await setNodeSource( await setNodeSource(
page, page,
"python_2", "python_2",
"def process(reading, params):\n return {}\n", "def process(reading):\n return {}\n",
) )
await page.reload() await page.reload()
+2 -2
View File
@@ -12,11 +12,11 @@ test.use({ storageState: "playwright/.auth/user.json" })
test.describe.configure({ mode: "serial" }) test.describe.configure({ mode: "serial" })
const PRINTING_NODE = `def process(params): const PRINTING_NODE = `def process():
print("sensor read 21.5 degrees") print("sensor read 21.5 degrees")
return {"reading": 21.5} return {"reading": 21.5}
` `
const BROKEN_NODE = `def process(reading, params): const BROKEN_NODE = `def process(reading):
raise RuntimeError("downstream blew up") raise RuntimeError("downstream blew up")
` `