nominal-streaming Python Bindings

nominal-streaming is a thin python wrapper around the existing nominal-streaming rust crate. Usage semantics remain largely the same, but with some slight alterations to allow for a more pythonic interface.

Requires Python 3.11 or newer.

The library aims to balance three concerns:

1. Data should exist in-memory only for a limited, configurable amount of time before it's sent to Core.
2. Writes should fall back to disk if there are network failures.
3. Backpressure should be applied to incoming requests when network throughput is saturated.

This library streams data to Nominal Core, to a file, or to Nominal Core with a file as backup (recommended to protect against network failures). It also provides configuration to manage the tradeoff between above listed concerns.

Usage example: streaming from memory to Nominal Core with file fallback

import pathlib
import time

from nominal.core import NominalClient
from nominal_streaming import NominalDatasetStream, PyNominalStreamOpts

if __name__ == "__main__":
    num_points = 100_000
    stream = (
        NominalDatasetStream(
            auth_header="<api key>",
            opts=PyNominalStreamOpts(),
        )
        .enable_logging("info") # can set debug, warn, etc.
        .with_core_consumer("<dataset rid>")
        .with_file_fallback(pathlib.Path("local_fallback.avro"))
    )

    with stream:
        # Stream 100_000 live readings (made up values)
        for idx in range(num_points):
            time_ns = int(time.time() * 1e9)
            value = (idx % 50) + 0.5
            stream.enqueue("channel_name", time_ns, value, tags={"tag_key": "tag_value"})

        # Stream 100_000 points in one batch
        start_time = int(time.time() * 1e9)
        timestamp_offsets = int(1e9 / 1600)
        timestamps = [start_time + timestamp_offsets * idx for idx in range(num_points)]
        values = [(idx % 50) + 0.5 for idx in range(num_points)]
        stream.enqueue_batch(
            "channel_name",
            timestamps,
            values,
            tags={"tag_key": "tag_value"}
        )

Runtime metrics

Enable PyNominalStreamOpts(track_metrics=True) (or pass track_metrics=True to NominalDatasetStream.create) to emit dictionary enqueue staleness and Core request latency metrics. Metrics are disabled by default.

NumPy batches

NumPy is optional: install nominal-streaming[numpy], or use the NumPy installation already provided by your application. Pass arrays directly to enqueue_batch:

import numpy as np

timestamps = np.array([1_700_000_000_000_000_000, 1_700_000_000_000_000_001], dtype="uint64")
values = np.array([1.25, 2.5], dtype="float64")
stream.enqueue_batch("temperature", timestamps, values)

Native-endian, aligned, one-dimensional NumPy arrays are copied into Rust-owned memory without creating a Python object for every element. The fast path supports int64/uint64 timestamps and float32/float64/int64/uint64 values, including strided, reversed, and read-only views. Integer values retain the API's existing conversion to doubles. Inputs can be changed or freed after the call returns.

Other dtypes, unaligned or non-native-endian arrays, and ndarray subclasses use the existing element-wise conversion. Masked values and datetime/timedelta values still require explicit conversion. Lists and tuples continue to work without NumPy.

Building and testing wheels

All builds use the Python 3.11 stable ABI (cp311-abi3) with native NumPy buffer support. Build a wheel with the standard workspace command:

uv build --package nominal-streaming

NumPy is never needed to import the library or use lists. CI shares one wheel across parallel Python 3.11 through 3.15 jobs. Each job exercises lists without NumPy, then runs the batch regression tests and type checks with the locked test dependencies. To run tests locally with the test dependency group:

uv run --group test python -m unittest discover -s py-nominal-streaming/tests -v

Measuring enqueue performance

With a wheel and the test dependencies installed:

python py-nominal-streaming/benchmarks/enqueue_batch.py

This compares direct arrays, pre-existing lists, and .tolist() plus the enqueue call. It reports all samples and medians and reads the local Avro output to verify point counts and value checksums. It measures the complete public enqueue call with buffer capacity available; startup and draining are excluded. It does not measure network throughput. Downstream serialization, compression, and upload backpressure can limit the overall streaming improvement.