# 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*](https://docs.python.org/3/builtins/functions.html#int "(in Python v3.15)"))

      Maximum number of items to batch before flushing.
    - **max_wait** ([*timedelta*](https://docs.python.org/3/library/datetime.html#datetime.timedelta "(in Python v3.15)"))

      Maximum time to wait before flushing a batch.
    - **process_batch** ([*Callable*](https://docs.python.org/3/library/typing.html#typing.Callable "(in Python v3.15)")*[**[*[*Sequence*](https://docs.python.org/3/library/typing.html#typing.Sequence "(in Python v3.15)")*[**BatchItem**[**StreamType**]**]**]**,* *None**]*)

      Callable to process batches of items.

Return type:
:   [*Self*](https://docs.python.org/3/library/typing.html#typing.Self "(in Python v3.15)")

### 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*](https://docs.python.org/3/builtins/stdtypes.html#str "(in Python v3.15)"))

      Name of the channel to upload data for.
    - **timestamps** ([*Sequence*](https://docs.python.org/3/library/typing.html#typing.Sequence "(in Python v3.15)")*[*[*str*](https://docs.python.org/3/builtins/stdtypes.html#str "(in Python v3.15)") *|* [*datetime*](https://docs.python.org/3/library/datetime.html#datetime.datetime "(in Python v3.15)") *|* [*int*](https://docs.python.org/3/builtins/functions.html#int "(in Python v3.15)")*]*)

      Absolute timestamps of the data being uploaded.
    - **values** ([*Sequence*](https://docs.python.org/3/library/typing.html#typing.Sequence "(in Python v3.15)")*[**StreamType**]*)

      Values to write to the specified channel.
    - **tags** ([*Mapping*](https://docs.python.org/3/library/typing.html#typing.Mapping "(in Python v3.15)")*[*[*str*](https://docs.python.org/3/builtins/stdtypes.html#str "(in Python v3.15)")*,* [*str*](https://docs.python.org/3/builtins/stdtypes.html#str "(in Python v3.15)")*]* *|* *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*](https://docs.python.org/3/builtins/stdtypes.html#str "(in Python v3.15)"))

      Name of the channel to upload data for.
    - **timestamp** ([*str*](https://docs.python.org/3/builtins/stdtypes.html#str "(in Python v3.15)") *|* [*datetime*](https://docs.python.org/3/library/datetime.html#datetime.datetime "(in Python v3.15)") *|* [*int*](https://docs.python.org/3/builtins/functions.html#int "(in Python v3.15)"))

      Absolute timestamp of the data being uploaded.
    - **value** ([*list*](https://docs.python.org/3/builtins/stdtypes.html#list "(in Python v3.15)")*[*[*float*](https://docs.python.org/3/builtins/functions.html#float "(in Python v3.15)")*]*)

      List of float values to write to the specified channel.
    - **tags** ([*Mapping*](https://docs.python.org/3/library/typing.html#typing.Mapping "(in Python v3.15)")*[*[*str*](https://docs.python.org/3/builtins/stdtypes.html#str "(in Python v3.15)")*,* [*str*](https://docs.python.org/3/builtins/stdtypes.html#str "(in Python v3.15)")*]* *|* *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*](https://docs.python.org/3/builtins/stdtypes.html#str "(in Python v3.15)") *|* [*datetime*](https://docs.python.org/3/library/datetime.html#datetime.datetime "(in Python v3.15)") *|* [*int*](https://docs.python.org/3/builtins/functions.html#int "(in Python v3.15)"))

      The shared timestamp to use for all items to enqueue.
    - **channel_values** ([*Mapping*](https://docs.python.org/3/library/typing.html#typing.Mapping "(in Python v3.15)")*[*[*str*](https://docs.python.org/3/builtins/stdtypes.html#str "(in Python v3.15)")*,* *StreamType**]*)

      A dictionary mapping channel names to their respective values.
    - **tags** ([*Mapping*](https://docs.python.org/3/library/typing.html#typing.Mapping "(in Python v3.15)")*[*[*str*](https://docs.python.org/3/builtins/stdtypes.html#str "(in Python v3.15)")*,* [*str*](https://docs.python.org/3/builtins/stdtypes.html#str "(in Python v3.15)")*]* *|* *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*](https://docs.python.org/3/builtins/stdtypes.html#str "(in Python v3.15)"))

      Name of the channel to upload data for.
    - **timestamp** ([*str*](https://docs.python.org/3/builtins/stdtypes.html#str "(in Python v3.15)") *|* [*datetime*](https://docs.python.org/3/library/datetime.html#datetime.datetime "(in Python v3.15)") *|* [*int*](https://docs.python.org/3/builtins/functions.html#int "(in Python v3.15)"))

      Absolute timestamp of the data being uploaded.
    - **value** ([*list*](https://docs.python.org/3/builtins/stdtypes.html#list "(in Python v3.15)")*[*[*str*](https://docs.python.org/3/builtins/stdtypes.html#str "(in Python v3.15)")*]*)

      List of string values to write to the specified channel.
    - **tags** ([*Mapping*](https://docs.python.org/3/library/typing.html#typing.Mapping "(in Python v3.15)")*[*[*str*](https://docs.python.org/3/builtins/stdtypes.html#str "(in Python v3.15)")*,* [*str*](https://docs.python.org/3/builtins/stdtypes.html#str "(in Python v3.15)")*]* *|* *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*](https://docs.python.org/3/builtins/stdtypes.html#str "(in Python v3.15)"))

      Name of the channel to upload data for.
    - **timestamp** ([*str*](https://docs.python.org/3/builtins/stdtypes.html#str "(in Python v3.15)") *|* [*datetime*](https://docs.python.org/3/library/datetime.html#datetime.datetime "(in Python v3.15)") *|* [*int*](https://docs.python.org/3/builtins/functions.html#int "(in Python v3.15)"))

      Absolute timestamp of the data being uploaded.
    - **value** ([*dict*](https://docs.python.org/3/builtins/stdtypes.html#dict "(in Python v3.15)")*[*[*str*](https://docs.python.org/3/builtins/stdtypes.html#str "(in Python v3.15)")*,* [*Any*](https://docs.python.org/3/library/typing.html#typing.Any "(in Python v3.15)")*]*)

      Dict to write to the specified channel. Must be JSON-serializable.
    - **tags** ([*Mapping*](https://docs.python.org/3/library/typing.html#typing.Mapping "(in Python v3.15)")*[*[*str*](https://docs.python.org/3/builtins/stdtypes.html#str "(in Python v3.15)")*,* [*str*](https://docs.python.org/3/builtins/stdtypes.html#str "(in Python v3.15)")*]* *|* *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*](https://docs.python.org/3/builtins/functions.html#bool "(in Python v3.15)"))

      If true, wait for the batch to complete uploading before returning
    - **timeout** ([*float*](https://docs.python.org/3/builtins/functions.html#float "(in Python v3.15)") *|* *None*)

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

      **Note:** If None, waits indefinitely.