juturna.components package

Subpackages

Module contents

class juturna.components.Message(creator: str | None = None, version: int = -1, payload: T_Input = None, timers_from: Message = None)

Bases: Generic

A message is a container object that all nodes produce and read to and from buffers, plus offering few extra utilities.

created_at
creator
id
meta
property payload: T_Input

Returns the payload of the message.

timeit(timer_name: str) → Iterator[None]

Time the enclosed block, and store its duration in seconds as a timer with the given name when the block exits. The duration is measured with a monotonic clock.

Parameters:

timer_name (str) – The name of the timer.

timer(timer_name: str, timer_value: float | None = None)

Records a timer with the given name and value. If no value is provided, the current time is used.

Parameters:

  • timer_name (str) – The name of the timer.

  • timer_value (float, optional) – The value of the timer. If None, the current time is used.

timers
to_dict() → dict

Convert the message to a dictionary representation.

Returns:

A dictionary representation of the message.

Return type:

dict

to_json(encoder: Callable | None = None, indent: int | None = None) → str

Convert the message to a JSON string. A custom encoder can be provided to serialise non-serialisable content, otherwise the default serialize method defined on the payload will be used.

Parameters:

  • encoder (callable, optional) – A custom JSON encoder. The default is None, which uses the default JSON encoder.

  • indent (int) – Indentation level for the serialisation, Defaulted to None.

Returns:

The JSON string representation of the message.

Return type:

str

version
class juturna.components.Node(node_name: str = '', pipe_name: str = '', transport: TransportBackend | None = None)

Bases: Generic

Use this class to design custom nodes. BaseNode comes with a number of utility methods and fields that can be either used as they are or extended in the derived classes.

SKIP_UPDATE = <object object>
add_destination(name: str, destination: Node)
clear_destination(name: str)
clear_destinations()
clear_source()
compile_template(template_name: str, arguments: dict) → str

Compile a template string

Parameters:

  • template_name (str) – Path of the template.

  • arguments (dict) – Dictionary of template anrguments and their values.

Returns:

Compiled template string.

Return type:

str

property configuration: dict
configure()
property destinations: list
destroy()
dump_json(message: Message, file_name: str) → str | None
flush_telemetry()

Send any buffered telemetry entries below the batch size

property health: dict

Counter of how many failues the node experienced across its threads. Values are split per-thread, so it will give the total for update, worker and source (if present).

join()

Wait for all internal threads to terminate. This method should be called after stop() to ensure the node has fully shut down before its resources are released.

kill()

Stop the node without processing pending data. All threads are signalled at once and exit as soon as their current iteration is over: any message still in the input queue is discarded. Messages a node keeps in its state are left there. Subclass cleanup in stop() still runs.

Unlike stop(), kill() does not wait for a busy update() or source call to return: each thread gets JUTURNA_THREAD_JOIN_TIMEOUT seconds, then it is left behind (it exits on its own as soon as the call returns, without processing anything else).

Pick the state passed to update() for each message.

By default every message is processed with the state linked by link_state(). A node shared by several pipelines (the remote service) needs a different state per message instead: the resolver is called with the message about to be processed, on the update thread, and returns its state, or Node.SKIP_UPDATE to drop the message without calling update(). None restores the default.

Get notified every time update() returns.

The observer is called on the update thread, after update() has processed a message, with that message and the exception it raised, or None. Messages skipped by the state resolver are not reported. The remote service uses it to answer a request when its update is over, whether it transmitted something or not. None removes it.

property origins: list
prepare_template(template_name: str, file_destination_name: str, arguments: dict) → Path

Fetch a template file from the node folder, compile it, and save the produced file to the node pipeline folder. The template will be compiled with basic substitution of the passed arguments.

Parameters:

  • template_name (str) – The name of the template file to retrieve in the node folder.

  • file_destination_name (str) – The name of the destination file in the node pipeline folder.

  • arguments (dict) – The argument values to substitute in the template file.

Returns:

The path of the filed compiled and saved from the template.

Return type:

pathlib.Path

Raises:

ValueError – If the node is not part of a pipeline, and pipe_path is not set.

put(message: Message)
set_on_config(prop: str, value: Any)
set_source(source: Callable, by: float = 0, mode: str = 'post')

Set the node source (to be used for source nodes). The source is a callable invoked periodically: the message it returns is fed to the node, while None means there is nothing to deliver this time.

Parameters:

  • source (callable) – The function producing the messages of the node.

  • by (float, optional) – The time interval (in seconds) between two consecutive calls to the source. With the default of 0 the source is called again right away, so it is expected to block until data is available.

  • mode (str, optional) – Whether to apply the by timer before or after the source call. The default is post: the source is called, then the node waits for by seconds before delivering the message (or before the next call, if the source returned None). If set to pre, the node waits for by seconds before every call.

Notes

An exception raised by the source is logged and counted in health, and the next call is delayed by an exponential backoff starting at by (at least 0.1 seconds) and capped to JUTURNA_SOURCE_MAX_BACKOFF, which resets at the first successful call.

signal_kill()

Signal all the node threads to exit without processing pending data. This method does not block: use kill() to wait for the threads to terminate and discard pending data.

start()

Start the node and begin processing. This method is called automatically when the parent pipeline is started. If you override this method in your custom node class, make sure to call the parent method to ensure the node is started correctly.

property static_path: Path

Path to the directory where the node is defined. This is useful for storing static files (e.g. configuration files) that are needed by the node.

property status: ComponentStatus | None
stop()

Stop the node and begin processing. This method is called automatically when the parent pipeline is stopped. If you override this method in your custom node class, make sure to call the parent method to ensure the node is stopped correctly.

Threads are stopped one at a time: the source first, then the worker once the source has terminated. The worker consumes the messages left in the input queue before exiting, so all the messages received before the stop call are processed, and stop() waits for an update() call in progress to return, however long it takes.

When the node is being killed (see kill()), each thread is only waited for JUTURNA_THREAD_JOIN_TIMEOUT seconds: a thread still busy after that is left behind, and listed in health under stuck_workers.

transmit(message: Message)

Transmit a message. This method is used to send data from the node to its destinations. Messages are frozen before transmission, so that immutability is ensured.

Parameters:

message (Message | None) – The message to be transmitted.

update(message: Message, **kwargs)

Process a message. Override this method in your custom node class.

Messages are delivered one at a time, in the order they reach the input queue, whatever node sent them (message.creator tells which). A node that needs several messages before producing an output, for example to join the outputs of more upstream nodes, keeps the pending ones in kwargs['state'], and removes them once they are used.

Parameters:

  • message (Message) – The message received from an upstream node.

  • kwargs (dict) – state is the State the pipeline keeps for the node across messages.

warmup()
class juturna.components.Pipeline(config: dict, worker: str = 'main')

Bases: object

A pipeline aggregates nodes to create a data workflow from a source node to destination nodes.

property DAG: DAG
destroy()

Destroy the pipeline and all its nodes.

This method cleans up all the resources used by the pipeline and its nodes. It clears the source and destination buffers for each node, destroys the nodes, and removes them from the pipeline. This is important to ensure that all resources are properly released and that there are no memory leaks. The pipeline is set to None, and garbage collection is triggered to free up any remaining resources.

destroy() is legal from every status except DESTROYED itself, on which it is a semi-idempotent no-op: repeated cleanup calls (e.g. from a caller’s try/finally) do not raise. Like the other lifecycle methods, it claims the pipeline before doing any work, so a concurrent call raises PipelineBusyError instead of racing on _nodes.

static from_json(json_path: str, worker: str = 'main') → Pipeline

Create a pipeline starting from the path of the configuration file rather than from the actual configuration content.

Parameters:

  • json_path (str) – The path to the pipeline configuration file.

  • worker (str) – The worker this pipeline instance runs in, see Pipeline.

Returns:

The pipeline object.

Return type:

Pipeline

kill()

Kill the pipeline and all its nodes.

Unlike stop(), pending messages in the nodes are not processed: all nodes are signalled at once, then each one is joined and its input queue and buffer are discarded.

property name: str
property pipe_id: str
property pipe_path: str
start()

Start the pipeline and all its nodes.

This method starts all the nodes in the pipeline, allowing them to process data. The nodes are started in reverse order of their configuration, ensuring that the source node is the last one to be started. This is important to ensure that the data flow is properly established and that all nodes are ready to receive data.

If a node fails to start, the nodes already started (including the failing one) are killed and telemetry is stopped before the error is raised: the pipeline stays READY with no worker running, so start() can be retried, or the pipeline destroyed.

property status: dict
stop()

Stop the pipeline and all its nodes.

This method stops all the nodes in the pipeline, sending a stopping signal to each node that will cause them to stop processing upcoming data and to propagate the stopping signal to their destinations

update_node(node_name: str, property_name: str, property_value: Any)
warmup()

Prepare the pipeline and all its nodes.

This method creates all the concrete nodes in the pipe, allocating their required resources. If a node is flagged as to be deployed remotely, that node will be replaced with a proc warp node.

Only the nodes of this pipeline’s worker are built. A link from one of them to a node of another worker delivers messages to the object returned by the transport’s remote_destination(node_name), so the transport must provide it; a link from another worker’s node only adds that node to the origins of the local one. Cycles are checked on the local graph only.

Node names and the telemetry file name are joined to the pipeline folder, so they must be plain file names (see jt_utils.check_name): a ValueError is raised before anything is written otherwise. So is a link that is repeated, or that names a node missing from the configuration.

If warmup fails, the nodes built so far are destroyed and the pipeline stays NEW with nothing built, so warmup() can be retried.

exception juturna.components.PipelineAlreadyRunningError(pipeline_name: str, message: str)

Bases: PipelineStateError

Raised when warmup() is called on a pipeline that is running.

exception juturna.components.PipelineBusyError(pipeline_name: str, message: str)

Bases: PipelineStateError

Raised when a lifecycle operation is called while another one is already in progress on the same pipeline.

Only one of warmup()/start()/stop()/destroy() can be claimed at a time: the transition in flight owns the pipeline’s internal state (nodes, DAG, node threads) until it completes, so a concurrent call cannot be safely evaluated against the pipeline’s current status.

exception juturna.components.PipelineDestroyedError(pipeline_name: str, message: str)

Bases: PipelineStateError

Raised when a lifecycle operation is called on a destroyed pipeline.

destroy() itself is exempted: calling it on an already destroyed pipeline is a no-op.

exception juturna.components.PipelineNotReadyError(pipeline_name: str, message: str)

Bases: PipelineStateError

Raised when start() is called before the pipeline was warmed up.

exception juturna.components.PipelineNotRunningError(pipeline_name: str, message: str)

Bases: PipelineStateError

Raised when stop() is called on a pipeline that is not running.

exception juturna.components.PipelineStateError(pipeline_name: str, message: str)

Bases: RuntimeError

Base exception for illegal Pipeline lifecycle transitions.

exception juturna.components.PipelineStoppedError(pipeline_name: str, message: str)

Bases: PipelineStateError

Raised when warmup() or start() is called on a stopped pipeline.

A stopped pipeline cannot be resumed: it must be destroyed and a new pipeline created in its place.

class juturna.components.State(*args, **kwargs)

Bases: UserDict