WriteStream

class nominal.core.WriteStream(batch_size: int, max_wait: timedelta, _process_batch: Callable[[Sequence[BatchItem[StreamType]]], None], _executor: ThreadPoolExecutor, _thread_safe_batch: ThreadSafeBatch[StreamType], _stop: Event, _pending_jobs: BoundedSemaphore)

Bases: WriteStreamBase[StreamType]

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

batch_size: int

max_wait: timedelta

close(wait: bool = True) → None

Close the Nominal Stream.

Stop the process timeout thread and flush any remaining batches.

classmethod create(batch_size: int, max_wait: timedelta, process_batch: Callable[[Sequence[BatchItem[StreamType]]], None]) → Self

Create the stream.

Parameters:

• batch_size – Maximum number of items to batch before flushing.

• max_wait – Maximum time to wait before flushing a batch.

• process_batch – Callable to process batches of items.

enqueue(channel_name: str, timestamp: str | datetime | int, value: StreamType, tags: Mapping[str, str] | None = None) → 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: str, timestamps: Sequence[str | datetime | int], values: Sequence[StreamType], tags: Mapping[str, str] | None = None) → 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 – Name of the channel to upload data for.

• timestamps – Absolute timestamps of the data being uploaded.

• values – Values to write to the specified channel.

• tags – 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: str, timestamp: str | datetime | int, value: list[float], tags: Mapping[str, str] | None = None) → None

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

Parameters:

• channel_name – Name of the channel to upload data for.

• timestamp – Absolute timestamp of the data being uploaded.

• value – List of float values to write to the specified channel.

• tags – Key-value tags associated with the data being uploaded.

enqueue_from_dict(timestamp: str | datetime | int, channel_values: Mapping[str, StreamType], tags: Mapping[str, str] | None = None) → 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 – The shared timestamp to use for all items to enqueue.

• channel_values – A dictionary mapping channel names to their respective values.

• tags – 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: str, timestamp: str | datetime | int, value: list[str], tags: Mapping[str, str] | None = None) → None

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

Parameters:

• channel_name – Name of the channel to upload data for.

• timestamp – Absolute timestamp of the data being uploaded.

• value – List of string values to write to the specified channel.

• tags – Key-value tags associated with the data being uploaded.

enqueue_struct(channel_name: str, timestamp: str | datetime | int, value: dict[str, Any], tags: Mapping[str, str] | None = None) → None

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

Parameters:

• channel_name – Name of the channel to upload data for.

• timestamp – Absolute timestamp of the data being uploaded.

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

• tags – Key-value tags associated with the data being uploaded.

flush(wait: bool = False, timeout: float | None = None) → None

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

Parameters:

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

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

  Note: If None, waits indefinitely.