WriteStream#

class nominal.core.WriteStream(batch_size, max_wait)#

Bases: WriteStreamBase[StreamType]

A buffered asynchronous writer for time-series or log data.

batch_size: int#
max_wait: timedelta#
close(wait=True)#

Close the Nominal Stream.

Stop the process timeout thread and flush any remaining batches.

classmethod create(batch_size, max_wait, process_batch)#

Create the stream.

Parameters:

  • batch_size (int)

    Maximum number of items to batch before flushing.

  • max_wait (timedelta)

    Maximum time to wait before flushing a batch.

  • process_batch (Callable[[Sequence[BatchItem[StreamType]]], None])

    Callable to process batches of items.

Return type:

Self

enqueue(channel_name, timestamp, value, tags=None)#

Add a message to the queue after normalizing the timestamp to IntegralNanosecondsUTC.

The message is added to the thread-safe batch and flushed if the batch size is reached.

enqueue_batch(channel_name, timestamps, values, tags=None)#

Add a sequence of messages to the queue to upload to Nominal.

Messages are added one-by-one (with timestamp normalization) and flushed based on the batch conditions.

Parameters:

  • channel_name (str)

    Name of the channel to upload data for.

  • timestamps (Sequence[str | datetime | int])

    Absolute timestamps of the data being uploaded.

  • values (Sequence[StreamType])

    Values to write to the specified channel.

  • tags (Mapping[str, str] | None)

    Key-value tags associated with the data being uploaded.

    Note: This must include all required_tags configured for the dataset.

enqueue_float_array(channel_name, timestamp, value, tags=None)#

Add an array of floats to the queue after normalizing the timestamp.

Parameters:

  • channel_name (str)

    Name of the channel to upload data for.

  • timestamp (str | datetime | int)

    Absolute timestamp of the data being uploaded.

  • value (list[float])

    List of float values to write to the specified channel.

  • tags (Mapping[str, str] | None)

    Key-value tags associated with the data being uploaded.

enqueue_from_dict(timestamp, channel_values, tags=None)#

Write multiple channel values at a given timestamp using a flattened dictionary.

Each key in the dictionary is treated as a channel name and the corresponding value is enqueued with the given timestamp.

Parameters:

  • timestamp (str | datetime | int)

    The shared timestamp to use for all items to enqueue.

  • channel_values (Mapping[str, StreamType])

    A dictionary mapping channel names to their respective values.

  • tags (Mapping[str, str] | None)

    Key-value tags associated with the data being uploaded.

    Note: This should include all required_tags configured for the dataset.

enqueue_string_array(channel_name, timestamp, value, tags=None)#

Add an array of strings to the queue after normalizing the timestamp.

Parameters:

  • channel_name (str)

    Name of the channel to upload data for.

  • timestamp (str | datetime | int)

    Absolute timestamp of the data being uploaded.

  • value (list[str])

    List of string values to write to the specified channel.

  • tags (Mapping[str, str] | None)

    Key-value tags associated with the data being uploaded.

enqueue_struct(channel_name, timestamp, value, tags=None)#

Add a struct (dict) to the queue after normalizing the timestamp.

Parameters:

  • channel_name (str)

    Name of the channel to upload data for.

  • timestamp (str | datetime | int)

    Absolute timestamp of the data being uploaded.

  • value (dict[str, Any])

    Dict to write to the specified channel. Must be JSON-serializable.

  • tags (Mapping[str, str] | None)

    Key-value tags associated with the data being uploaded.

flush(wait=False, timeout=None)#

Flush current batch of records to nominal in a background thread.

Parameters:

  • wait (bool)

    If true, wait for the batch to complete uploading before returning

  • timeout (float | None)

    If wait is true, the time to wait for flush completion in seconds.

    Note: If None, waits indefinitely.