Melvin Strobl
cb5e8f0ecc
Connectors are the device-facing node class third parties write, so the surface they build against is versioned and documented: ConnectorNode carries a declared contract version, a polling loop that publishes only what changed and reports health around it, and parameters whose credential fields are marked x-secret so the editor offers the secrets store instead of a text box. They are found through the fluksio.node_types entry point group, with the package's own metadata as the manifest. docs/connectors/ has the contract and the authoring guide; connector-skeleton/ is a working one to copy. The controller no longer knows what any node type is: start, stop and report_health are protocol methods on Node, and the built-ins were migrated to them first, so the hooks a connector implements are the ones the engine has been driving all along. Marking a node reusable moves its source to _lib/ and points the node at it by name. Other flows instantiate it with their own ports and settings, one fix reaches all of them, and a shared source still in use cannot be deleted. Ports gained an interval: an output publishes, and an input wakes its node, at most every n seconds. State keeps the latest value, so only the delivery is skipped, and pressing Run is never throttled. Also fixes autosave sending no version on its first save of a session, which made every flow saved more than once conflict with itself.
158 lines
3.8 KiB
Python
158 lines
3.8 KiB
Python
"""The persisted shape of a flow, shared by the store, the API and the editor.
|
|
|
|
A flow is structure plus code: this module is the structure. Node logic for
|
|
``python`` nodes lives beside it as a plain ``.py`` file.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import re
|
|
from typing import Any, Literal
|
|
|
|
from pydantic import BaseModel, Field, field_validator
|
|
|
|
from app.flow.messages import MessageSpec
|
|
|
|
NAME_PATTERN = re.compile(r"^[a-z][a-z0-9_]*$")
|
|
|
|
|
|
def _validate_name(value: str) -> str:
|
|
if not NAME_PATTERN.match(value):
|
|
raise ValueError(
|
|
"Use lowercase letters, digits and underscores, starting with a letter"
|
|
)
|
|
return value
|
|
|
|
|
|
class Position(BaseModel):
|
|
"""Where a node sits on the canvas."""
|
|
|
|
x: float = 0
|
|
y: float = 0
|
|
|
|
|
|
class NodeDef(BaseModel):
|
|
"""A node as stored: identity, placement, configuration and ports."""
|
|
|
|
id: str
|
|
type: str = "python"
|
|
title: str = ""
|
|
position: Position = Position()
|
|
params: dict[str, Any] = Field(default_factory=dict)
|
|
requires: list[MessageSpec] = Field(default_factory=list)
|
|
provides: list[MessageSpec] = Field(default_factory=list)
|
|
#: Name of a shared source in the library, instead of this node's own file.
|
|
#: Editing it edits the copy every flow using it runs.
|
|
source_ref: str | None = None
|
|
|
|
@field_validator("id")
|
|
@classmethod
|
|
def _check_id(cls, value: str) -> str:
|
|
return _validate_name(value)
|
|
|
|
|
|
class FlowInput(BaseModel):
|
|
"""A message the flow starts with rather than computes."""
|
|
|
|
spec: MessageSpec
|
|
initial: Any | None = None
|
|
|
|
|
|
class FlowDef(BaseModel):
|
|
"""One atomic flow."""
|
|
|
|
name: str
|
|
title: str = ""
|
|
nodes: list[NodeDef] = Field(default_factory=list)
|
|
inputs: list[FlowInput] = Field(default_factory=list)
|
|
version: int = 1
|
|
|
|
@field_validator("name")
|
|
@classmethod
|
|
def _check_name(cls, value: str) -> str:
|
|
return _validate_name(value)
|
|
|
|
|
|
class NodeSource(BaseModel):
|
|
"""The Python source of a node."""
|
|
|
|
code: str
|
|
|
|
|
|
Health = Literal["ok", "degraded", "down"]
|
|
|
|
|
|
class NodeStatusPublic(BaseModel):
|
|
"""Whether a node loaded, and how its connection is doing."""
|
|
|
|
id: str
|
|
status: str = "active"
|
|
error: str | None = None
|
|
health: Health = "ok"
|
|
health_detail: str | None = None
|
|
|
|
|
|
class MessageValue(BaseModel):
|
|
"""The last payload seen on a message."""
|
|
|
|
value: Any = None
|
|
ts: float | None = None
|
|
|
|
|
|
class HistoryPoint(BaseModel):
|
|
"""One numeric value a message carried, and when."""
|
|
|
|
ts: float
|
|
value: float
|
|
|
|
|
|
class MessageHistory(BaseModel):
|
|
"""A message's recent numeric values, oldest first.
|
|
|
|
Only numbers are recorded, so ``numeric`` tells the panel whether an empty
|
|
series means "nothing plottable here" or "nothing has arrived yet".
|
|
"""
|
|
|
|
message: str
|
|
numeric: bool = False
|
|
points: list[HistoryPoint] = Field(default_factory=list)
|
|
|
|
|
|
class FlowSummary(BaseModel):
|
|
name: str
|
|
title: str = ""
|
|
node_count: int = 0
|
|
error_count: int = 0
|
|
has_draft: bool = False
|
|
enabled: bool = True
|
|
paused: bool = False
|
|
|
|
|
|
class FlowsPublic(BaseModel):
|
|
data: list[FlowSummary]
|
|
count: int
|
|
|
|
|
|
class LibraryNode(BaseModel):
|
|
"""A node source shared across flows, and who is using it."""
|
|
|
|
name: str
|
|
used_by: list[str] = Field(default_factory=list)
|
|
|
|
|
|
class FlowStatePublic(BaseModel):
|
|
values: dict[str, MessageValue] = Field(default_factory=dict)
|
|
nodes: list[NodeStatusPublic] = Field(default_factory=list)
|
|
|
|
|
|
class NodeTypeInfo(BaseModel):
|
|
"""A node type the editor can offer, with its parameter schema."""
|
|
|
|
type: str
|
|
title: str
|
|
description: str
|
|
params_schema: dict[str, Any] = Field(default_factory=dict)
|
|
has_source: bool = False
|
|
#: The package a connector came from; empty for the built-in types.
|
|
plugin: str | None = None
|