Nodes and Flows

This page describes the building blocks of a Videoflow application and the rules that make a graph deployable in a distributed setting.

The three node types

Type	Base class	Method to implement	Role
Producer	`ProducerNode`	`next()`	Creates data from an external source (file, camera, stream). Set `is_finite=False` for unbounded ones.
Processor	`ProcessorNode`	`process(*inputs)`	Transforms inputs into an output. Supports `nb_tasks`, `device_type`, `partition_by`, `join_policy`.
Consumer	`ConsumerNode`	`consume(item)`	Terminal sink; produces no output. Supports `idempotent` and `join_policy`.

Every node also has open() and close() lifecycle hooks for acquiring and releasing resources. Any lifecycle or processing method may be an async def (the worker awaits it) and may take a final ctx parameter to receive a runtime context — see Writing your own components.

Node identity: the `name`

Each node has a stable, unique string name. In a distributed flow that name is the node’s identity everywhere it matters: the broker subjects it publishes and subscribes to, and the Kubernetes resources generated for it. If you do not pass a name, one is generated from the class name, but for anything you intend to deploy you should give each node an explicit, meaningful name:

detector = ObjectDetector(name='detector')(frame)

Names must be unique within a flow; a duplicate raises a ValueError when the flow is built.

Constructor arguments must be serializable

Because a worker reconstructs only its one node from configuration (it never receives a live Python object), a node’s constructor arguments must be JSON-serializable — numbers, strings, booleans, and lists/dicts of those.

Two consequences follow:

Do heavy or stateful setup in open(), not __init__. Opening a camera, loading a model, or creating a database connection belongs in open(), which runs inside the worker just before processing begins.
Store each constructor argument on self under the same name so it can be captured automatically for reconstruction. Always accept and forward **kwargs to super().__init__().

See Writing your own components for full examples.

Building a Flow

A Flow is created from the consumers (the leaves of the graph); producers are discovered automatically:

from videoflow.core import Flow
from videoflow.core.constants import REALTIME

flow = Flow([consumer_a, consumer_b], flow_type=REALTIME, flow_id='my-flow')

flow_type is REALTIME (drop stale messages, never block producers) or BATCH (at-least-once, loss-free delivery). See Batch versus realtime mode.
flow_id namespaces the broker subjects and Kubernetes resources; pass a stable value when you want to redeploy the same logical flow.

When the flow is built it validates the graph: it must be acyclic, every consumer must be reachable from a producer, all names must be unique, and a replicated multi-parent join node must set partition_by (see Scaling and node allocation).

Running a Flow

A flow runs through an execution engine:

from videoflow.engines.local import LocalProcessEngine
flow.run(LocalProcessEngine())   # local subprocesses; or KubernetesExecutionEngine
flow.join()                       # block until done
# flow.stop()                     # signal termination early

flow.run(engine) mints a fresh run_id and is non-blocking. flow.join() blocks until the flow finishes naturally (all producers exhausted). flow.stop() publishes a termination signal on the broker’s control channel that every worker is subscribed to, then waits for them to drain and exit.

See Distributed execution for the available engines.

Reliability

Each call to flow.run() is scoped by a run id, so re-running a flow gets a fresh set of broker streams instead of colliding with a previous run.

Delivery is at-least-once: a worker acknowledges a message only after it has processed it (and published its output), so a crash mid-processing causes redelivery, not loss. In BATCH mode a message that keeps failing is retried a few times and then dead-lettered to a DLQ stream (vf-<flow>-<run>-dlq) with the error attached, instead of being dropped or crashing the pod. See Distributed execution and Batch versus realtime mode.