Fluksio/app
0
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Add the flow API: typed messages, git-backed store, REST and live events

Browse Source 
Makes the flow engine reachable from the API, which is what M3 needs before
any of it can reach the browser.

- app/flow is a package now; the prototype's watch-dir scripts and the
  matplotlib/networkx visualiser are gone with their dependencies.
- Messages carry a serializable dtype instead of a live Python type, and a
  port name, so the graph can speak qualified names while node functions keep
  local arguments. Redis state is JSON, not pickle.
- Message names are namespaced per flow ("heating.temp"); a bare name resolves
  to its own flow, a dotted one crosses flows.
- Several nodes may provide the same message: producers are a list, so fan-in
  is a real edge instead of a silently dropped one.
- Flows are stored as flow.json plus node sources in a git repository, one
  commit per save, with identical saves skipped so autosave stays quiet.
- Node failures are isolated and reported per node; validate() returns cycles
  and unconnected inputs instead of raising deep in a run.
- Credentials live in an encrypted store and are referenced as {"$secret": …}.
- Engine events reach websocket clients through a bus, so values, node status
  and execution show up live.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_016WzrvW7rjQbynnhF6pxh6i
This commit is contained in:
Melvin Strobl
2026-08-15 17:23:36 +02:00
co-authored by Claude Fable 5
parent 61be29827d
commit 06a4506767
54 changed files with 2586 additions and 4978 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/app/api/deps.py
+32 -1
View File
@@ -2,7 +2,7 @@ from collections.abc import Generator
from typing import Annotated
import jwt
from fastapi import Depends, HTTPException, status
from fastapi import Depends, HTTPException, Request, status
from fastapi.security import OAuth2PasswordBearer
from jwt.exceptions import InvalidTokenError
from pydantic import ValidationError
@@ -11,6 +11,7 @@ from sqlmodel import Session
from app.core import security
from app.core.config import settings
from app.core.db import engine
from app.flow.controller import FlowController
from app.models import TokenPayload, User
reusable_oauth2 = OAuth2PasswordBearer(
@@ -27,6 +28,24 @@ SessionDep = Annotated[Session, Depends(get_db)]
TokenDep = Annotated[str, Depends(reusable_oauth2)]
def user_from_token(session: Session, token: str) -> User | None:
"""Resolve a bearer token to its user, or None if it does not hold up.
Shared with the websocket, which cannot use the HTTP security scheme.
"""
try:
payload = jwt.decode(
token, settings.SECRET_KEY, algorithms=[security.ALGORITHM]
)
token_data = TokenPayload(**payload)
except (InvalidTokenError, ValidationError):
return None
user = session.get(User, token_data.sub)
if user is None or not user.is_active:
return None
return user
def get_current_user(session: SessionDep, token: TokenDep) -> User:
try:
payload = jwt.decode(
@@ -49,6 +68,18 @@ def get_current_user(session: SessionDep, token: TokenDep) -> User:
CurrentUser = Annotated[User, Depends(get_current_user)]
def get_flow_controller(request: Request) -> FlowController:
controller: FlowController | None = getattr(
request.app.state, "flow_controller", None
)
if controller is None:
raise HTTPException(status_code=503, detail="The flow engine is not running")
return controller
FlowControllerDep = Annotated[FlowController, Depends(get_flow_controller)]
def get_current_active_superuser(current_user: CurrentUser) -> User:
if not current_user.is_superuser:
raise HTTPException(
backend/app/api/main.py
+4 -1
View File
@@ -1,6 +1,6 @@
from fastapi import APIRouter
from app.api.routes import items, login, private, users, utils
from app.api.routes import flows, items, login, private, secrets, users, utils
from app.core.config import settings
api_router = APIRouter()
@@ -8,6 +8,9 @@ api_router.include_router(login.router)
api_router.include_router(users.router)
api_router.include_router(utils.router)
api_router.include_router(items.router)
api_router.include_router(flows.router)
api_router.include_router(flows.ws_router)
api_router.include_router(secrets.router)
if settings.ENVIRONMENT == "local":
backend/app/api/routes/flows.py
+295
View File
@@ -0,0 +1,295 @@
"""The flow API. Everything the editor can do is available here first."""
import asyncio
from typing import Any
from fastapi import (
APIRouter,
Depends,
HTTPException,
WebSocket,
WebSocketDisconnect,
)
from fastapi.concurrency import run_in_threadpool
from pydantic import BaseModel
from sqlmodel import Session
from app.api.deps import FlowControllerDep, get_current_user, user_from_token
from app.core.db import engine
from app.flow.controller import FlowController
from app.flow.events import event_bus
from app.flow.messages import qualify
from app.flow.pipeline import ValidationIssue
from app.flow.schemas import (
FlowDef,
FlowsPublic,
FlowStatePublic,
FlowSummary,
MessageValue,
NodeSource,
NodeStatusPublic,
NodeTypeInfo,
)
from app.flow.store import FlowNotFound
from app.models import Message
router = APIRouter(
prefix="/flows", tags=["flows"], dependencies=[Depends(get_current_user)]
)
# The websocket authenticates from its query string, so it stays off the
# router that enforces the HTTP bearer scheme.
ws_router = APIRouter(prefix="/flows", tags=["flows"])
class FlowDetail(BaseModel):
"""A flow plus how it is currently doing."""
definition: FlowDef
nodes: list[NodeStatusPublic] = []
issues: list[ValidationIssue] = []
class ValidationResult(BaseModel):
issues: list[ValidationIssue] = []
class RunRequest(BaseModel):
inputs: dict[str, Any] = {}
class TriggerRequest(BaseModel):
values: dict[str, Any] = {}
def _detail(controller: FlowController, definition: FlowDef) -> FlowDetail:
return FlowDetail(
definition=definition,
nodes=controller.node_statuses(definition.name),
issues=controller.flow_issues(definition.name),
)
def _read_flow(controller: FlowController, name: str) -> FlowDef:
try:
return controller.store.read_flow(name)
except FlowNotFound:
raise HTTPException(status_code=404, detail=f"No flow named '{name}'")
def _flow_state(controller: FlowController, name: str) -> FlowStatePublic:
return FlowStatePublic(
values={
key: MessageValue(**value) for key, value in controller.values(name).items()
},
nodes=controller.node_statuses(name),
)
# -----------------------------------------------------------------------------
# Flows
# -----------------------------------------------------------------------------
@router.get("/", response_model=FlowsPublic)
def read_flows(controller: FlowControllerDep) -> Any:
"""List every flow."""
summaries = []
for definition in controller.store.read_all():
statuses = controller.node_statuses(definition.name)
summaries.append(
FlowSummary(
name=definition.name,
title=definition.title,
node_count=len(definition.nodes),
error_count=sum(1 for s in statuses if s.status == "error"),
)
)
return FlowsPublic(data=summaries, count=len(summaries))
@router.get("/node-types", response_model=list[NodeTypeInfo])
def read_node_types() -> Any:
"""The node types that can be placed on a canvas."""
from app.flow.controller import node_type_info
return node_type_info()
@router.get("/{name}", response_model=FlowDetail)
def read_flow(name: str, controller: FlowControllerDep) -> Any:
"""Read one flow, with the state of its nodes."""
return _detail(controller, _read_flow(controller, name))
@router.put("/{name}", response_model=FlowDetail)
async def save_flow(
name: str,
definition: FlowDef,
controller: FlowControllerDep,
) -> Any:
"""Create or replace a flow. Saving the same content again changes nothing."""
if definition.name != name:
raise HTTPException(
status_code=400, detail="The flow name in the body must match the URL"
)
duplicates = {n.id for n in definition.nodes}
if len(duplicates) != len(definition.nodes):
raise HTTPException(status_code=400, detail="Node names must be unique")
changed = await run_in_threadpool(controller.store.write_flow, definition)
if changed:
await controller.reload()
return _detail(controller, definition)
@router.delete("/{name}", response_model=Message)
async def delete_flow(name: str, controller: FlowControllerDep) -> Any:
"""Delete a flow and everything in it."""
try:
await run_in_threadpool(controller.store.delete_flow, name)
except FlowNotFound:
raise HTTPException(status_code=404, detail=f"No flow named '{name}'")
await controller.reload()
return Message(message=f"Deleted flow '{name}'")
# -----------------------------------------------------------------------------
# Node source
# -----------------------------------------------------------------------------
@router.get("/{name}/nodes/{node_id}/source", response_model=NodeSource)
def read_node_source(
name: str,
node_id: str,
controller: FlowControllerDep,
) -> Any:
"""Read a node's Python source."""
_read_flow(controller, name)
return NodeSource(code=controller.store.read_node_source(name, node_id))
@router.put("/{name}/nodes/{node_id}/source", response_model=NodeStatusPublic)
async def save_node_source(
name: str,
node_id: str,
source: NodeSource,
controller: FlowControllerDep,
) -> Any:
"""Save a node's source and report whether it loads."""
_read_flow(controller, name)
changed = await run_in_threadpool(
controller.store.write_node_source, name, node_id, source.code
)
if changed:
await controller.reload()
status = next(
(s for s in controller.node_statuses(name) if s.id == f"{name}.{node_id}"),
None,
)
return status or NodeStatusPublic(id=f"{name}.{node_id}")
# -----------------------------------------------------------------------------
# Validation and execution
# -----------------------------------------------------------------------------
@router.post("/{name}/validate", response_model=ValidationResult)
def validate_flow(name: str, controller: FlowControllerDep) -> Any:
"""Report what would keep this flow from running."""
_read_flow(controller, name)
return ValidationResult(issues=controller.flow_issues(name))
@router.post("/{name}/run", response_model=FlowStatePublic)
async def run_flow(
name: str,
body: RunRequest,
controller: FlowControllerDep,
) -> Any:
"""Run every node of a flow once."""
_read_flow(controller, name)
inputs = {qualify(name, key): value for key, value in body.inputs.items()}
await run_in_threadpool(controller.run_flow, name, inputs)
return _flow_state(controller, name)
@router.post("/{name}/nodes/{node_id}/trigger", response_model=FlowStatePublic)
async def trigger_node(
name: str,
node_id: str,
body: TriggerRequest,
controller: FlowControllerDep,
) -> Any:
"""Feed values into a single node."""
try:
await run_in_threadpool(
controller.trigger_node, f"{name}.{node_id}", body.values
)
except KeyError:
raise HTTPException(
status_code=404, detail=f"No node named '{node_id}' in flow '{name}'"
)
return _flow_state(controller, name)
@router.get("/{name}/state", response_model=FlowStatePublic)
def read_flow_state(name: str, controller: FlowControllerDep) -> Any:
"""The last value seen on every message of this flow."""
_read_flow(controller, name)
return _flow_state(controller, name)
# -----------------------------------------------------------------------------
# Live updates
# -----------------------------------------------------------------------------
@ws_router.websocket("/ws")
async def flow_events(websocket: WebSocket, token: str = "") -> None:
"""Stream values, node status and execution events as they happen.
The token goes in the query string because browsers cannot set headers on
a websocket handshake.
"""
with Session(engine) as session:
user = user_from_token(session, token)
if user is None:
await websocket.close(code=1008)
return
await websocket.accept()
controller: FlowController | None = getattr(
websocket.app.state, "flow_controller", None
)
if controller is not None:
await websocket.send_json(
{
"type": "snapshot",
"values": controller.values(),
"nodes": [s.model_dump() for s in controller.node_statuses()],
"issues": [i.model_dump() for i in controller.issues],
}
)
async with event_bus.subscribe() as queue:
receiver = asyncio.create_task(websocket.receive_text())
try:
while True:
sender = asyncio.create_task(queue.get())
done, _ = await asyncio.wait(
{sender, receiver}, return_when=asyncio.FIRST_COMPLETED
)
if receiver in done:
# The client went away.
sender.cancel()
break
await websocket.send_json(sender.result())
except WebSocketDisconnect:
pass
finally:
receiver.cancel()
backend/app/api/routes/secrets.py
+51
View File
@@ -0,0 +1,51 @@
"""Credentials for node integrations.
Values go in and are never handed back out — the API only ever lists names.
"""
from typing import Any
from fastapi import APIRouter, Depends, HTTPException
from fastapi.concurrency import run_in_threadpool
from pydantic import BaseModel
from app.api.deps import get_current_user
from app.flow.secrets import SecretNotFound, get_secrets
from app.models import Message
router = APIRouter(
prefix="/secrets", tags=["secrets"], dependencies=[Depends(get_current_user)]
)
class SecretNames(BaseModel):
data: list[str]
count: int
class SecretValue(BaseModel):
value: str
@router.get("/", response_model=SecretNames)
def read_secrets() -> Any:
"""List the names of stored secrets."""
names = get_secrets().list()
return SecretNames(data=names, count=len(names))
@router.put("/{name}", response_model=Message)
async def save_secret(name: str, body: SecretValue) -> Any:
"""Store a secret under a name that nodes can reference."""
await run_in_threadpool(get_secrets().set, name, body.value)
return Message(message=f"Saved secret '{name}'")
@router.delete("/{name}", response_model=Message)
async def delete_secret(name: str) -> Any:
"""Delete a secret."""
try:
await run_in_threadpool(get_secrets().delete, name)
except SecretNotFound:
raise HTTPException(status_code=404, detail=f"No secret named '{name}'")
return Message(message=f"Deleted secret '{name}'")
backend/app/core/config.py
+9
View File
@@ -1,5 +1,6 @@
import secrets
import warnings
from pathlib import Path
from typing import Annotated, Any, Literal
from pydantic import (
@@ -37,6 +38,14 @@ class Settings(BaseSettings):
FRONTEND_HOST: str = "http://localhost:5173"
ENVIRONMENT: Literal["local", "staging", "production"] = "local"
# Flows live on disk as a git repository; secrets stay outside it.
FLOWS_DIR: Path = Path("flow-data/flows")
SECRETS_FILE: Path = Path("flow-data/secrets.enc")
FLOW_MAX_WORKERS: int = 4
# Without a Redis host the engine keeps its state in memory.
REDIS_HOST: str | None = None
REDIS_PORT: int = 6379
BACKEND_CORS_ORIGINS: Annotated[
list[AnyUrl] | str, BeforeValidator(parse_cors)
] = []
backend/app/flow/__init__.py
+26
View File
@@ -0,0 +1,26 @@
"""The flow engine: nodes, the pipeline that runs them, and their storage."""
from app.flow.controller import FlowController, NodeStatus
from app.flow.events import EventBus, event_bus
from app.flow.messages import DType, MessageSpec, qualify
from app.flow.nodes import Node
from app.flow.pipeline import Pipeline, ValidationIssue
from app.flow.state import MemoryState, RedisState, StateBackend
from app.flow.store import FlowStore
__all__ = [
"DType",
"EventBus",
"FlowController",
"FlowStore",
"MemoryState",
"MessageSpec",
"Node",
"NodeStatus",
"Pipeline",
"RedisState",
"StateBackend",
"ValidationIssue",
"event_bus",
"qualify",
]
backend/app/flow/controller.py
+313 -734
View File
File diff suppressed because it is too large Load Diff
backend/app/flow/dag_generator.py
-218
View File
@@ -1,218 +0,0 @@
"""
DAG generation utilities for testing pipelines.
This module provides functions to generate random directed acyclic graphs (DAGs)
with guaranteed properties like connectivity and no cycles.
"""
from __future__ import annotations
import networkx as nx
import numpy as np
from typing import Callable
from util import Message, NodeParams
from nodes import Node, MLPNode
def generate_dag(
num_nodes: int,
edge_probability: float = 0.3,
seed: int | None = None,
) -> nx.DiGraph:
"""
Generate a random DAG using the Erdős-Rényi model with topological ordering.
Creates edges only from lower-indexed to higher-indexed nodes to guarantee
acyclicity, then removes isolated nodes to ensure connectivity.
:param num_nodes: Number of nodes in the graph.
:type num_nodes: int
:param edge_probability: Probability of edge between any two nodes.
:type edge_probability: float
:param seed: Random seed for reproducibility.
:type seed: int | None
:returns: A random DAG.
:rtype: nx.DiGraph
"""
rng = np.random.default_rng(seed)
G = nx.DiGraph()
G.add_nodes_from(range(num_nodes))
# Add edges only from lower to higher index (guarantees DAG)
for i in range(num_nodes):
for j in range(i + 1, num_nodes):
if rng.random() < edge_probability:
G.add_edge(i, j)
# Ensure connectivity: connect isolated nodes
_ensure_connected(G, rng)
return G
def _ensure_connected(G: nx.DiGraph, rng: np.random.Generator) -> None:
"""
Ensure the DAG is weakly connected by adding edges.
:param G: The graph to modify in place.
:type G: nx.DiGraph
:param rng: Random number generator.
:type rng: np.random.Generator
"""
# Get weakly connected components
components = list(nx.weakly_connected_components(G))
if len(components) <= 1:
return
# Sort nodes in each component by index for DAG-safe edge addition
sorted_components = [sorted(c) for c in components]
# Connect components by adding edge from max of one to min of next
for i in range(len(sorted_components) - 1):
src = sorted_components[i][-1] # Last (highest) node in component
dst = sorted_components[i + 1][0] # First (lowest) node in next component
# Ensure edge direction maintains DAG property
if src < dst:
G.add_edge(src, dst)
else:
G.add_edge(dst, src)
def dag_to_pipeline_nodes(
G: nx.DiGraph,
node_factory: Callable[..., Node] | None = None,
params: dict | None = None,
seed: int | None = None,
) -> tuple[list[Node], list[Message], list[Node]]:
"""
Convert a networkx DAG to pipeline nodes with messages.
:param G: The DAG to convert.
:type G: nx.DiGraph
:param node_factory: Factory function to create nodes (default: MLPNode).
:type node_factory: Callable[..., Node] | None
:param params: Parameters to pass to node factory.
:type params: dict | None
:param seed: Random seed for reproducibility.
:type seed: int | None
:returns: Tuple of (nodes, all_messages, trigger_nodes).
:rtype: tuple[list[Node], list[Message], list[Node]]
"""
rng = np.random.default_rng(seed)
params = params or {}
# Create a message for each edge
edge_messages: dict[tuple[int, int], Message] = {}
for src, dst in G.edges():
edge_messages[(src, dst)] = Message(name=f"msg_{src}_{dst}")
# Create nodes in topological order
nodes: list[Node] = []
trigger_nodes: list[Node] = []
for node_id in nx.topological_sort(G):
# Inputs: messages from incoming edges
requires = [edge_messages[(src, node_id)] for src in G.predecessors(node_id)]
# Outputs: messages for outgoing edges
provides = [edge_messages[(node_id, dst)] for dst in G.successors(node_id)]
# Skip nodes with no outputs (sink nodes produce no messages)
# But we still need to create them to consume inputs
if node_factory:
node = node_factory(
requires=requires,
provides=provides,
params=params,
name=f"{node_id}",
)
else:
node = MLPNode(
requires=requires,
provides=provides,
params=params,
name=f"{node_id}",
)
nodes.append(node)
# Trigger nodes are sources (no inputs)
if len(requires) == 0:
trigger_nodes.append(node)
all_messages = list(edge_messages.values())
return nodes, all_messages, trigger_nodes
def generate_multi_pipeline_dag(
num_pipelines: int,
nodes_per_pipeline: int,
cross_pipeline_edges: int = 2,
edge_probability: float = 0.3,
seed: int | None = None,
) -> tuple[list[nx.DiGraph], nx.DiGraph]:
"""
Generate multiple DAGs that can be composed into a parent pipeline.
Creates separate DAGs for each pipeline and adds cross-pipeline edges
to create dependencies between them.
:param num_pipelines: Number of child pipelines.
:type num_pipelines: int
:param nodes_per_pipeline: Nodes in each pipeline.
:type nodes_per_pipeline: int
:param cross_pipeline_edges: Number of edges connecting pipelines.
:type cross_pipeline_edges: int
:param edge_probability: Edge probability within each pipeline.
:type edge_probability: float
:param seed: Random seed for reproducibility.
:type seed: int | None
:returns: Tuple of (list of pipeline DAGs, combined DAG).
:rtype: tuple[list[nx.DiGraph], nx.DiGraph]
"""
rng = np.random.default_rng(seed)
pipeline_graphs: list[nx.DiGraph] = []
combined = nx.DiGraph()
# Generate each pipeline's DAG
for p in range(num_pipelines):
offset = p * nodes_per_pipeline
G = generate_dag(
nodes_per_pipeline, edge_probability, seed=rng.integers(0, 10000)
)
# Relabel nodes with offset
mapping = {n: n + offset for n in G.nodes()}
G = nx.relabel_nodes(G, mapping)
# Add pipeline attribute
for n in G.nodes():
G.nodes[n]["pipeline"] = p
pipeline_graphs.append(G)
combined = nx.compose(combined, G)
# Add cross-pipeline edges (from later nodes to earlier nodes of next pipeline)
for _ in range(cross_pipeline_edges):
p1, p2 = rng.choice(num_pipelines, size=2, replace=False)
if p1 > p2:
p1, p2 = p2, p1
# Get sink nodes from p1 (nodes with no outgoing edges within pipeline)
p1_nodes = [n for n in pipeline_graphs[p1].nodes()]
p1_sinks = [n for n in p1_nodes if pipeline_graphs[p1].out_degree(n) == 0]
# Get source-ish nodes from p2 (nodes with few incoming edges)
p2_nodes = [n for n in pipeline_graphs[p2].nodes()]
p2_sources = [n for n in p2_nodes if combined.in_degree(n) <= 1]
if p1_sinks and p2_sources:
src = rng.choice(p1_sinks)
dst = rng.choice(p2_sources)
combined.add_edge(src, dst)
return pipeline_graphs, combined
backend/app/flow/events.py
+64
View File
@@ -0,0 +1,64 @@
"""Event bus bridging the engine's worker threads to async subscribers.
Nodes execute in a thread pool; websocket clients live on the event loop.
Publishers are therefore thread-safe and never block: a subscriber that cannot
keep up loses its oldest queued events rather than stalling the engine.
"""
from __future__ import annotations
import asyncio
import logging
from collections.abc import AsyncIterator
from contextlib import asynccontextmanager
from typing import Any
logger = logging.getLogger(__name__)
QUEUE_SIZE = 256
class EventBus:
"""Fan-out of engine events to any number of async subscribers."""
def __init__(self) -> None:
self._loop: asyncio.AbstractEventLoop | None = None
self._subscribers: set[asyncio.Queue[dict[str, Any]]] = set()
def bind(self, loop: asyncio.AbstractEventLoop) -> None:
"""Attach the bus to the running event loop (called once at startup)."""
self._loop = loop
def publish(self, event: dict[str, Any]) -> None:
"""Publish an event from any thread."""
loop = self._loop
if loop is None or not self._subscribers:
return
try:
loop.call_soon_threadsafe(self._dispatch, event)
except RuntimeError:
# Loop already closed — shutting down.
pass
def _dispatch(self, event: dict[str, Any]) -> None:
for queue in self._subscribers:
if queue.full():
# Drop the oldest so a slow client never blocks the engine.
try:
queue.get_nowait()
except asyncio.QueueEmpty:
pass
queue.put_nowait(event)
@asynccontextmanager
async def subscribe(self) -> AsyncIterator[asyncio.Queue[dict[str, Any]]]:
"""Yield a queue receiving every event published while subscribed."""
queue: asyncio.Queue[dict[str, Any]] = asyncio.Queue(maxsize=QUEUE_SIZE)
self._subscribers.add(queue)
try:
yield queue
finally:
self._subscribers.discard(queue)
event_bus = EventBus()
backend/app/flow/messages.py
+112
View File
@@ -0,0 +1,112 @@
"""Message specifications: the typed contract between nodes.
A node declares ports; each port binds to a message name. The message name is
the wiring: a node consuming ``heating.temperature`` receives whatever any node
provides under that name. Names are namespaced per flow — a bare name is
qualified with the owning flow (``temperature`` in flow ``heating`` becomes
``heating.temperature``), a dotted name is used as written, which is how flows
consume each other's messages.
"""
from __future__ import annotations
import json
from enum import Enum
from typing import Any
from pydantic import BaseModel, ConfigDict, model_validator
class DType(str, Enum):
"""Serializable payload types.
Binary payloads (tensors, images) will arrive later as explicitly declared
codec fields; until then everything on the wire is JSON.
"""
FLOAT = "float"
INT = "int"
STR = "str"
BOOL = "bool"
JSON = "json"
_JSON_TYPES = (dict, list, str, int, float, bool, type(None))
class MessageSpec(BaseModel):
"""A single port of a node, and the message it is bound to.
:param name: The message this port binds to. Bare names are qualified with
the flow name at load time; empty means the port is unbound.
:param port: The identifier the node function sees. Defaults to the last
segment of ``name``, so unqualified flows read naturally.
:param dtype: Payload type, validated on every message that passes through.
"""
model_config = ConfigDict(frozen=True)
name: str = ""
port: str = ""
dtype: DType = DType.FLOAT
@model_validator(mode="after")
def _default_port(self) -> MessageSpec:
if not self.port and self.name:
object.__setattr__(self, "port", self.name.rsplit(".", 1)[-1])
return self
def check(self, value: Any) -> None:
"""Raise if ``value`` does not match this port's declared type."""
if self.dtype is DType.BOOL:
ok = isinstance(value, bool)
elif self.dtype is DType.INT:
# bool is an int subclass; a flag is not a number here.
ok = isinstance(value, int) and not isinstance(value, bool)
elif self.dtype is DType.FLOAT:
ok = isinstance(value, (int, float)) and not isinstance(value, bool)
elif self.dtype is DType.STR:
ok = isinstance(value, str)
else:
ok = isinstance(value, _JSON_TYPES)
if not ok:
raise TypeError(
f"{self.name or self.port}: expected {self.dtype.value}, "
f"got {type(value).__name__}"
)
def coerce(self, value: Any) -> Any:
"""Best-effort conversion of an external value into this port's type.
Used where payloads arrive as text (HTTP query strings, MQTT), never on
the path between nodes — there a wrong type is an error, not a hint.
"""
if self.dtype is DType.FLOAT:
return float(value)
if self.dtype is DType.INT:
return int(value)
if self.dtype is DType.BOOL:
if isinstance(value, bool):
return value
return str(value).lower() in ("true", "1", "yes", "on")
if self.dtype is DType.STR:
return value if isinstance(value, str) else json.dumps(value)
return value
def __repr__(self) -> str:
return f"MessageSpec({self.name or self.port})"
def __hash__(self) -> int:
return hash((self.name, self.port))
def qualify(flow: str, name: str) -> str:
"""Resolve a message name against its flow namespace."""
if not name:
return ""
return name if "." in name else f"{flow}.{name}"
def flow_of(qualified: str) -> str:
"""The flow a qualified message name belongs to."""
return qualified.split(".", 1)[0]
backend/app/flow/nodes.py
+365 -406
View File
File diff suppressed because it is too large Load Diff
backend/app/flow/nodes/alarm_test.py
-20
View File
@@ -1,20 +0,0 @@
from nodes import DelayNode
from util import Message
import numpy as np
import logging
logging.basicConfig(
level=logging.INFO,
format="%(asctime)s - %(name)s - %(levelname)s - %(message)s",
)
logger = logging.getLogger(__name__)
def create_node(params: dict) -> DelayNode:
return DelayNode(
requires=[],
provides=[Message(name="timestamp", dtype=float)],
params={"cron": "* * * * * *"},
name="alarm",
)
backend/app/flow/nodes/delay_test.py
-25
View File
@@ -1,25 +0,0 @@
from nodes import DelayNode
from util import Message
import logging
logging.basicConfig(
level=logging.INFO,
format="%(asctime)s - %(name)s - %(levelname)s - %(message)s",
)
logger = logging.getLogger(__name__)
def create_node(params: dict) -> DelayNode:
return DelayNode(
params={
"delay": 1,
},
requires=[
Message(name="random_value_rec_rate", dtype=int),
],
provides=[
Message(name="random_value_rec_rate_delayed", dtype=int),
],
name="delay",
)
backend/app/flow/nodes/influx_test.py
-35
View File
@@ -1,35 +0,0 @@
from nodes import InfluxDbNode
from util import Message
import logging
logging.basicConfig(
level=logging.INFO,
format="%(asctime)s - %(name)s - %(levelname)s - %(message)s",
)
logger = logging.getLogger(__name__)
def create_node(params: dict) -> InfluxDbNode:
return InfluxDbNode(
requires=[
Message(name="random_value_rec_rate", dtype=int),
],
params={
"url": params.get("influxdb_url", "http://10.200.200.115:8086"),
"token": params.get(
"influxdb_token",
"***REMOVED-INFLUXDB-TOKEN***==",
),
"org": params.get("influxdb_org", "strobl"),
"bucket": params.get("influxdb_bucket", "test"),
"writes": {
"random_value_rec_rate": {
"measurement": "test",
"field": "random",
},
},
"synchronous": True,
},
name="influxdb",
)
backend/app/flow/nodes/mqtt_test_a.py
-25
View File
@@ -1,25 +0,0 @@
from nodes import MqttNode
from util import Message
import logging
logging.basicConfig(
level=logging.INFO,
format="%(asctime)s - %(name)s - %(levelname)s - %(message)s",
)
logger = logging.getLogger(__name__)
def create_node(params: dict) -> MqttNode:
return MqttNode(
requires=[
Message(name="random_value_send", dtype=int),
],
params={
"broker_host": "127.0.0.1",
"topic": {
"random_value_send": "random_value",
},
},
name="mqtt_a",
)
backend/app/flow/nodes/mqtt_test_b.py
-25
View File
@@ -1,25 +0,0 @@
from nodes import MqttNode
from util import Message
import logging
logging.basicConfig(
level=logging.INFO,
format="%(asctime)s - %(name)s - %(levelname)s - %(message)s",
)
logger = logging.getLogger(__name__)
def create_node(params: dict) -> MqttNode:
return MqttNode(
provides=[
Message(name="random_value_rec", dtype=int),
],
params={
"broker_host": "127.0.0.1",
"topic": {
"random_value_rec": "random_value",
},
},
name="mqtt_b",
)
backend/app/flow/nodes/notify_test.py
-30
View File
@@ -1,30 +0,0 @@
from nodes import Node
from util import Message
import logging
logging.basicConfig(
level=logging.INFO,
format="%(asctime)s - %(name)s - %(levelname)s - %(message)s",
)
logger = logging.getLogger(__name__)
def create_node(params: dict) -> Node:
def check_power(params, random_value_rec_rate_delayed: int, **kwargs):
threshold = params.get("alert_threshold", 1000.0)
if random_value_rec_rate_delayed > 50:
logger.info(f"[alert_node] ⚠️ Input >50")
else:
logger.info(f"[alert_node] ✓ Input < 50")
return Node(
f=check_power,
requires=[
Message(name="random_value_rec_rate_delayed", dtype=int),
],
provides=[], # Using object for optional string
params=params,
name="notify",
)
backend/app/flow/nodes/random_test.py
-32
View File
@@ -1,32 +0,0 @@
from nodes import Node
from util import Message
import numpy as np
import logging
logging.basicConfig(
level=logging.INFO,
format="%(asctime)s - %(name)s - %(levelname)s - %(message)s",
)
logger = logging.getLogger(__name__)
def create_node(params: dict) -> Node:
def f(params, timestamp):
seed = int(params.get("seed", 1000))
rng = np.random.default_rng(seed=seed)
rand_value = int(100 * rng.uniform())
return {"random_value_send": rand_value}
return Node(
provides=[
Message(name="random_value_send", dtype=int),
],
requires=[Message(name="timestamp", dtype=float)],
f=f,
params={
"seed": "1000",
},
name="random",
)
backend/app/flow/nodes/rate_test.py
-26
View File
@@ -1,26 +0,0 @@
from nodes import DelayNode
from util import Message
import logging
logging.basicConfig(
level=logging.INFO,
format="%(asctime)s - %(name)s - %(levelname)s - %(message)s",
)
logger = logging.getLogger(__name__)
def create_node(params: dict) -> DelayNode:
return DelayNode(
params={
"delay": 0,
"interval": 3,
},
requires=[
Message(name="random_value_rec", dtype=int),
],
provides=[
Message(name="random_value_rec_rate", dtype=int),
],
name="rate",
)
backend/app/flow/nodes_example/alert_node.py
-44
View File
@@ -1,44 +0,0 @@
"""
Example sink node: Alert Generator
This node consumes the comfort index and generates alerts
when comfort drops below a threshold. It demonstrates a
sink node (produces no outputs for other nodes).
"""
from nodes import Node
from util import Message
def create_node(params: dict) -> Node:
"""Create the alert generator node."""
def check_comfort_alert(params, comfort_index: float = 50.0, **kwargs):
"""
Check if comfort level requires an alert.
Generates an alert message if comfort is too low.
"""
threshold = params.get("alert_threshold", 60.0)
if comfort_index < threshold:
alert_msg = f"LOW COMFORT ALERT: Index {comfort_index:.1f} below threshold {threshold}"
print(f"[alert_node] ⚠️ {alert_msg}")
# In a real implementation, this could:
# - Send email/SMS
# - Push to a message queue
# - Trigger home automation
return {"alert": alert_msg}
else:
print(f"[alert_node] ✓ Comfort level OK ({comfort_index:.1f})")
return {"alert": None}
return Node(
f=check_comfort_alert,
requires=[Message(name="comfort_index", dtype=float)],
provides=[
Message(name="alert", dtype=object)
], # Using object for optional string
params=params,
name="alert_node",
)
backend/app/flow/nodes_example/comfort_calculator.py
-46
View File
@@ -1,46 +0,0 @@
"""
Example processing node: Comfort Calculator
This node takes temperature and humidity as inputs and calculates
a comfort index. It demonstrates how nodes can have dependencies
that are automatically resolved by the pipeline.
"""
from nodes import Node
from util import Message
def create_node(params: dict) -> Node:
"""Create the comfort calculator node."""
def calculate_comfort(
params, temperature: float = 20.0, humidity: float = 50.0, **kwargs
):
"""
Calculate comfort index based on temperature and humidity.
Uses a simplified heat index formula.
"""
# Simplified comfort calculation
# Ideal: 22°C, 45% humidity
temp_diff = abs(temperature - 22.0)
humidity_diff = abs(humidity - 45.0)
comfort = 100.0 - (temp_diff * 3) - (humidity_diff * 0.5)
comfort = max(0, min(100, comfort))
print(
f"[comfort_calculator] T={temperature:.1f}°C, H={humidity:.1f}% -> Comfort={comfort:.1f}"
)
return {"comfort_index": comfort}
return Node(
f=calculate_comfort,
requires=[
Message(name="temperature", dtype=float),
Message(name="humidity", dtype=float),
],
provides=[Message(name="comfort_index", dtype=float)],
params=params,
name="comfort_calculator",
)
backend/app/flow/nodes_example/http_sender.py
-35
View File
@@ -1,35 +0,0 @@
"""
Example HTTP sender node: Data Publisher
This node sends processed data to an external API endpoint
via HTTP POST requests.
"""
from nodes import HttpNode
from util import Message
def create_node(params: dict) -> HttpNode:
"""
Create an HTTP sender node that publishes comfort data.
:param params: Parameters passed from the controller.
:type params: dict
:returns: Configured HttpNode instance in sender mode.
:rtype: HttpNode
"""
# Get the target URL from params or use a default
target_url = params.get("webhook_url", "https://httpbin.org/post")
return HttpNode(
url=target_url,
method="POST",
requires=[
Message(name="comfort_index", dtype=float),
Message(name="alert", dtype=object),
],
params=params,
name="data_publisher",
timeout=10.0,
headers={"X-Source": "fluksio-pipeline"},
)
backend/app/flow/nodes_example/http_trigger.py
-30
View File
@@ -1,30 +0,0 @@
"""
Example HTTP trigger node: Webhook Receiver
This node acts as a webhook endpoint that receives temperature data
via HTTP POST requests and injects it into the pipeline.
"""
from nodes import HttpNode
from util import Message
def create_node(params: dict) -> HttpNode:
"""
Create an HTTP trigger node that receives temperature data.
:param params: Parameters passed from the controller.
:type params: dict
:returns: Configured HttpNode instance in trigger mode.
:rtype: HttpNode
"""
return HttpNode(
url="/api/sensors/temperature",
method="POST",
provides=[
Message(name="temperature", dtype=float),
Message(name="sensor_id", dtype=str),
],
params=params,
name="temperature_webhook",
)
backend/app/flow/nodes_example/humidity_sensor.py
-29
View File
@@ -1,29 +0,0 @@
"""
Example trigger node: Humidity Sensor
Another trigger node that simulates a humidity sensor.
"""
from nodes import Node
from util import Message
def create_node(params: dict) -> Node:
"""Create the humidity sensor node."""
def read_humidity(params, **kwargs):
"""Simulate reading humidity from a sensor."""
import random
humidity = 50.0 + random.gauss(0, 10) # 50% ± 10%
humidity = max(0, min(100, humidity)) # Clamp to [0, 100]
print(f"[humidity_sensor] Read humidity: {humidity:.1f}%")
return {"humidity": humidity}
return Node(
f=read_humidity,
requires=[],
provides=[Message(name="humidity", dtype=float)],
params=params,
name="humidity_sensor",
)
backend/app/flow/nodes_example/influxdb_reader.py
-52
View File
@@ -1,52 +0,0 @@
"""
Example InfluxDB reader node: Average Temperature Provider
This node queries InfluxDB for average temperature and provides it
to downstream nodes.
"""
from nodes import InfluxDbNode
from util import Message
def create_node(params: dict) -> InfluxDbNode:
"""
Create an InfluxDB reader node that queries average temperature.
:param params: Parameters passed from the controller, should include:
- ``influxdb_url``: InfluxDB server URL
- ``influxdb_token``: Authentication token
- ``influxdb_org``: Organization name
- ``influxdb_bucket``: Bucket name
:type params: dict
:returns: Configured InfluxDbNode instance.
:rtype: InfluxDbNode
"""
return InfluxDbNode(
provides=[
Message(name="avg_temperature", dtype=float),
Message(name="max_temperature", dtype=float),
],
params={
"url": params.get("influxdb_url", "http://localhost:8086"),
"token": params.get("influxdb_token", "my-token"),
"org": params.get("influxdb_org", "my-org"),
"bucket": params.get("influxdb_bucket", "sensors"),
"query_range": "-1h",
"queries": {
"avg_temperature": {
"measurement": "temperature",
"field": "value",
"tags": {"location": "room1"},
"aggregation": "mean",
},
"max_temperature": {
"measurement": "temperature",
"field": "value",
"tags": {"location": "room1"},
"aggregation": "max",
},
},
},
name="influxdb_temperature_reader",
)
backend/app/flow/nodes_example/influxdb_writer.py
-54
View File
@@ -1,54 +0,0 @@
"""
Example InfluxDB writer node: Temperature Logger
This node writes temperature readings to InfluxDB.
The input is just a float value - the measurement, field, and tags
are configured via params.
"""
from nodes import InfluxDbNode
from util import Message
def create_node(params: dict) -> InfluxDbNode:
"""
Create an InfluxDB writer node that logs temperature data.
The node receives a simple float value and writes it to InfluxDB
with the configured measurement, field, and tags.
:param params: Parameters passed from the controller, should include:
- ``influxdb_url``: InfluxDB server URL
- ``influxdb_token``: Authentication token
- ``influxdb_org``: Organization name
- ``influxdb_bucket``: Bucket name
:type params: dict
:returns: Configured InfluxDbNode instance.
:rtype: InfluxDbNode
"""
return InfluxDbNode(
requires=[
Message(name="temperature", dtype=float),
Message(name="humidity", dtype=float),
],
params={
"url": params.get("influxdb_url", "http://localhost:8086"),
"token": params.get("influxdb_token", "my-token"),
"org": params.get("influxdb_org", "my-org"),
"bucket": params.get("influxdb_bucket", "sensors"),
"write_precision": "ms",
"writes": {
"temperature": {
"measurement": "environment",
"field": "temp_celsius",
"tags": {"location": "room1", "sensor": "dht22"},
},
"humidity": {
"measurement": "environment",
"field": "humidity_percent",
"tags": {"location": "room1", "sensor": "dht22"},
},
},
},
name="influxdb_temperature_writer",
)
backend/app/flow/nodes_example/mqtt_publisher.py
-36
View File
@@ -1,36 +0,0 @@
"""
Example MQTT publisher node: Alert Publisher
This node publishes alerts to an MQTT topic when comfort levels
are outside acceptable ranges.
"""
from nodes import MqttNode
from util import Message
def create_node(params: dict) -> MqttNode:
"""
Create an MQTT publisher node that sends alert messages.
:param params: Parameters passed from the controller, should include:
- ``broker_host``: MQTT broker hostname
- ``broker_port``: MQTT broker port (optional, default 1883)
- ``qos``: Quality of Service level (optional, default 0)
:type params: dict
:returns: Configured MqttNode instance in publisher mode.
:rtype: MqttNode
"""
return MqttNode(
topic="alerts/comfort",
requires=[
Message(name="comfort_index", dtype=float),
Message(name="alert", dtype=object),
],
params={
**params,
"qos": params.get("qos", 1), # Use QoS 1 for alerts
"retain": True, # Retain last alert
},
name="mqtt_alert_publisher",
)
backend/app/flow/nodes_example/mqtt_subscriber.py
-31
View File
@@ -1,31 +0,0 @@
"""
Example MQTT subscriber node: Temperature Sensor Listener
This node subscribes to an MQTT topic and triggers the pipeline
when temperature readings are received.
"""
from nodes import MqttNode
from util import Message
def create_node(params: dict) -> MqttNode:
"""
Create an MQTT subscriber node that listens for temperature data.
:param params: Parameters passed from the controller, should include:
- ``broker_host``: MQTT broker hostname
- ``broker_port``: MQTT broker port (optional, default 1883)
:type params: dict
:returns: Configured MqttNode instance in subscriber mode.
:rtype: MqttNode
"""
return MqttNode(
topic="sensors/temperature",
provides=[
Message(name="temperature", dtype=float),
Message(name="sensor_id", dtype=str),
],
params=params,
name="mqtt_temperature_listener",
)
backend/app/flow/nodes_example/temperature_sensor.py
-37
View File
@@ -1,37 +0,0 @@
"""
Example trigger node: Temperature Sensor
This is a trigger node (no inputs) that simulates a temperature sensor.
Trigger nodes act as entry points to the pipeline - they can be triggered
externally via HTTP, MQTT, or other mechanisms.
"""
from nodes import Node
from util import Message
def create_node(params: dict) -> Node:
"""
Create the temperature sensor node.
:param params: Parameters passed from the controller.
:type params: dict
:returns: Configured Node instance.
:rtype: Node
"""
def read_temperature(params, **kwargs):
"""Simulate reading temperature from a sensor."""
import random
temperature = 20.0 + random.gauss(0, 2) # 20°C ± 2°C
print(f"[temperature_sensor] Read temperature: {temperature:.2f}°C")
return {"temperature": temperature}
return Node(
f=read_temperature,
requires=[], # No inputs - this is a trigger node
provides=[Message(name="temperature", dtype=float)],
params=params,
name="temperature_sensor",
)
backend/app/flow/pipeline.py
+232 -410
View File
@@ -1,210 +1,134 @@
"""
Pipeline module for directed acyclic graph execution.
"""Pipeline: the executable graph.
This module provides a Pipeline class that manages nodes with automatic
dependency resolution and supports both sequential and parallel execution.
One pipeline holds the nodes of every loaded flow. Edges are not declared —
they follow from message names, so a node consuming ``heating.setpoint`` is
downstream of every node providing it. Several producers of one message are
allowed: each publication triggers the consumers, and the latest value wins.
"""
from __future__ import annotations
import logging
import time
from collections import deque
from concurrent.futures import ThreadPoolExecutor, Future, wait
from concurrent.futures import Future, ThreadPoolExecutor, wait
from typing import Any, Literal
import matplotlib.pyplot as plt
import networkx as nx
from pydantic import BaseModel
from nodes import Node
from state import StateBackend, MemoryState
from util import Message
from app.flow.events import EventBus
from app.flow.messages import flow_of
from app.flow.nodes import Node
from app.flow.state import MemoryState, StateBackend
logger = logging.getLogger(__name__)
class ValidationIssue(BaseModel):
"""A problem that keeps a flow from running correctly."""
code: Literal[
"cycle",
"unconnected_input",
"missing_initial_value",
"node_error",
]
message: str
flow: str = ""
nodes: list[str] = []
node: str | None = None
port: str | None = None
message_name: str | None = None
class Pipeline:
"""
Directed acyclic graph of nodes with automatic dependency resolution.
A Pipeline manages a collection of nodes, automatically resolving their
dependencies and executing them in the correct order. Supports parallel
execution, external triggering, and composition of multiple pipelines.
:param nodes: List of nodes owned by this pipeline.
:type nodes: list[Node] | None
:param pipelines: Child pipelines to compose into this pipeline.
:type pipelines: list[Pipeline] | None
:param inputs: External input messages (not produced by any node).
:type inputs: list[Message] | None
:param outputs: Output messages (for documentation purposes).
:type outputs: list[Message] | None
:param max_workers: Maximum thread pool workers for parallel execution.
:type max_workers: int | None
:param state: State backend for storing pipeline values.
:type state: StateBackend | None
:example:
>>> pipeline = Pipeline(
... nodes=[node_a, node_b],
... pipelines=[child_pipeline],
... max_workers=4
... )
>>> result = pipeline.run()
"""
"""Directed graph of nodes with automatic dependency resolution."""
__slots__ = (
"_own_nodes",
"_child_pipelines",
"_parent",
"_nodes",
"_state",
"_events",
"_max_workers",
"produces",
"dependencies",
"_edges",
"_execution_order",
"_downstream_cache",
"_max_workers",
)
def __init__(
self,
nodes: list[Node] | None = None,
pipelines: list[Pipeline] | None = None,
inputs: list[Message] | None = None,
outputs: list[Message] | None = None,
max_workers: int | None = None,
state: StateBackend | None = None,
events: EventBus | None = None,
max_workers: int | None = None,
initial_values: dict[str, Any] | None = None,
) -> None:
self._own_nodes = nodes or []
self._child_pipelines = pipelines or []
self._parent: Pipeline | None = None
self._nodes = nodes or []
self._state: StateBackend = state or MemoryState()
self._events = events
self._max_workers = max_workers
# Set parent reference for child pipelines
for child in self._child_pipelines:
child._parent = self
# A message may have several producers; every one of them is upstream
# of the nodes consuming it.
self.produces: dict[str, list[Node]] = {}
for node in self._nodes:
for msg in node.provides:
self.produces.setdefault(msg, []).append(node)
# Collect all nodes and build graph
all_nodes = self._collect_all_nodes()
# Use provided state backend or create default MemoryState
self._state: StateBackend = state or MemoryState()
# Build producer map and dependency graph
self.produces: dict[str, Node | None] = {
**({m.name: None for m in inputs} if inputs else {}),
**{msg: node for node in all_nodes for msg in node.provides},
}
self.dependencies: dict[Node, frozenset[Node]] = {
node: frozenset(
self.produces[msg] for msg in node.requires if self.produces.get(msg)
producer
for msg in node.requires
for producer in self.produces.get(msg, ())
if producer is not node
)
for node in all_nodes
for node in self._nodes
}
# Lazy-initialized caches
self._edges: dict[Node, set[Node]] | None = None
self._execution_order: list[Node] | None = None
self._downstream_cache: dict[Node, list[Node]] = {}
# Bind all nodes to root pipeline
for node in all_nodes:
node.bind(self._root)
if initial_values:
with self._state.lock():
for name, value in initial_values.items():
if name not in self._state:
self._state[name] = value
for node in self._nodes:
node.bind(self)
# -------------------------------------------------------------------------
# Properties
# Graph
# -------------------------------------------------------------------------
@property
def _root(self) -> Pipeline:
"""
Get the root pipeline in the hierarchy.
:returns: The topmost parent pipeline, or self if no parent.
:rtype: Pipeline
"""
return self._parent._root if self._parent else self
@property
def nodes(self) -> list[Node]:
"""
All nodes in this pipeline and child pipelines.
return self._nodes
:returns: Flattened list of all nodes.
:rtype: list[Node]
"""
return self._collect_all_nodes()
@property
def state(self) -> StateBackend:
return self._state
@property
def edges(self) -> dict[Node, set[Node]]:
"""
Reverse dependency graph mapping producers to consumers.
Lazily built on first access.
:returns: Dict mapping each node to its downstream consumers.
:rtype: dict[Node, set[Node]]
"""
"""Producers mapped to their consumers, built on first access."""
if self._edges is None:
self._edges = {node: set() for node in self.nodes}
self._edges = {node: set() for node in self._nodes}
for consumer, producers in self.dependencies.items():
for producer in producers:
self._edges[producer].add(consumer)
return self._edges
@property
def state(self) -> StateBackend:
"""
State backend for storing pipeline values.
def get_node_by_id(self, nid: str) -> Node | None:
return next((n for n in self._nodes if n.id == nid), None)
:returns: State backend.
:rtype: StateBackend
"""
return self._state._data
# -------------------------------------------------------------------------
# Node Access
# -------------------------------------------------------------------------
def _collect_all_nodes(self) -> list[Node]:
"""
Recursively collect nodes from this pipeline and all children.
:returns: List of all nodes in the hierarchy.
:rtype: list[Node]
"""
nodes = list(self._own_nodes)
for child in self._child_pipelines:
nodes.extend(child._collect_all_nodes())
return nodes
def get_node_by_id(self, nid) -> Node | None:
"""
Find a node by its ID.
Searches this pipeline's own nodes first, then child pipelines.
:param nid: The ID of the node to find.
:type nid: Any
:returns: The node with the given ID, or None if not found.
:rtype: Node | None
"""
if node := next((n for n in self._own_nodes if n.id is nid), None):
return node
for child in self._child_pipelines:
if node := child.get_node_by_id(nid):
return node
return None
# -------------------------------------------------------------------------
# Graph Algorithms
# -------------------------------------------------------------------------
def flow_nodes(self, flow: str) -> set[Node]:
return {n for n in self._nodes if n.flow == flow}
def _topological_sort(self) -> list[Node]:
"""
Compute topological ordering of nodes using Kahn's algorithm.
Results are cached for subsequent calls.
:returns: Nodes in topologically sorted order.
:rtype: list[Node]
:raises ValueError: If a cycle is detected in the graph.
"""
"""Kahn's algorithm; nodes left over are part of a cycle."""
if self._execution_order is not None:
return self._execution_order
@@ -220,23 +144,10 @@ class Pipeline:
if in_degree[consumer] == 0:
queue.append(consumer)
if len(result) != len(self.nodes):
raise ValueError("Cycle detected in pipeline graph")
self._execution_order = result
return result
def _get_downstream(self, start: Node) -> list[Node]:
"""
Get all downstream nodes from a starting node in topological order.
Results are cached per start node.
:param start: The node to find downstream nodes from.
:type start: Node
:returns: Topologically sorted downstream nodes.
:rtype: list[Node]
"""
if start not in self._downstream_cache:
reachable: set[Node] = set()
queue = deque([start])
@@ -245,186 +156,186 @@ class Pipeline:
if consumer not in reachable:
reachable.add(consumer)
queue.append(consumer)
self._downstream_cache[start] = [
n for n in self._topological_sort() if n in reachable
]
order = self._topological_sort()
ordered = [n for n in order if n in reachable]
# Nodes inside a cycle never make it into the topological order.
ordered += [n for n in reachable if n not in order]
self._downstream_cache[start] = ordered
return self._downstream_cache[start]
# -------------------------------------------------------------------------
# Synchronous Node Support
# Validation
# -------------------------------------------------------------------------
def validate(self, known_inputs: set[str] | None = None) -> list[ValidationIssue]:
"""Report everything that would keep this graph from running.
:param known_inputs: Message names supplied from outside the graph
(flow inputs with an initial value).
"""
known = known_inputs or set()
issues: list[ValidationIssue] = []
ordered = set(self._topological_sort())
if len(ordered) != len(self._nodes):
cyclic = sorted(n.id for n in self._nodes if n not in ordered)
issues.append(
ValidationIssue(
code="cycle",
message=(
"These nodes depend on each other in a loop, so none of "
"them can run: " + ", ".join(cyclic)
),
flow=flow_of(cyclic[0]) if cyclic else "",
nodes=cyclic,
)
)
for node in self._nodes:
for msg_name, spec in node.requires.items():
if msg_name in self.produces or msg_name in known:
continue
issues.append(
ValidationIssue(
code="unconnected_input",
message=(
f"'{node.local_id}' waits for '{msg_name}', "
"which nothing provides."
),
flow=node.flow,
node=node.id,
port=spec.port,
message_name=msg_name,
)
)
return issues
# -------------------------------------------------------------------------
# Readiness (synchronous nodes)
# -------------------------------------------------------------------------
def _version_key(self, msg_name: str) -> str:
"""Get the state key for a message's version number."""
return f"__version__:{msg_name}"
def _last_seen_key(self, node_name: str, msg_name: str) -> str:
"""Get the state key for the version a node last processed."""
return f"__last_seen__:{node_name}:{msg_name}"
def _increment_message_versions(self, outputs: dict) -> None:
"""
Increment version numbers for all output messages.
def _timestamp_key(self, msg_name: str) -> str:
return f"__ts__:{msg_name}"
:param outputs: Dict of message names to values.
:type outputs: dict
"""
def _increment_message_versions(self, outputs: dict) -> None:
for msg_name in outputs:
self._state.increment(self._version_key(msg_name))
def _check_synchronous_ready(self, node: Node) -> tuple[bool, dict[str, int]]:
"""
Check if a synchronous node is ready to execute.
A synchronous node is ready when ALL its required inputs have a newer
version than what the node last processed.
:param node: The synchronous node to check.
:type node: Node
:returns: Tuple of (is_ready, current_versions dict).
:rtype: tuple[bool, dict[str, int]]
"""
"""A synchronous node runs once every input is newer than last time."""
if not node.requires:
return True, {}
# Build all keys we need to fetch
version_keys = [self._version_key(msg) for msg in node.requires]
last_seen_keys = [self._last_seen_key(node.name, msg) for msg in node.requires]
last_seen_keys = [self._last_seen_key(node.id, msg) for msg in node.requires]
values = self._state.get_multi(version_keys + last_seen_keys)
# Fetch all values atomically
all_keys = version_keys + last_seen_keys
values = self._state.get_multi(all_keys)
# Check each input
current_versions = {}
all_newer = True
for msg_name in node.requires:
version_key = self._version_key(msg_name)
last_seen_key = self._last_seen_key(node.name, msg_name)
current_version = values.get(version_key) or 0
last_seen_version = values.get(last_seen_key) or 0
current_versions[msg_name] = current_version
# For synchronous nodes, version must be:
# 1. Greater than 0 (message has been received at least once)
# 2. Strictly greater than last seen (message has been updated since last execution)
if current_version == 0 or current_version <= last_seen_version:
current = values.get(self._version_key(msg_name)) or 0
last_seen = values.get(self._last_seen_key(node.id, msg_name)) or 0
current_versions[msg_name] = current
if current == 0 or current <= last_seen:
all_newer = False
return all_newer, current_versions
def _try_acquire_synchronous_execution(
self,
node: Node,
current_versions: dict[str, int],
self, node: Node, current_versions: dict[str, int]
) -> bool:
"""
Attempt to acquire exclusive execution rights for a synchronous node.
Uses compare-and-swap to atomically verify versions haven't changed
and update last_seen versions. This prevents race conditions when
multiple threads try to execute the same synchronous node.
:param node: The node attempting to execute.
:type node: Node
:param current_versions: The versions that were checked.
:type current_versions: dict[str, int]
:returns: True if execution rights acquired, False otherwise.
:rtype: bool
"""
"""Claim the right to execute, so concurrent triggers run a node once."""
if not current_versions:
return True
# Build expected values and updates
expected = {}
updates = {}
for msg_name, version in current_versions.items():
version_key = self._version_key(msg_name)
last_seen_key = self._last_seen_key(node.name, msg_name)
# Expect the version hasn't changed since we checked
expected[version_key] = version
# Update last_seen to this version
updates[last_seen_key] = version
expected[self._version_key(msg_name)] = version
updates[self._last_seen_key(node.id, msg_name)] = version
return self._state.compare_and_swap_multi(expected, updates)
def _is_node_ready(self, node: Node, state: StateBackend) -> bool:
"""
Check if a node is ready to execute.
For non-synchronous nodes: ready when all required inputs exist in state.
For synchronous nodes: ready when all inputs are newer than last processed
AND we can acquire exclusive execution rights.
:param node: The node to check.
:type node: Node
:param state: State backend containing input values.
:type state: StateBackend
:returns: True if the node should execute.
:rtype: bool
"""
# First check: all required inputs must exist in state
with state.lock():
for msg_name in node.requires:
if msg_name not in state:
return False
# For non-synchronous nodes, that's all we need
if not node.synchronous:
return True
# For synchronous nodes, check if all inputs have been updated
is_ready, current_versions = self._check_synchronous_ready(node)
if not is_ready:
return False
# Try to acquire execution rights atomically
# This prevents race conditions when multiple triggers happen concurrently
return self._try_acquire_synchronous_execution(node, current_versions)
# -------------------------------------------------------------------------
# Execution Core
# Execution
# -------------------------------------------------------------------------
def _execute_node(
self,
node: Node,
state: StateBackend,
) -> dict | None:
"""
Execute a single node without triggering downstream propagation.
def _publish(self, event: dict[str, Any]) -> None:
if self._events is not None:
self._events.publish(event)
:param node: The node to execute.
:type node: Node
:param state: State backend containing input values.
:type state: StateBackend
:returns: Node outputs, or None if no outputs.
:rtype: dict | None
"""
node._pipeline = None
def _execute_node(self, node: Node, state: StateBackend) -> dict | None:
"""Run one node and record its outputs. Never raises."""
started = time.perf_counter()
try:
# Gather inputs with locking
with state.lock():
inputs = {k: state[k] for k in node.requires if k in state}
result = node(inputs)
result = node.execute(inputs)
# Update state with outputs and increment versions
if result:
ts = time.time()
with state.lock():
state.update(result)
# Increment version numbers for synchronous node tracking
state.update(
{self._timestamp_key(name): ts for name in result},
)
self._increment_message_versions(result)
for name, value in result.items():
self._publish(
{
"type": "message_value",
"flow": flow_of(name),
"name": name,
"value": value,
"ts": ts,
}
)
self._publish(
{
"type": "node_executed",
"flow": node.flow,
"node": node.id,
"duration_ms": round((time.perf_counter() - started) * 1000, 2),
"ts": time.time(),
}
)
return result
finally:
node._pipeline = self._root
except Exception as exc:
# One failing node must not take the rest of the graph down.
logger.exception("Node '%s' failed", node.id)
self._publish(
{
"type": "node_error",
"flow": node.flow,
"node": node.id,
"error": f"{type(exc).__name__}: {exc}",
"ts": time.time(),
}
)
return None
def _execute_parallel(
self,
@@ -432,51 +343,34 @@ class Pipeline:
state: StateBackend,
check_ready: bool = False,
) -> StateBackend:
"""
Execute nodes in parallel using dynamic scheduling.
:param nodes_subset: Subset of nodes to execute, or None for all nodes.
:type nodes_subset: set[Node] | None
:param state: State backend for inputs/outputs.
:type state: StateBackend
:param check_ready: If True, check state for required inputs before scheduling.
:type check_ready: bool
:returns: State backend after execution.
:rtype: StateBackend
"""
target_nodes = nodes_subset or set(self.nodes)
"""Execute nodes concurrently, scheduling each as its inputs arrive."""
target_nodes = nodes_subset if nodes_subset is not None else set(self._nodes)
if not target_nodes:
return state
# Build in-degree map (only counting deps within target set)
in_degree = {
n: sum(1 for dep in self.dependencies[n] if dep in target_nodes)
for n in target_nodes
}
# Track submitted/completed/skipped nodes
submitted: set[Node] = set()
skipped: set[Node] = set() # Synchronous nodes that weren't ready
skipped: set[Node] = set()
node_futures: dict[Node, Future] = {}
def is_ready(n: Node) -> bool:
"""Check if node can be scheduled."""
if in_degree[n] != 0:
return False
if check_ready:
# Use the enhanced readiness check that handles synchronous nodes
return self._is_node_ready(n, state)
return True
def submit_ready(executor: ThreadPoolExecutor) -> None:
"""Submit all currently ready nodes."""
for n in target_nodes:
if n not in submitted and n not in skipped and is_ready(n):
submitted.add(n)
node_futures[n] = executor.submit(self._execute_node, n, state)
elif n not in submitted and n.synchronous and in_degree[n] == 0:
# Mark synchronous nodes that weren't ready as skipped
# They may become ready on a future trigger
# Not ready now; a later trigger may make it ready.
skipped.add(n)
with ThreadPoolExecutor(max_workers=self._max_workers) as executor:
@@ -487,15 +381,10 @@ class Pipeline:
completed = [n for n, f in node_futures.items() if f in done]
for n in completed:
future = node_futures.pop(n)
if exc := future.exception():
raise exc
# Only propagate to downstream nodes if this node produced output
# If result is None, the node chose not to forward data (e.g., rate limiting)
result = future.result()
result = node_futures.pop(n).result()
# A node returning nothing (rate limiting, an error) stops
# propagation along its branch.
if result is not None:
# Update in-degrees and submit newly ready nodes
for consumer in self.edges[n]:
if consumer in target_nodes:
in_degree[consumer] -= 1
@@ -503,124 +392,57 @@ class Pipeline:
return state
# -------------------------------------------------------------------------
# Public Execution API
# -------------------------------------------------------------------------
def run(self, inputs: dict | None = None) -> StateBackend:
"""
Execute the entire pipeline.
:param inputs: Initial input values to seed the state.
:type inputs: dict | None
:returns: State backend containing all computed values.
:rtype: StateBackend
:example:
>>> result = pipeline.run({"input": 42})
"""
# Create a fresh state for this run
state = MemoryState()
def run(
self, inputs: dict | None = None, nodes: set[Node] | None = None
) -> StateBackend:
"""Execute the graph (or one flow's nodes) against the shared state."""
if inputs:
state.update(inputs)
with self._state.lock():
self._state.update(inputs)
self._increment_message_versions(inputs)
return self._execute_parallel(None, state, check_ready=False)
return self._execute_parallel(nodes, self._state, check_ready=False)
def trigger(self, node: Node, outputs: dict | None) -> StateBackend:
"""
Trigger execution after a node runs externally.
Updates the shared state with node outputs, increments message versions
for synchronous node tracking, and executes all downstream nodes whose
dependencies are satisfied.
:param node: The node that was triggered.
:type node: Node
:param outputs: Outputs produced by the triggered node.
:type outputs: dict | None
:returns: Updated shared state.
:rtype: StateBackend
"""
root = self._root
state = root._state
"""Publish a node's outputs and run everything downstream of it."""
state = self._state
if outputs:
ts = time.time()
with state.lock():
state.update(outputs)
# Increment version numbers for synchronous node tracking
root._increment_message_versions(outputs)
state.update({self._timestamp_key(name): ts for name in outputs})
self._increment_message_versions(outputs)
for name, value in outputs.items():
self._publish(
{
"type": "message_value",
"flow": flow_of(name),
"name": name,
"value": value,
"ts": ts,
}
)
downstream = set(root._get_downstream(node))
downstream = set(self._get_downstream(node))
if not downstream:
return state
return root._execute_parallel(downstream, state, check_ready=True)
return self._execute_parallel(downstream, state, check_ready=True)
def values(self, flow: str | None = None) -> dict[str, dict[str, Any]]:
"""Last value and timestamp of every message, optionally one flow's."""
out: dict[str, dict[str, Any]] = {}
with self._state.lock():
keys = [k for k in self._state.keys() if not k.startswith("__")]
for key in keys:
if flow and flow_of(key) != flow:
continue
out[key] = {
"value": self._state.get(key),
"ts": self._state.get(self._timestamp_key(key)),
}
return out
def reset(self) -> None:
"""
Reset pipeline state for a new execution cycle.
Clears all computed values from the shared state.
"""
self._root._state.clear()
# -------------------------------------------------------------------------
# Visualization
# -------------------------------------------------------------------------
def visualize(self, filename: str = "pipeline.png") -> None:
"""
Save a visualization of the pipeline graph.
:param filename: Output filename for the image.
:type filename: str
"""
G = nx.DiGraph()
G.add_nodes_from(self.nodes)
G.add_edges_from(
(producer, consumer)
for consumer, producers in self.dependencies.items()
for producer in producers
)
plt.figure(figsize=(10, 6))
pos = nx.spring_layout(G, k=0.8, seed=42)
node_size = 1800
node_radius = (node_size**0.5) / 2
nx.draw_networkx_nodes(
G,
pos,
node_size=node_size,
node_color="#59849B",
edgecolors="#59849B",
linewidths=1.2,
alpha=0.95,
)
nx.draw_networkx_edges(
G,
pos,
arrows=True,
arrowstyle="-|>",
arrowsize=18,
width=2.8,
edge_color="#DE8F6E",
connectionstyle="arc3,rad=0.05",
min_source_margin=node_radius,
min_target_margin=node_radius,
)
nx.draw_networkx_labels(
G,
pos,
font_size=15,
font_color="#F5F9E9",
font_weight="bold",
)
plt.axis("off")
plt.gca().set_facecolor("#333232")
plt.gcf().set_facecolor("#333232")
plt.tight_layout()
plt.savefig(filename, dpi=200)
plt.close()
self._state.clear()
backend/app/flow/schemas.py
+118
View File
@@ -0,0 +1,118 @@
"""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
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)
@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
class NodeStatusPublic(BaseModel):
"""Whether a node loaded, and why not."""
id: str
status: str = "active"
error: str | None = None
class MessageValue(BaseModel):
"""The last payload seen on a message."""
value: Any = None
ts: float | None = None
class FlowSummary(BaseModel):
name: str
title: str = ""
node_count: int = 0
error_count: int = 0
class FlowsPublic(BaseModel):
data: list[FlowSummary]
count: int
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
backend/app/flow/secrets.py
+115
View File
@@ -0,0 +1,115 @@
"""Encrypted credential store for node integrations.
Nodes never hold credentials: a parameter written as ``{"$secret": "name"}``
is replaced with the stored value when the node is built. Secrets are kept
encrypted outside the flows repository, so what gets committed and shared
never contains a password.
"""
from __future__ import annotations
import base64
import hashlib
import json
import logging
from pathlib import Path
from typing import Any
from cryptography.fernet import Fernet, InvalidToken
logger = logging.getLogger(__name__)
SECRET_REF = "$secret"
class SecretNotFound(KeyError):
"""A node asked for a secret that is not in the store."""
def __init__(self, name: str) -> None:
super().__init__(name)
self.name = name
def __str__(self) -> str:
return f"No secret named '{self.name}' — add it under Settings › Secrets."
class SecretsStore:
"""Named credentials, encrypted at rest with a key derived from the app key."""
def __init__(self, path: Path, key: str) -> None:
self._path = path
self._fernet = Fernet(
base64.urlsafe_b64encode(hashlib.sha256(key.encode()).digest())
)
def _read(self) -> dict[str, str]:
if not self._path.exists():
return {}
try:
data: dict[str, str] = json.loads(
self._fernet.decrypt(self._path.read_bytes())
)
return data
except (InvalidToken, ValueError):
# A changed app key makes existing secrets unreadable.
logger.error(
"Cannot decrypt %s — it was written with a different SECRET_KEY",
self._path,
)
return {}
def _write(self, data: dict[str, str]) -> None:
self._path.parent.mkdir(parents=True, exist_ok=True)
self._path.write_bytes(self._fernet.encrypt(json.dumps(data).encode()))
def list(self) -> list[str]:
return sorted(self._read())
def get(self, name: str) -> str:
data = self._read()
if name not in data:
raise SecretNotFound(name)
return data[name]
def set(self, name: str, value: str) -> None:
data = self._read()
data[name] = value
self._write(data)
def delete(self, name: str) -> None:
data = self._read()
if data.pop(name, None) is None:
raise SecretNotFound(name)
self._write(data)
_store: SecretsStore | None = None
def init_secrets(path: Path, key: str) -> SecretsStore:
"""Create the process-wide store (called once at startup)."""
global _store
_store = SecretsStore(path, key)
return _store
def get_secrets() -> SecretsStore:
if _store is None:
raise RuntimeError("Secrets store not initialised")
return _store
def get(name: str) -> str:
"""Look up a secret by name — the entry point for custom node code."""
return get_secrets().get(name)
def resolve_params(params: Any) -> Any:
"""Replace every ``{"$secret": "name"}`` reference with its value."""
if isinstance(params, dict):
if set(params) == {SECRET_REF} and isinstance(params[SECRET_REF], str):
return get(params[SECRET_REF])
return {k: resolve_params(v) for k, v in params.items()}
if isinstance(params, list):
return [resolve_params(v) for v in params]
return params
backend/app/flow/state.py
+8 -7
View File
@@ -7,11 +7,12 @@ supporting both in-memory storage and Redis for distributed execution.
from __future__ import annotations
import pickle
import json
from abc import ABC, abstractmethod
from collections.abc import Iterator
from contextlib import contextmanager
from threading import RLock
from typing import Any, Iterator
from typing import Any
import redis
@@ -255,8 +256,8 @@ class RedisState(StateBackend):
"""
Redis-based state backend for distributed execution.
Supports automatic serialization using pickle and distributed locking.
Uses a namespace prefix to isolate different pipeline executions.
Values are stored as JSON — anything a node passes through must be
serializable. Uses a namespace prefix to isolate pipeline executions.
:param host: Redis host address.
:type host: str
@@ -306,11 +307,11 @@ class RedisState(StateBackend):
def _serialize(self, value: Any) -> bytes:
"""Serialize value for storage."""
return pickle.dumps(value)
return json.dumps(value).encode()
def _deserialize(self, data: bytes | None) -> Any:
"""Deserialize value from storage."""
return pickle.loads(data) if data else None
return json.loads(data) if data else None
def get(self, key: str, default: Any = None) -> Any:
data = self._client.get(self._key(key))
@@ -418,7 +419,7 @@ class RedisState(StateBackend):
values = self._client.mget(full_keys)
result = {}
for key, value in zip(keys, values):
for key, value in zip(keys, values, strict=True):
result[key] = self._deserialize(value) if value is not None else None
return result
backend/app/flow/store.py
+154
View File
@@ -0,0 +1,154 @@
"""Flows on disk, versioned with git.
Every flow is a directory: ``flow.json`` for the structure, ``nodes/*.py`` for
node logic. The whole tree is a git repository and each saved change is a
commit, so a flow's history is readable with ordinary git tooling and two
flows never collide in one file.
"""
from __future__ import annotations
import logging
import shutil
import subprocess
from pathlib import Path
from app.flow.schemas import FlowDef
logger = logging.getLogger(__name__)
DEFAULT_SOURCE = '''"""A new node. Return a dict keyed by your output ports."""
def process(params):
return {}
'''
class FlowNotFound(KeyError):
def __init__(self, name: str) -> None:
super().__init__(name)
self.name = name
def __str__(self) -> str:
return f"No flow named '{self.name}'"
class FlowStore:
"""Reads and writes flows, committing every change."""
def __init__(self, root: Path) -> None:
self.root = root
self.root.mkdir(parents=True, exist_ok=True)
if not (self.root / ".git").exists():
self._git("init", "-q")
self._commit("Initialise flow store", allow_empty=True)
# -------------------------------------------------------------------------
# git
# -------------------------------------------------------------------------
def _git(self, *args: str) -> subprocess.CompletedProcess[str]:
return subprocess.run(
["git", "-C", str(self.root), *args],
capture_output=True,
text=True,
check=False,
)
def _commit(self, message: str, allow_empty: bool = False) -> None:
self._git("add", "-A")
result = self._git(
"-c",
"user.name=fluksio",
"-c",
"user.email=fluksio@localhost",
"commit",
"-q",
*(["--allow-empty"] if allow_empty else []),
"-m",
message,
)
if result.returncode != 0 and "nothing to commit" not in result.stdout:
logger.warning("Could not commit flow change: %s", result.stdout.strip())
# -------------------------------------------------------------------------
# Paths
# -------------------------------------------------------------------------
def _flow_dir(self, name: str) -> Path:
return self.root / name
def _flow_file(self, name: str) -> Path:
return self._flow_dir(name) / "flow.json"
def _node_file(self, flow: str, node_id: str) -> Path:
return self._flow_dir(flow) / "nodes" / f"{node_id}.py"
# -------------------------------------------------------------------------
# Flows
# -------------------------------------------------------------------------
def list_flows(self) -> list[str]:
return sorted(path.parent.name for path in self.root.glob("*/flow.json"))
def exists(self, name: str) -> bool:
return self._flow_file(name).exists()
def read_flow(self, name: str) -> FlowDef:
path = self._flow_file(name)
if not path.exists():
raise FlowNotFound(name)
return FlowDef.model_validate_json(path.read_text())
def read_all(self) -> list[FlowDef]:
flows = []
for name in self.list_flows():
try:
flows.append(self.read_flow(name))
except Exception:
logger.exception("Skipping unreadable flow '%s'", name)
return flows
def write_flow(self, flow: FlowDef) -> bool:
"""Save a flow. Returns False when nothing actually changed."""
path = self._flow_file(flow.name)
content = flow.model_dump_json(indent=2) + "\n"
if path.exists() and path.read_text() == content:
return False
path.parent.mkdir(parents=True, exist_ok=True)
path.write_text(content)
self._commit(f"Update flow '{flow.name}'")
return True
def delete_flow(self, name: str) -> None:
directory = self._flow_dir(name)
if not directory.exists():
raise FlowNotFound(name)
shutil.rmtree(directory)
self._commit(f"Delete flow '{name}'")
# -------------------------------------------------------------------------
# Node source
# -------------------------------------------------------------------------
def read_node_source(self, flow: str, node_id: str) -> str:
path = self._node_file(flow, node_id)
return path.read_text() if path.exists() else DEFAULT_SOURCE
def write_node_source(self, flow: str, node_id: str, code: str) -> bool:
path = self._node_file(flow, node_id)
if path.exists() and path.read_text() == code:
return False
path.parent.mkdir(parents=True, exist_ok=True)
path.write_text(code)
self._commit(f"Update node '{flow}.{node_id}'")
return True
def delete_node_source(self, flow: str, node_id: str) -> None:
path = self._node_file(flow, node_id)
if path.exists():
path.unlink()
self._commit(f"Delete node '{flow}.{node_id}'")
backend/app/flow/test.py
-73
View File
@@ -1,73 +0,0 @@
from pipeline import Pipeline
from state import MemoryState, RedisState
from dag_generator import (
generate_multi_pipeline_dag,
dag_to_pipeline_nodes,
)
import time
import numpy as np
import threading
# Thread-safe RNG
rng_lock = threading.Lock()
rng = np.random.default_rng(1000)
# Parameters for nodes
params = {"rng": rng, "rng_lock": rng_lock}
state = RedisState(
host="localhost",
port=6379,
namespace="fluksio_pipeline",
ttl=3600, # Keys expire after 1 hour
)
print("Using Redis state backend")
assert state.ping(), "Can't connect to redis"
n_pipelines = 30
n_nodes_per_pipeline = 100
print("\n=== Generating multi-pipeline DAG ===")
start = time.time()
pipeline_graphs, combined_graph = generate_multi_pipeline_dag(
num_pipelines=n_pipelines,
nodes_per_pipeline=n_nodes_per_pipeline,
cross_pipeline_edges=2,
edge_probability=0.4,
seed=5,
)
# Create pipelines from each sub-graph
child_pipelines = []
all_trigger_nodes = []
for i, G in enumerate(pipeline_graphs):
nodes_i, _, triggers_i = dag_to_pipeline_nodes(G, params=params, seed=42 + i)
child_pipelines.append(Pipeline(nodes=nodes_i))
all_trigger_nodes.extend(triggers_i)
print(f"Pipeline {i}: {len(nodes_i)} nodes, {len(triggers_i)} triggers")
# Compose into parent pipeline
pipeline = Pipeline(
pipelines=child_pipelines,
max_workers=4,
state=state,
)
print(f"Generated DAG in {time.time() - start:.2f} seconds")
# Visualize the complete pipeline
if n_pipelines * n_nodes_per_pipeline < 1000:
pipeline.visualize("composed_pipeline.png")
# =============================================================================
# Run the pipeline by triggering source nodes
# =============================================================================
print(f"\n=== Triggering {len(all_trigger_nodes)} source nodes ===")
for i, trigger_node in enumerate(all_trigger_nodes):
print(f"Triggering node {i + 1}/{len(all_trigger_nodes)}: {trigger_node.name}")
pipeline.get_node_by_id(trigger_node.id).trigger()
print(f"State contains {len(list(pipeline._state.keys()))} values")
print("\n=== Pipeline execution complete ===")
backend/app/flow/test_controller.py
-194
View File
@@ -1,194 +0,0 @@
"""
Test script for the PipelineController.
This script demonstrates the controller's ability to:
1. Scan a nodes directory and build a pipeline
2. Watch for file changes and update the pipeline dynamically
3. Handle errors gracefully without breaking the pipeline
4. Accept node code from external sources (simulating frontend input)
"""
import asyncio
import sys
from pathlib import Path
# Add flow directory to path for imports
sys.path.insert(0, str(Path(__file__).parent))
from controller import PipelineController, NodeStatus
from state import RedisState, MemoryState
async def main():
"""Run the controller test."""
print("=" * 60)
print("Pipeline Controller Test")
print("=" * 60)
# Use Redis if available, otherwise fall back to memory
try:
state = RedisState(
host="localhost",
port=6379,
namespace="fluksio_controller_test",
ttl=3600,
)
if state.ping():
print("✓ Using Redis state backend")
else:
raise ConnectionError("Redis ping failed")
except Exception as e:
print(f"⚠ Redis not available ({e}), using MemoryState")
state = MemoryState()
# Create controller pointing to example nodes
nodes_dir = Path(__file__).parent / "nodes_example"
controller = PipelineController(
nodes_dir=nodes_dir,
state_backend=state,
max_workers=4,
node_params={"alert_threshold": 70.0}, # Custom param for alert node
)
# Register callbacks to see what's happening
controller.on_node_loaded(
lambda nid, node: print(
f" ✓ Loaded: {nid} ({len(node.requires)} inputs, {len(node.provides)} outputs)"
)
)
controller.on_node_error(
lambda nid, err: print(f" ✗ Error in {nid}: {err.split(chr(10))[0]}")
)
controller.on_node_removed(lambda nid: print(f" ⊘ Removed: {nid}"))
controller.on_pipeline_rebuilt(
lambda p: print(
f" ⟳ Pipeline rebuilt: {len(p.nodes)} nodes, {len(p.dependencies)} dependencies"
)
)
# Start the controller
print("\n--- Starting Controller ---")
await controller.start()
# Show current state
print("\n--- Current State ---")
print(f"Active nodes: {[n.name for n in controller.active_nodes]}")
print(f"Error nodes: {list(controller.error_nodes.keys())}")
if controller.pipeline:
print(f"\nPipeline dependency graph:")
for node, deps in controller.pipeline.dependencies.items():
dep_names = [d.name for d in deps] if deps else ["(trigger)"]
print(f" {node.name} <- {dep_names}")
# Test triggering nodes
print("\n--- Testing Node Triggers ---")
if controller.pipeline:
# Find trigger nodes (nodes with no dependencies)
trigger_nodes = [
n
for n in controller.active_nodes
if not controller.pipeline.dependencies.get(n)
]
print(f"Trigger nodes: {[n.name for n in trigger_nodes]}")
# Trigger each trigger node
for node in trigger_nodes:
print(f"\nTriggering: {node.name}")
try:
node.trigger()
except Exception as e:
print(f" Error: {e}")
# Show state after triggers
print(f"\nState after triggers:")
for key in controller.state_backend.keys():
value = controller.state_backend.get(key)
print(f" {key}: {value}")
# Test dynamic node addition
print("\n--- Testing Dynamic Node Addition ---")
new_node_code = '''
"""Dynamically added node that logs all sensor data."""
from nodes import Node
from util import Message
def create_node(params):
def log_data(params, temperature=None, humidity=None, comfort_index=None, **kwargs):
"""Log all available sensor data."""
print(f"[data_logger] Logging: T={temperature}, H={humidity}, Comfort={comfort_index}")
return {"log_entry": f"T={temperature}, H={humidity}, C={comfort_index}"}
return Node(
f=log_data,
requires=[
Message(name="temperature", dtype=float),
Message(name="humidity", dtype=float),
Message(name="comfort_index", dtype=float),
],
provides=[Message(name="log_entry", dtype=str)],
params=params,
name="data_logger",
)
'''
print("Adding 'data_logger' node from code...")
controller.add_node_from_code("data_logger", new_node_code)
# Wait for file watcher to pick up the change
await asyncio.sleep(1.0)
print(f"\nActive nodes after addition: {[n.name for n in controller.active_nodes]}")
# Test adding a broken node
print("\n--- Testing Error Handling ---")
broken_node_code = '''
"""This node has an intentional error."""
from nodes import Node
from util import Message
def create_node(params):
# This will cause a NameError
return undefined_variable_that_does_not_exist
'''
print("Adding 'broken_node' with intentional error...")
controller.add_node_from_code("broken_node", broken_node_code)
await asyncio.sleep(1.0)
print(
f"\nActive nodes (should exclude broken): {[n.name for n in controller.active_nodes]}"
)
print(f"Error nodes: {list(controller.error_nodes.keys())}")
# The pipeline should still work with the other nodes
if controller.pipeline:
print(f"\nPipeline still has {len(controller.pipeline.nodes)} working nodes")
# Clean up test nodes
print("\n--- Cleaning Up ---")
controller.remove_node("data_logger")
controller.remove_node("broken_node")
await asyncio.sleep(0.5)
print(f"Active nodes after cleanup: {[n.name for n in controller.active_nodes]}")
# Stop the controller
await controller.stop()
print("\n✓ Controller stopped")
if __name__ == "__main__":
asyncio.run(main())
backend/app/flow/test_http_node.py
-270
View File
@@ -1,270 +0,0 @@
"""
Test script for HttpNode functionality.
This script demonstrates both trigger (receiver) and sender modes of HttpNode,
including integration with FastAPI and the PipelineController.
"""
import asyncio
import sys
from pathlib import Path
# Add flow directory to path for imports
sys.path.insert(0, str(Path(__file__).parent))
from fastapi import FastAPI
from starlette.testclient import TestClient
from nodes import HttpNode, Node
from pipeline import Pipeline
from state import MemoryState
from util import Message
from controller import PipelineController
def test_http_node_modes():
"""Test HttpNode mode detection."""
print("=" * 60)
print("Testing HttpNode Mode Detection")
print("=" * 60)
# Trigger mode: only provides
trigger = HttpNode(
url="/api/data",
method="POST",
provides=[Message(name="value", dtype=float)],
)
print(f"✓ Trigger node mode: {trigger.mode}")
assert trigger.mode == HttpNode.Mode.TRIGGER
# Sender mode: has requires
sender = HttpNode(
url="https://httpbin.org/post",
method="POST",
requires=[Message(name="value", dtype=float)],
)
print(f"✓ Sender node mode: {sender.mode}")
assert sender.mode == HttpNode.Mode.SENDER
# Both requires and provides = sender mode
hybrid = HttpNode(
url="https://httpbin.org/post",
method="POST",
requires=[Message(name="input", dtype=float)],
provides=[Message(name="output", dtype=float)],
)
print(f"✓ Hybrid node mode: {hybrid.mode}")
assert hybrid.mode == HttpNode.Mode.SENDER
# Error: neither requires nor provides
try:
invalid = HttpNode(url="/api/nothing", method="GET")
assert False, "Should have raised ValueError"
except ValueError as e:
print(f"✓ Correctly rejected invalid config: {e}")
print("\n✓ All mode detection tests passed!\n")
def test_http_trigger_with_fastapi():
"""Test HTTP trigger node with FastAPI."""
print("=" * 60)
print("Testing HTTP Trigger with FastAPI")
print("=" * 60)
# Create FastAPI app
app = FastAPI()
# Create a simple pipeline with HTTP trigger
trigger = HttpNode(
url="/api/sensors/temperature",
method="POST",
provides=[
Message(name="temperature", dtype=float),
Message(name="sensor_id", dtype=str),
],
name="temp_trigger",
)
# Create a processing node
def process_temp(params, temperature=0.0, sensor_id="unknown", **kwargs):
celsius = temperature
fahrenheit = celsius * 9 / 5 + 32
print(f"[processor] Sensor {sensor_id}: {celsius}°C = {fahrenheit}°F")
return {"fahrenheit": fahrenheit}
processor = Node(
f=process_temp,
requires=[
Message(name="temperature", dtype=float),
Message(name="sensor_id", dtype=str),
],
provides=[Message(name="fahrenheit", dtype=float)],
params={},
name="temp_processor",
)
# Build pipeline
state = MemoryState()
pipeline = Pipeline(
nodes=[trigger, processor],
state=state,
max_workers=2,
)
# Register the HTTP route
trigger.register_route(app)
# Test with TestClient
client = TestClient(app)
print("\nSending POST request to /api/sensors/temperature...")
response = client.post(
"/api/sensors/temperature",
json={"temperature": 25.5, "sensor_id": "sensor_001"},
)
print(f"Response status: {response.status_code}")
print(f"Response body: {response.json()}")
assert response.status_code == 200
data = response.json()
assert data["status"] == "triggered"
assert data["node"] == "temp_trigger"
# Check pipeline state
print(f"\nPipeline state after trigger:")
for key in state.keys():
print(f" {key}: {state.get(key)}")
print("\n✓ HTTP trigger test passed!\n")
def test_http_sender():
"""Test HTTP sender node."""
print("=" * 60)
print("Testing HTTP Sender")
print("=" * 60)
# Create a sender node that posts to httpbin
sender = HttpNode(
url="https://httpbin.org/post",
method="POST",
requires=[
Message(name="temperature", dtype=float),
Message(name="humidity", dtype=float),
],
name="data_sender",
params=dict(
timeout=10.0,
headers={"X-Custom-Header": "fluksio-test"},
),
)
print(f"Sender node created: {sender.name}")
print(f" URL: {sender.url}")
print(f" Method: {sender.method}")
print(f" Mode: {sender.mode}")
# Note: Actual HTTP call would need async context or be mocked
# For this test, we just verify the node is configured correctly
print("\n✓ HTTP sender configuration test passed!\n")
async def test_controller_with_http_nodes():
"""Test PipelineController with HTTP nodes."""
print("=" * 60)
print("Testing Controller with HTTP Nodes")
print("=" * 60)
import tempfile
# Create FastAPI app
app = FastAPI()
with tempfile.TemporaryDirectory() as tmpdir:
nodes_dir = Path(tmpdir) / "nodes"
nodes_dir.mkdir()
# Create an HTTP trigger node file
trigger_code = """
from nodes import HttpNode
from util import Message
def create_node(params):
return HttpNode(
url="/api/webhook",
method="POST",
provides=[
Message(name="event_type", dtype=str),
Message(name="payload", dtype=object),
],
name="webhook_receiver",
)
"""
(nodes_dir / "webhook.py").write_text(trigger_code)
# Create controller with FastAPI app
controller = PipelineController(
nodes_dir=nodes_dir,
max_workers=2,
fastapi_app=app,
)
# Register callbacks
controller.on_node_loaded(
lambda nid, node: print(
f" ✓ Loaded: {nid} (mode: {getattr(node, 'mode', 'N/A')})"
)
)
# Start controller
print("\nStarting controller...")
await controller.start()
print(f"\nActive nodes: {[n.name for n in controller.active_nodes]}")
print(f"HTTP trigger nodes: {[n.name for n in controller.http_trigger_nodes]}")
print(f"HTTP sender nodes: {[n.name for n in controller.http_sender_nodes]}")
# Check that route was registered
route_paths = [r.path for r in app.routes if hasattr(r, "path")]
print(f"\nRegistered routes: {route_paths}")
if "/api/webhook" in route_paths:
print("✓ Webhook route registered!")
# Test the endpoint
client = TestClient(app)
response = client.post(
"/api/webhook",
json={"event_type": "test", "payload": {"key": "value"}},
)
print(f"Webhook response: {response.json()}")
# Stop controller
await controller.stop()
print("\n✓ Controller with HTTP nodes test passed!\n")
def main():
"""Run all tests."""
print("\n" + "=" * 60)
print("HttpNode Test Suite")
print("=" * 60 + "\n")
# Synchronous tests
test_http_node_modes()
test_http_trigger_with_fastapi()
test_http_sender()
# Async tests
asyncio.run(test_controller_with_http_nodes())
print("=" * 60)
print("All tests completed successfully!")
print("=" * 60)
if __name__ == "__main__":
main()
backend/app/flow/test_influxdb_node.py
-447
View File
@@ -1,447 +0,0 @@
"""
Test script for InfluxDbNode functionality.
This script demonstrates write and read operations with InfluxDB.
Requires an InfluxDB instance running.
To run a local InfluxDB instance:
docker run -d -p 8086:8086 \
-e DOCKER_INFLUXDB_INIT_MODE=setup \
-e DOCKER_INFLUXDB_INIT_USERNAME=admin \
-e DOCKER_INFLUXDB_INIT_PASSWORD=adminpass \
-e DOCKER_INFLUXDB_INIT_ORG=my-org \
-e DOCKER_INFLUXDB_INIT_BUCKET=sensors \
-e DOCKER_INFLUXDB_INIT_ADMIN_TOKEN=my-super-secret-token \
influxdb:2.7
"""
import asyncio
import sys
import time
from pathlib import Path
# Add flow directory to path for imports
sys.path.insert(0, str(Path(__file__).parent))
from nodes import InfluxDbNode, Node
from pipeline import Pipeline
from state import MemoryState
from util import Message
# Default test configuration
TEST_CONFIG = {
"url": "http://10.200.200.115:8086",
"token": "***REMOVED-INFLUXDB-TOKEN***==",
"org": "strobl",
"bucket": "test",
}
def test_influxdb_node_validation():
"""Test InfluxDbNode parameter validation."""
print("=" * 60)
print("Testing InfluxDbNode Validation")
print("=" * 60)
# Test missing required params
try:
node = InfluxDbNode(
requires=[Message(name="data", dtype=float)],
params={"url": "http://localhost:8086"}, # Missing token, org, bucket
)
assert False, "Should have raised ValueError"
except ValueError as e:
print(f"✓ Correctly rejected missing params: {e}")
# Test neither requires nor provides
try:
node = InfluxDbNode(params=TEST_CONFIG)
assert False, "Should have raised ValueError"
except ValueError as e:
print(f"✓ Correctly rejected empty requires/provides: {e}")
# Test valid write-only node
writer = InfluxDbNode(
requires=[Message(name="temperature", dtype=float)],
params={
**TEST_CONFIG,
"writes": {
"temperature": {
"measurement": "sensor_data",
"field": "temp_celsius",
"tags": {"location": "room1"},
}
},
},
)
print(f"✓ Created write-only node: {writer.name}")
assert len(writer.requires) == 1
assert len(writer.provides) == 0
# Test valid read-only node
reader = InfluxDbNode(
provides=[Message(name="avg_temperature", dtype=float)],
params={
**TEST_CONFIG,
"queries": {
"avg_temperature": {
"measurement": "sensor_data",
"field": "temp_celsius",
"aggregation": "mean",
}
},
},
)
print(f"✓ Created read-only node: {reader.name}")
assert len(reader.requires) == 0
assert len(reader.provides) == 1
# Test combined read/write node
combined = InfluxDbNode(
requires=[Message(name="raw_temp", dtype=float)],
provides=[Message(name="avg_temp", dtype=float)],
params={
**TEST_CONFIG,
"writes": {
"raw_temp": {
"measurement": "temperature",
"field": "value",
"tags": {"source": "sensor"},
}
},
"queries": {
"avg_temp": {
"measurement": "temperature",
"field": "value",
"aggregation": "mean",
}
},
},
)
print(f"✓ Created combined node: {combined.name}")
assert len(combined.requires) == 1
assert len(combined.provides) == 1
print("\n✓ All validation tests passed!\n")
def test_flux_query_building():
"""Test Flux query string generation."""
print("=" * 60)
print("Testing Flux Query Building")
print("=" * 60)
node = InfluxDbNode(
provides=[Message(name="value", dtype=float)],
params={
**TEST_CONFIG,
"queries": {
"value": {
"measurement": "temperature",
"field": "celsius",
"tags": {"location": "room1", "sensor": "dht22"},
"range": "-24h",
"aggregation": "mean",
}
},
},
)
query = node._build_flux_query(
measurement="temperature",
field="celsius",
tags={"location": "room1", "sensor": "dht22"},
time_range="-24h",
aggregation="mean",
)
print(f"Generated Flux query:\n{query}\n")
# Verify query components
assert 'from(bucket: "test")' in query
assert "range(start: -24h)" in query
assert 'r["_measurement"] == "temperature"' in query
assert 'r["_field"] == "celsius"' in query
assert 'r["location"] == "room1"' in query
assert 'r["sensor"] == "dht22"' in query
assert "mean()" in query
print("✓ Query contains all expected components")
# Test different aggregations
for agg in ["last", "first", "max", "min", "sum", "count"]:
query = node._build_flux_query(
measurement="test",
field="value",
tags={},
time_range="-1h",
aggregation=agg,
)
assert f"{agg}()" in query
print(f"✓ Aggregation '{agg}' works")
print("\n✓ All query building tests passed!\n")
def test_influxdb_write(skip_if_no_server: bool = True):
"""Test writing data to InfluxDB."""
print("=" * 60)
print("Testing InfluxDB Write")
print("=" * 60)
# Create writer node with write configuration
writer = InfluxDbNode(
requires=[Message(name="temperature", dtype=float)],
params={
**TEST_CONFIG,
"writes": {
"temperature": {
"measurement": "test_temperature",
"field": "value",
"tags": {"location": "test_room", "sensor": "test_sensor"},
}
},
},
name="test_writer",
)
# Build a simple pipeline
state = MemoryState()
pipeline = Pipeline(nodes=[writer], state=state, max_workers=1)
# Test data to write - now just a simple value!
test_data = {"temperature": 25.5}
print(f"Attempting to write: {test_data}")
try:
writer._write_points(test_data)
print("✓ Write successful!")
except Exception as e:
if skip_if_no_server:
print(f"⚠ Write failed (server may not be running): {e}")
else:
raise
# Test with runtime tags override
test_data_with_tags = {
"temperature": {"value": 26.0, "tags": {"sensor": "override_sensor"}}
}
print(f"Attempting to write with runtime tags: {test_data_with_tags}")
try:
writer._write_points(test_data_with_tags)
print("✓ Write with runtime tags successful!")
except Exception as e:
if skip_if_no_server:
print(f"⚠ Write failed (server may not be running): {e}")
else:
raise
print("\n✓ Write test completed!\n")
def test_influxdb_read(skip_if_no_server: bool = True):
"""Test reading data from InfluxDB."""
print("=" * 60)
print("Testing InfluxDB Read")
print("=" * 60)
# Create reader node
reader = InfluxDbNode(
provides=[
Message(name="last_temp", dtype=float),
Message(name="avg_temp", dtype=float),
],
params={
**TEST_CONFIG,
"query_range": "-1h",
"queries": {
"last_temp": {
"measurement": "test_temperature",
"field": "value",
"tags": {"location": "test_room"},
"aggregation": "last",
},
"avg_temp": {
"measurement": "test_temperature",
"field": "value",
"tags": {"location": "test_room"},
"aggregation": "mean",
},
},
},
name="test_reader",
)
print(f"Attempting to query data...")
try:
results = reader._query_data()
print(f"✓ Query successful! Results: {results}")
if results:
for key, value in results.items():
print(f" {key}: {value}")
else:
print(" No data found (this is OK if no data was written)")
except Exception as e:
if skip_if_no_server:
print(f"⚠ Query failed (server may not be running): {e}")
else:
raise
print("\n✓ Read test completed!\n")
def test_influxdb_pipeline_integration(skip_if_no_server: bool = True):
"""Test InfluxDbNode in a complete pipeline."""
print("=" * 60)
print("Testing InfluxDB Pipeline Integration")
print("=" * 60)
# Create a source node that generates sensor data (now just a float!)
def generate_sensor_data(params, **kwargs):
"""Generate test sensor data."""
import random
return {"temperature": 20.0 + random.random() * 10}
source = Node(
f=generate_sensor_data,
requires=[],
provides=[Message(name="temperature", dtype=float)],
params={},
name="sensor_source",
)
# Create InfluxDB writer that receives the sensor data
writer = InfluxDbNode(
requires=[Message(name="temperature", dtype=float)],
params={
**TEST_CONFIG,
"writes": {
"temperature": {
"measurement": "pipeline_test",
"field": "temp_value",
"tags": {
"source": "test_pipeline",
"run_id": str(int(time.time())),
},
}
},
},
name="influx_writer",
)
# Build pipeline
state = MemoryState()
pipeline = Pipeline(
nodes=[source, writer],
state=state,
max_workers=2,
)
print(f"Pipeline created with nodes: {[n.name for n in pipeline.nodes]}")
print(f"Dependencies: {pipeline.dependencies}")
try:
# Trigger the source node to start the pipeline
print("\nTriggering source node...")
result = source.inject({}) # Source has no inputs, just generates output
print(f"Pipeline execution result: {result}")
print("✓ Pipeline integration successful!")
except Exception as e:
if skip_if_no_server:
print(f"⚠ Pipeline test failed (server may not be running): {e}")
else:
raise
print("\n✓ Pipeline integration test completed!\n")
async def test_influxdb_async_operations(skip_if_no_server: bool = True):
"""Test async write and query operations."""
print("=" * 60)
print("Testing InfluxDB Async Operations")
print("=" * 60)
# Create a combined node with write and query configs
node = InfluxDbNode(
requires=[Message(name="temperature", dtype=float)],
provides=[Message(name="last_value", dtype=float)],
params={
**TEST_CONFIG,
"writes": {
"temperature": {
"measurement": "async_test",
"field": "value",
"tags": {"test": "async"},
}
},
"queries": {
"last_value": {
"measurement": "async_test",
"field": "value",
"aggregation": "last",
}
},
},
name="async_test_node",
)
try:
# Test async write - now just a simple value!
print("Testing async write...")
await node.write_async({"temperature": 42.0})
print("✓ Async write successful!")
# Wait a moment for the write to be visible
await asyncio.sleep(0.5)
# Test async query
print("Testing async query...")
results = await node.query_async()
print(f"✓ Async query successful! Results: {results}")
except Exception as e:
if skip_if_no_server:
print(f"⚠ Async test failed (server may not be running): {e}")
else:
raise
print("\n✓ Async operations test completed!\n")
def main():
"""Run all tests."""
print("\n" + "=" * 60)
print("InfluxDbNode Test Suite")
print("=" * 60 + "\n")
# Synchronous tests (always run)
test_influxdb_node_validation()
test_flux_query_building()
# Tests that require InfluxDB server
print("\n" + "-" * 60)
print("The following tests require InfluxDB on localhost:8086")
print("-" * 60 + "\n")
try:
test_influxdb_write(skip_if_no_server=True)
test_influxdb_read(skip_if_no_server=True)
test_influxdb_pipeline_integration(skip_if_no_server=True)
asyncio.run(test_influxdb_async_operations(skip_if_no_server=True))
except KeyboardInterrupt:
print("\nTests interrupted by user")
print("=" * 60)
print("All tests completed!")
print("=" * 60)
if __name__ == "__main__":
main()
backend/app/flow/test_mqtt_node.py
-334
View File
@@ -1,334 +0,0 @@
"""
Test script for MqttNode functionality.
This script demonstrates both subscriber (trigger) and publisher (sender) modes
of MqttNode. Requires an MQTT broker running (e.g., mosquitto).
To run a local mosquitto broker:
docker run -it -p 1883:1883 eclipse-mosquitto mosquitto -c /mosquitto-no-auth.conf
Or install locally:
sudo apt install mosquitto mosquitto-clients
sudo systemctl start mosquitto
"""
import asyncio
import sys
import json
from pathlib import Path
# Add flow directory to path for imports
sys.path.insert(0, str(Path(__file__).parent))
from nodes import MqttNode, Node
from pipeline import Pipeline
from state import MemoryState
from util import Message
host = "127.0.0.1"
def test_mqtt_node_modes():
"""Test MqttNode mode detection."""
print("=" * 60)
print("Testing MqttNode Mode Detection")
print("=" * 60)
# Subscriber mode: only provides
subscriber = MqttNode(
topic="sensors/temperature",
provides=[Message(name="temperature", dtype=float)],
params={"broker_host": host},
)
print(f"✓ Subscriber node mode: {subscriber.mode}")
assert subscriber.mode == MqttNode.Mode.SUBSCRIBER
# Publisher mode: has requires
publisher = MqttNode(
topic="actuators/hvac",
requires=[Message(name="target_temp", dtype=float)],
params={"broker_host": host},
)
print(f"✓ Publisher node mode: {publisher.mode}")
assert publisher.mode == MqttNode.Mode.PUBLISHER
# Both requires and provides = publisher mode
hybrid = MqttNode(
topic="devices/thermostat",
requires=[Message(name="input", dtype=float)],
provides=[Message(name="output", dtype=float)],
params={"broker_host": host},
)
print(f"✓ Hybrid node mode: {hybrid.mode}")
assert hybrid.mode == MqttNode.Mode.PUBLISHER
# Error: neither requires nor provides
try:
invalid = MqttNode(topic="invalid/topic", params={})
assert False, "Should have raised ValueError"
except ValueError as e:
print(f"✓ Correctly rejected invalid config: {e}")
print("\n✓ All mode detection tests passed!\n")
def test_mqtt_node_config():
"""Test MqttNode configuration from params."""
print("=" * 60)
print("Testing MqttNode Configuration")
print("=" * 60)
node = MqttNode(
topic="test/topic",
provides=[Message(name="value", dtype=float)],
params={
"broker_host": "mqtt.example.com",
"broker_port": 8883,
"username": "user",
"password": "secret",
"client_id": "test-client",
"qos": 2,
"retain": True,
"keepalive": 120,
},
)
assert node.broker_host == "mqtt.example.com"
assert node.broker_port == 8883
assert node.username == "user"
assert node.password == "secret"
assert node.client_id == "test-client"
assert node.qos == 2
assert node.retain is True
assert node.keepalive == 120
print(f"✓ broker_host: {node.broker_host}")
print(f"✓ broker_port: {node.broker_port}")
print(f"✓ username: {node.username}")
print(f"✓ qos: {node.qos}")
print(f"✓ retain: {node.retain}")
print(f"✓ keepalive: {node.keepalive}")
print("\n✓ Configuration test passed!\n")
async def test_mqtt_publisher():
"""Test MQTT publisher node (requires broker)."""
print("=" * 60)
print("Testing MQTT Publisher")
print("=" * 60)
publisher = MqttNode(
topic="test/fluksio/output",
requires=[
Message(name="temperature", dtype=float),
Message(name="humidity", dtype=float),
],
params={
"broker_host": host,
"broker_port": 1883,
"qos": 1,
},
name="test_publisher",
)
print(f"Publisher node created: {publisher.name}")
print(f" Topic: {publisher.topic}")
print(f" Mode: {publisher.mode}")
print(f" Broker: {publisher.broker_host}:{publisher.broker_port}")
# Try to publish (will fail if no broker is running)
try:
# Create a simple pipeline to bind the node
state = MemoryState()
pipeline = Pipeline(nodes=[publisher], state=state, max_workers=1)
# Directly call the publish method
await publisher._publish_message({"temperature": 25.5, "humidity": 60.0})
print("✓ Published message successfully!")
except Exception as e:
print(f"⚠ Could not publish (broker may not be running): {e}")
print("\n✓ Publisher test completed!\n")
async def test_mqtt_subscriber():
"""Test MQTT subscriber node (requires broker)."""
print("=" * 60)
print("Testing MQTT Subscriber")
print("=" * 60)
# Create a subscriber node
subscriber = MqttNode(
topic="test/fluksio/input",
provides=[
Message(name="value", dtype=float),
Message(name="unit", dtype=str),
],
params={
"broker_host": host,
"broker_port": 1883,
},
name="test_subscriber",
)
# Create a processing node that will be triggered
received_data = []
def process_data(params, value=0.0, unit="unknown", **kwargs):
print(f"[processor] Received: value={value}, unit={unit}")
received_data.append({"value": value, "unit": unit})
return {"processed": True}
processor = Node(
f=process_data,
requires=[
Message(name="value", dtype=float),
Message(name="unit", dtype=str),
],
provides=[Message(name="processed", dtype=bool)],
params={},
name="data_processor",
)
# Build pipeline
state = MemoryState()
pipeline = Pipeline(
nodes=[subscriber, processor],
state=state,
max_workers=2,
)
print(f"Subscriber node created: {subscriber.name}")
print(f" Topic: {subscriber.topic}")
print(f" Mode: {subscriber.mode}")
# Try to start subscription
try:
await subscriber.start_subscription()
print("✓ Subscription started!")
# Wait a bit for potential messages
print(" Listening for 3 seconds...")
await asyncio.sleep(3)
# Stop subscription
await subscriber.stop_subscription()
print("✓ Subscription stopped!")
if received_data:
print(f"✓ Received {len(received_data)} messages")
else:
print(" No messages received (publish to test/fluksio/input to test)")
except Exception as e:
print(f"⚠ Could not subscribe (broker may not be running): {e}")
print("\n✓ Subscriber test completed!\n")
async def test_mqtt_integration():
"""Test full MQTT pub/sub integration (requires broker)."""
print("=" * 60)
print("Testing MQTT Integration (Pub/Sub)")
print("=" * 60)
import aiomqtt
topic = "test/fluksio/integration"
# Create subscriber
subscriber = MqttNode(
topic=topic,
provides=[Message(name="sensor_value", dtype=float)],
params={"broker_host": host},
name="integration_subscriber",
)
# Track received messages
received_values = []
def track_value(params, sensor_value=0.0, **kwargs):
print(f"[tracker] Received sensor_value={sensor_value}")
received_values.append(sensor_value)
return {"tracked": sensor_value}
tracker = Node(
f=track_value,
requires=[Message(name="sensor_value", dtype=float)],
provides=[Message(name="tracked", dtype=float)],
params={},
name="value_tracker",
)
# Build pipeline
state = MemoryState()
pipeline = Pipeline(
nodes=[subscriber, tracker],
state=state,
max_workers=2,
)
try:
# Start subscriber
await subscriber.start_subscription()
print("✓ Subscriber started")
# Publish some test messages
async with aiomqtt.Client(hostname=host) as client:
for i in range(3):
value = 20.0 + i * 5
payload = json.dumps({"sensor_value": value})
await client.publish(topic, payload)
print(f" Published: {payload}")
await asyncio.sleep(0.5)
# Wait for messages to be processed
await asyncio.sleep(1)
# Stop subscriber
await subscriber.stop_subscription()
print("✓ Subscriber stopped")
# Check results
print(f"\nReceived values: {received_values}")
if len(received_values) == 3:
print("✓ All messages received and processed!")
else:
print(f"⚠ Expected 3 messages, got {len(received_values)}")
except Exception as e:
print(f"⚠ Integration test failed (broker may not be running): {e}")
print("\n✓ Integration test completed!\n")
def main():
"""Run all tests."""
print("\n" + "=" * 60)
print("MqttNode Test Suite")
print("=" * 60 + "\n")
# Synchronous tests (always run)
test_mqtt_node_modes()
test_mqtt_node_config()
# Async tests (require broker)
print("\n" + "-" * 60)
print(f"The following tests require an MQTT broker on {host}:1883")
print("-" * 60 + "\n")
try:
asyncio.run(test_mqtt_publisher())
asyncio.run(test_mqtt_subscriber())
asyncio.run(test_mqtt_integration())
except KeyboardInterrupt:
print("\nTests interrupted by user")
print("=" * 60)
print("All tests completed!")
print("=" * 60)
if __name__ == "__main__":
main()
backend/app/flow/test_synchronous_nodes.py
-512
View File
@@ -1,512 +0,0 @@
"""
Test synchronous node execution in pipelines.
This module tests the "synchronous" flag which ensures a node only executes
when ALL its inputs have changed since the last execution.
Synchronous nodes are useful for:
- Aggregation nodes that need all inputs to be "fresh"
- Nodes that should only process complete "batches" of data
- Avoiding redundant executions when multiple inputs update at different rates
"""
import sys
import threading
import time
from pathlib import Path
# Add flow directory to path for imports
sys.path.insert(0, str(Path(__file__).parent))
from state import MemoryState, RedisState
from pipeline import Pipeline
from nodes import Node
from util import Message
def test_synchronous_basic():
"""Test basic synchronous node behavior."""
print("=" * 60)
print("Testing Basic Synchronous Node Behavior")
print("=" * 60)
execution_log = []
def sensor_a(params, **kwargs):
value = params.get("value", 1.0)
execution_log.append(f"sensor_a -> {value}")
return {"temp_a": value}
def sensor_b(params, **kwargs):
value = params.get("value", 2.0)
execution_log.append(f"sensor_b -> {value}")
return {"temp_b": value}
def sync_processor(params, temp_a=0, temp_b=0, **kwargs):
result = temp_a + temp_b
execution_log.append(f"run_sync_processor({temp_a}, {temp_b}) -> {result}")
return {"combined": result}
def async_processor(params, temp_a=0, temp_b=0, **kwargs):
result = temp_a * temp_b
execution_log.append(f"run_async_processor({temp_a}, {temp_b}) -> {result}")
return {"product": result}
# Create nodes
node_a = Node(
f=sensor_a,
requires=[],
provides=[Message(name="temp_a", dtype=float)],
params={"value": 10.0},
name="sensor_a",
)
node_b = Node(
f=sensor_b,
requires=[],
provides=[Message(name="temp_b", dtype=float)],
params={"value": 20.0},
name="sensor_b",
)
# Synchronous node - only executes when BOTH inputs change
sync_node = Node(
f=sync_processor,
requires=[
Message(name="temp_a", dtype=float),
Message(name="temp_b", dtype=float),
],
provides=[Message(name="combined", dtype=float)],
params={"synchronous": True},
name="sync_processor",
)
# Non-synchronous node - executes whenever any input is available
async_node = Node(
f=async_processor,
requires=[
Message(name="temp_a", dtype=float),
Message(name="temp_b", dtype=float),
],
provides=[Message(name="product", dtype=float)],
params={"synchronous": False},
name="async_processor",
)
# Build pipeline
state = MemoryState()
pipeline = Pipeline(
nodes=[node_a, node_b, sync_node, async_node],
state=state,
max_workers=1, # Single worker for predictable ordering
)
print("\n--- Triggering sensor_a (first time) ---")
execution_log.clear()
node_a.inject({})
print(f"Execution log: {execution_log}")
print(f"State: temp_a={state.get('temp_a')}, temp_b={state.get('temp_b')}")
# Neither processor should run yet (temp_b missing)
assert "run_async_processor" not in str(
execution_log
), "async_processor ran too early"
assert "run_sync_processor" not in str(
execution_log
), "sync_processor ran too early"
print("✓ Neither processor ran (temp_b not yet available)")
print("\n--- Triggering sensor_b (first time) ---")
execution_log.clear()
node_b.inject({})
print(f"Execution log: {execution_log}")
# Both should run now (first time both inputs are available)
assert "run_async_processor" in str(
execution_log
), "async_processor should have run"
assert "run_sync_processor" in str(execution_log), "sync_processor should have run"
print("✓ Both processors ran (first time both inputs available)")
print("\n--- Triggering sensor_a (second time) ---")
node_a.params["value"] = 15.0
execution_log.clear()
node_a.inject({})
print(f"Execution log: {execution_log}")
# async_processor SHOULD run (any input change triggers it)
# sync_processor should NOT run (only temp_a changed, not temp_b)
assert "run_async_processor" in str(
execution_log
), "async_processor should have run"
assert "run_sync_processor" not in str(
execution_log
), "sync_processor should NOT have run (only temp_a changed)"
print("✓ Only async_processor ran (temp_b hasn't changed)")
print("\n--- Triggering sensor_b (second time) ---")
node_b.params["value"] = 25.0
execution_log.clear()
node_b.inject({})
print(f"Execution log: {execution_log}")
# Both should run now (both inputs have changed since last sync execution)
assert "run_async_processor" in str(
execution_log
), "async_processor should have run"
assert "run_sync_processor" in str(
execution_log
), "sync_processor should have run (both inputs changed)"
print("✓ Both processors ran (both inputs have changed)")
# Verify final state values
assert (
pipeline.state.get("combined") == 40.0
), f"Expected combined=40.0, got {pipeline.state.get('combined')}"
assert (
pipeline.state.get("product") == 375.0
), f"Expected product=375.0, got {pipeline.state.get('product')}"
print(
f"\n✓ Final state: combined={pipeline.state.get('combined')}, product={pipeline.state.get('product')}"
)
print("\n✓ Basic synchronous test passed!\n")
def test_synchronous_multiple_triggers():
"""Test that synchronous nodes handle rapid successive triggers correctly."""
print("=" * 60)
print("Testing Synchronous Node with Multiple Rapid Triggers")
print("=" * 60)
sync_execution_count = {"count": 0}
async_execution_count = {"count": 0}
def source_x(params, **kwargs):
return {"x": params.get("value", 1)}
def source_y(params, **kwargs):
return {"y": params.get("value", 2)}
def sync_consumer(params, x=0, y=0, **kwargs):
sync_execution_count["count"] += 1
return {"sync_out": x + y}
def async_consumer(params, x=0, y=0, **kwargs):
async_execution_count["count"] += 1
return {"async_out": x * y}
node_x = Node(
f=source_x,
requires=[],
provides=[Message(name="x", dtype=int)],
params={"value": 1},
name="source_x",
)
node_y = Node(
f=source_y,
requires=[],
provides=[Message(name="y", dtype=int)],
params={"value": 1},
name="source_y",
)
sync_node = Node(
f=sync_consumer,
requires=[Message(name="x", dtype=int), Message(name="y", dtype=int)],
provides=[Message(name="sync_out", dtype=int)],
params={"synchronous": True},
name="sync_consumer",
)
async_node = Node(
f=async_consumer,
requires=[Message(name="x", dtype=int), Message(name="y", dtype=int)],
provides=[Message(name="async_out", dtype=int)],
params={"synchronous": False},
name="async_consumer",
)
state = MemoryState()
pipeline = Pipeline(
nodes=[node_x, node_y, sync_node, async_node],
state=state,
max_workers=1,
)
# Initial trigger to get both inputs
node_x.inject({})
node_y.inject({})
initial_sync = sync_execution_count["count"]
initial_async = async_execution_count["count"]
print(f"After initial triggers: sync={initial_sync}, async={initial_async}")
assert initial_sync == 1, "sync_consumer should have run once initially"
assert initial_async == 1, "async_consumer should have run once initially"
# Now trigger X multiple times without triggering Y
print("\nTriggering source_x 5 times without changing source_y...")
for i in range(5):
node_x.params["value"] = 10 + i
node_x.inject({})
final_sync = sync_execution_count["count"]
final_async = async_execution_count["count"]
print(f"After 5 more X triggers: sync={final_sync}, async={final_async}")
# sync_consumer should NOT have run again (Y didn't change)
assert (
final_sync == initial_sync
), f"sync_consumer should still be at {initial_sync}, got {final_sync}"
# async_consumer should have run 5 more times
assert (
final_async == initial_async + 5
), f"async_consumer should be at {initial_async + 5}, got {final_async}"
print("✓ Synchronous node correctly waited for both inputs to change")
# Now trigger Y once - sync should run
print("\nTriggering source_y once...")
node_y.params["value"] = 100
node_y.inject({})
after_y_sync = sync_execution_count["count"]
after_y_async = async_execution_count["count"]
print(f"After Y trigger: sync={after_y_sync}, async={after_y_async}")
assert after_y_sync == initial_sync + 1, f"sync_consumer should have run once more"
assert after_y_async == final_async + 1, f"async_consumer should have run once more"
print("\n✓ Multiple triggers test passed!\n")
def test_synchronous_race_condition():
"""Test that synchronous nodes handle concurrent triggers correctly."""
print("=" * 60)
print("Testing Synchronous Node Race Condition Handling")
print("=" * 60)
execution_count = {"sync": 0, "async": 0}
execution_lock = threading.Lock()
def source_a(params, **kwargs):
return {"data_a": params.get("value", 1)}
def source_b(params, **kwargs):
return {"data_b": params.get("value", 2)}
def sync_consumer(params, data_a=0, data_b=0, **kwargs):
with execution_lock:
execution_count["sync"] += 1
time.sleep(0.01) # Small delay to increase chance of race conditions
return {"sync_result": data_a + data_b}
def async_consumer(params, data_a=0, data_b=0, **kwargs):
with execution_lock:
execution_count["async"] += 1
return {"async_result": data_a * data_b}
node_a = Node(
f=source_a,
requires=[],
provides=[Message(name="data_a", dtype=int)],
params={"value": 10},
name="source_a",
)
node_b = Node(
f=source_b,
requires=[],
provides=[Message(name="data_b", dtype=int)],
params={"value": 20},
name="source_b",
)
sync_node = Node(
f=sync_consumer,
requires=[Message(name="data_a", dtype=int), Message(name="data_b", dtype=int)],
provides=[Message(name="sync_result", dtype=int)],
params={"synchronous": True},
name="sync_consumer",
)
async_node = Node(
f=async_consumer,
requires=[Message(name="data_a", dtype=int), Message(name="data_b", dtype=int)],
provides=[Message(name="async_result", dtype=int)],
params={"synchronous": False},
name="async_consumer",
)
state = MemoryState()
pipeline = Pipeline(
nodes=[node_a, node_b, sync_node, async_node],
state=state,
max_workers=4, # Multiple workers for concurrency
)
# Trigger both sensors multiple times concurrently
num_rounds = 5
print(f"\nTriggering both sensors {num_rounds} times concurrently...")
threads = []
for i in range(num_rounds):
node_a.params["value"] = 10 + i
node_b.params["value"] = 20 + i
t1 = threading.Thread(target=lambda: node_a.inject({}))
t2 = threading.Thread(target=lambda: node_b.inject({}))
threads.extend([t1, t2])
t1.start()
t2.start()
# Wait for all threads
for t in threads:
t.join()
# Give pipeline time to process
time.sleep(0.5)
print(f"\nExecution counts:")
print(f" sync_consumer: {execution_count['sync']}")
print(f" async_consumer: {execution_count['async']}")
# sync_consumer should execute at most num_rounds times
# (could be less if some triggers happened before the other input updated)
assert (
execution_count["sync"] <= num_rounds + 1
), f"sync_consumer ran too many times: {execution_count['sync']} (expected <= {num_rounds + 1})"
# async_consumer will run more frequently
assert (
execution_count["async"] >= execution_count["sync"]
), "async_consumer should run at least as often as sync_consumer"
print(
f"\n✓ sync_consumer ran {execution_count['sync']} times (max expected: {num_rounds + 1})"
)
print(f"✓ async_consumer ran {execution_count['async']} times")
print("\n✓ Race condition test passed!\n")
def test_state_backend_atomic_operations():
"""Test the atomic operations used for synchronous node support."""
print("=" * 60)
print("Testing State Backend Atomic Operations")
print("=" * 60)
state = MemoryState()
# Test increment
print("\nTesting increment...")
assert state.increment("counter") == 1
assert state.increment("counter") == 2
assert state.increment("counter") == 3
print("✓ increment works correctly")
# Test get_multi
print("\nTesting get_multi...")
state.set("a", 1)
state.set("b", 2)
state.set("c", 3)
result = state.get_multi(["a", "b", "c", "missing"])
assert result == {"a": 1, "b": 2, "c": 3, "missing": None}
print("✓ get_multi works correctly")
# Test compare_and_swap_multi - success case
print("\nTesting compare_and_swap_multi (success)...")
success = state.compare_and_swap_multi(
expected={"a": 1, "b": 2}, updates={"a": 10, "b": 20, "new_key": 100}
)
assert success, "compare_and_swap_multi should succeed"
assert state.get("a") == 10
assert state.get("b") == 20
assert state.get("new_key") == 100
print("✓ compare_and_swap_multi succeeded and applied updates")
# Test compare_and_swap_multi - failure case
print("\nTesting compare_and_swap_multi (failure)...")
success = state.compare_and_swap_multi(
expected={"a": 1, "b": 20}, updates={"a": 999, "b": 999} # a is now 10, not 1
)
assert not success, "compare_and_swap_multi should fail"
assert state.get("a") == 10, "a should be unchanged"
assert state.get("b") == 20, "b should be unchanged"
print("✓ compare_and_swap_multi correctly rejected mismatched expectation")
print("\n✓ State backend atomic operations test passed!\n")
def test_concurrent_compare_and_swap():
"""Test that compare_and_swap_multi handles concurrent access correctly."""
print("=" * 60)
print("Testing Concurrent Compare-and-Swap")
print("=" * 60)
state = MemoryState()
state.set("version", 0)
success_count = {"count": 0}
failure_count = {"count": 0}
count_lock = threading.Lock()
def try_swap(thread_id: int):
"""Try to atomically increment the version."""
for _ in range(10):
current = state.get("version")
success = state.compare_and_swap_multi(
expected={"version": current}, updates={"version": current + 1}
)
with count_lock:
if success:
success_count["count"] += 1
else:
failure_count["count"] += 1
# Start multiple threads trying to increment
threads = [threading.Thread(target=try_swap, args=(i,)) for i in range(5)]
for t in threads:
t.start()
for t in threads:
t.join()
final_version = state.get("version")
total_attempts = success_count["count"] + failure_count["count"]
print(f"Total attempts: {total_attempts}")
print(f"Successful swaps: {success_count['count']}")
print(f"Failed swaps (race lost): {failure_count['count']}")
print(f"Final version: {final_version}")
# The final version should equal the number of successful swaps
assert (
final_version == success_count["count"]
), f"Version mismatch: {final_version} != {success_count['count']}"
print("\n✓ Concurrent compare-and-swap test passed!\n")
def main():
"""Run all synchronous node tests."""
print("\n" + "=" * 60)
print("Synchronous Node Test Suite")
print("=" * 60 + "\n")
# Run all tests
test_state_backend_atomic_operations()
test_concurrent_compare_and_swap()
test_synchronous_basic()
test_synchronous_multiple_triggers()
test_synchronous_race_condition()
print("=" * 60)
print("All synchronous node tests completed!")
print("=" * 60)
if __name__ == "__main__":
main()
backend/app/flow/util.py
-47
View File
@@ -1,47 +0,0 @@
from pydantic import BaseModel, Field, field_validator, ConfigDict
from typing import Any, Tuple, Type
import hashlib
class Message(BaseModel):
"""Defines a message type with validation constraints."""
model_config = ConfigDict(frozen=True) # Immutable for hashability
name: str
dtype: Type = float
shape: Tuple[int, ...] = ()
vrange: Tuple[float, float] = (0.0, 1.0)
@property
def id(self) -> str:
return hashlib.sha256(self.name.encode()).hexdigest()[:12]
def check(self, value: Any) -> None:
if not isinstance(value, self.dtype):
raise TypeError(
f"{self.name}: expected {self.dtype.__name__}, got {type(value).__name__}"
)
def __repr__(self) -> str:
return f"Message({self.name})"
def __hash__(self) -> int:
return hash(self.name)
class NodeParams(BaseModel):
"""Parameters passed to node functions."""
model_config = ConfigDict(arbitrary_types_allowed=True)
params: dict[str, Any] = Field(default_factory=dict)
def __getitem__(self, key: str) -> Any:
return self.params[key]
def __contains__(self, key: str) -> bool:
return key in self.params
def get(self, key: str, default: Any = None) -> Any:
return self.params.get(key, default)
backend/app/main.py
+38
View File
@@ -1,3 +1,7 @@
import asyncio
from collections.abc import AsyncIterator
from contextlib import asynccontextmanager
import sentry_sdk
from fastapi import FastAPI
from fastapi.routing import APIRoute
@@ -5,6 +9,11 @@ from starlette.middleware.cors import CORSMiddleware
from app.api.main import api_router
from app.core.config import settings
from app.flow.controller import FlowController
from app.flow.events import event_bus
from app.flow.secrets import init_secrets
from app.flow.state import MemoryState, RedisState, StateBackend
from app.flow.store import FlowStore
def custom_generate_unique_id(route: APIRoute) -> str:
@@ -14,10 +23,39 @@ def custom_generate_unique_id(route: APIRoute) -> str:
if settings.SENTRY_DSN and settings.ENVIRONMENT != "local":
sentry_sdk.init(dsn=str(settings.SENTRY_DSN), enable_tracing=True)
def _state_backend() -> StateBackend:
if settings.REDIS_HOST:
return RedisState(host=settings.REDIS_HOST, port=settings.REDIS_PORT)
return MemoryState()
@asynccontextmanager
async def lifespan(app: FastAPI) -> AsyncIterator[None]:
"""Start the flow engine alongside the API."""
event_bus.bind(asyncio.get_running_loop())
init_secrets(settings.SECRETS_FILE, settings.SECRET_KEY)
controller = FlowController(
store=FlowStore(settings.FLOWS_DIR),
state=_state_backend(),
events=event_bus,
max_workers=settings.FLOW_MAX_WORKERS,
fastapi_app=app,
)
app.state.flow_controller = controller
await controller.start()
try:
yield
finally:
await controller.stop()
app = FastAPI(
title=settings.PROJECT_NAME,
openapi_url=f"{settings.API_V1_STR}/openapi.json",
generate_unique_id_function=custom_generate_unique_id,
lifespan=lifespan,
)
# Set all CORS enabled origins