python(feat): Update sift-stream bindings for FlowBuilder support

Author: tsift

python/lib/sift_client/_internal/low_level_wrappers/ingestion.py

@@ -38,9 +38,11 @@
     from sift_stream_bindings import (
         DurationPy,
         FlowConfigPy,
+        FlowDescriptorPy,
         FlowPy,
         IngestionConfigFormPy,
         IngestWithConfigDataStreamRequestPy,
+        IngestWithConfigDataStreamRequestWrapperPy,
         MetadataPy,
         RecoveryStrategyPy,
         RunFormPy,
@@ -129,12 +131,10 @@ class IngestionConfigStreamingLowLevelClient(LowLevelClientBase):
     DEFAULT_MAX_LOG_FILES = 7  # Equal to 1 week of logs
     DEFAULT_LOGFILE_PREFIX = "sift_stream_bindings.log"
     _sift_stream_instance: SiftStreamPy
-    _known_flows: dict[str, FlowConfig]
 
-    def __init__(self, sift_stream_instance: SiftStreamPy, known_flows: dict[str, FlowConfig]):
+    def __init__(self, sift_stream_instance: SiftStreamPy):
         super().__init__()
         self._sift_stream_instance = sift_stream_instance
-        self._known_flows = known_flows
 
     @classmethod
     async def create_sift_stream_instance(
@@ -194,12 +194,7 @@ async def create_sift_stream_instance(
 
         sift_stream_instance = await builder.build()
 
-        known_flows = {
-            flow_name: FlowConfig._from_rust_config(flow)
-            for flow_name, flow in sift_stream_instance.get_flows().items()
-        }
-
-        return cls(sift_stream_instance, known_flows)
+        return cls(sift_stream_instance)
 
     async def send(self, flow: FlowPy):
         await self._sift_stream_instance.send(flow)
@@ -210,12 +205,16 @@ async def batch_send(self, flows: Iterable[FlowPy]):
     async def send_requests(self, requests: list[IngestWithConfigDataStreamRequestPy]):
         await self._sift_stream_instance.send_requests(requests)
 
+    def send_requests_nonblocking(
+        self, requests: Iterable[IngestWithConfigDataStreamRequestWrapperPy]
+    ):
+        self._sift_stream_instance.send_requests_nonblocking(requests)
+
+    def get_flow_descriptor(self, flow_name: str) -> FlowDescriptorPy:
+        return self._sift_stream_instance.get_flow_descriptor(flow_name)
+
     async def add_new_flows(self, flow_configs: list[FlowConfigPy]):
         await self._sift_stream_instance.add_new_flows(flow_configs)
-        self._known_flows = {
-            flow_name: FlowConfig._from_rust_config(flow)
-            for flow_name, flow in self._sift_stream_instance.get_flows().items()
-        }
 
     async def attach_run(self, run_selector: RunSelectorPy):
         await self._sift_stream_instance.attach_run(run_selector)

python/lib/sift_client/resources/ingestion.py

@@ -18,9 +18,11 @@
     from sift_stream_bindings import (
         DiskBackupPolicyPy,
         DurationPy,
+        FlowDescriptorPy,
         FlowPy,
         IngestionConfigFormPy,
         IngestWithConfigDataStreamRequestPy,
+        IngestWithConfigDataStreamRequestWrapperPy,
         MetadataPy,
         RecoveryStrategyPy,
         RetryPolicyPy,
@@ -491,6 +493,31 @@ async def send_requests(self, requests: list[IngestWithConfigDataStreamRequestPy
         """
         await self._low_level_client.send_requests(requests)
 
+    def send_requests_nonblocking(
+        self, requests: Iterable[IngestWithConfigDataStreamRequestWrapperPy]
+    ):
+        """Send data in a manner identical to the raw gRPC service for ingestion-config based streaming.
+
+        This method offers a way to send data that matches the raw gRPC service interface. You are
+        expected to handle channel value ordering as well as empty values correctly.
+
+        Important:
+            If using this interface, you should use `FlowBuilderPy::request` to ensure proper
+            building of the request.
+
+        Args:
+            requests: List of ingestion requests to send to Sift.
+        """
+        self._low_level_client.send_requests_nonblocking(requests)
+
+    def get_flow_descriptor(self, flow_name: str) -> FlowDescriptorPy:
+        """Retrieve a flow descriptor by name.
+
+        Args:
+            flow_name: The name of the flow descriptor to retrieve.
+        """
+        return self._low_level_client.get_flow_descriptor(flow_name)
+
     async def add_new_flows(self, flow_configs: list[FlowConfig]):
         """Modify the existing ingestion config by adding new flows that weren't accounted for during initialization.
 
@@ -575,23 +602,6 @@ def get_metrics_snapshot(self) -> SiftStreamMetricsSnapshotPy:
         """
         return self._low_level_client.get_metrics_snapshot()
 
-    def get_flow_config(self, flow_name: str) -> FlowConfig:
-        """Retrieve a flow configuration by name.
-
-        Args:
-            flow_name: The name of the flow configuration to retrieve.
-
-        Returns:
-            The FlowConfig associated with the given flow name.
-
-        Raises:
-            KeyError: If the flow name is not found in the known flows.
-        """
-        flow_config = self._low_level_client._known_flows.get(flow_name)
-        if flow_config is None:
-            raise KeyError(f"FlowConfig {flow_name} is unknown to the ingestion client")
-        return flow_config
-
     async def __aenter__(self):
         return self
 

rust/crates/sift_stream/benches/message_to_ingest_req.rs

@@ -30,7 +30,7 @@ use std::hint::black_box;
 use sift_rs::ingestion_configs::v2::{ChannelConfig, FlowConfig};
 use sift_stream::stream::mode::ingestion_config::Flow;
 use sift_stream::{
-    ChannelDataType, ChannelValue, TimeValue, Value,
+    ChannelDataType, ChannelValue, FlowDescriptor, TimeValue, Value,
     stream::mode::bench::{message_to_ingest_req, message_to_ingest_req_direct},
 };
 
@@ -118,11 +118,9 @@ fn flow_randomized(name: &str, flow_config: &FlowConfig) -> Flow {
 }
 
 // Configuration constants - these can be adjusted to test different scenarios
-const NUM_FLOWS: usize = 10; // Number of flow configs to create
 const NUM_CHANNELS_PER_FLOW: usize = 2000; // Number of channels per flow
 const INGESTION_CONFIG_ID: &str = "benchmark-config";
 const RUN_ID: Option<String> = None;
-const FLOW_TO_RANDOMIZE: usize = 8;
 
 fn benchmark_message_to_ingest_req_direct(c: &mut Criterion) {
     // Create a flow with ordered channel values (matching the first flow config)
@@ -140,61 +138,42 @@ fn benchmark_message_to_ingest_req_direct(c: &mut Criterion) {
 }
 
 fn benchmark_message_to_ingest_req_ordered(c: &mut Criterion) {
-    // Create flow configs
-    let mut flow_configs = Vec::with_capacity(NUM_FLOWS);
-    for i in 0..NUM_FLOWS {
-        flow_configs.push(flow_config(&format!("flow_{i}"), NUM_CHANNELS_PER_FLOW));
-    }
+    // Create a flow with ordered channel values.
+    let flow = flow_config("my_benchmark_flow", NUM_CHANNELS_PER_FLOW);
+    let message = flow_ordered("my_benchmark_flow", &flow);
 
-    // Create a flow with ordered channel values (matching the first flow config)
-    let message = flow_ordered("flow_0", &flow_configs[FLOW_TO_RANDOMIZE]);
+    let descriptor = FlowDescriptor::try_from((INGESTION_CONFIG_ID, flow)).unwrap();
 
     c.bench_function("message_to_ingest_req_ordered", |b| {
-        b.iter(|| {
-            black_box(message_to_ingest_req(
-                &message,
-                INGESTION_CONFIG_ID,
-                RUN_ID.clone(),
-                &flow_configs,
-            ))
-        })
+        b.iter(|| black_box(message_to_ingest_req(&message, RUN_ID.clone(), &descriptor)))
     });
 }
 
 fn benchmark_message_to_ingest_req_randomized(c: &mut Criterion) {
-    // Create flow configs
-    let mut flow_configs = Vec::with_capacity(NUM_FLOWS);
-    for i in 0..NUM_FLOWS {
-        flow_configs.push(flow_config(&format!("flow_{i}"), NUM_CHANNELS_PER_FLOW));
-    }
+    // Create a flow with randomized channel values.
+    let flow = flow_config("my_benchmark_flow", NUM_CHANNELS_PER_FLOW);
+    let message = flow_randomized("my_benchmark_flow", &flow);
 
-    // Create a flow with randomized channel values (matching the first flow config)
-    let message = flow_randomized("flow_0", &flow_configs[FLOW_TO_RANDOMIZE]);
+    let descriptor = FlowDescriptor::try_from((INGESTION_CONFIG_ID, flow)).unwrap();
 
     c.bench_function("message_to_ingest_req_randomized", |b| {
-        b.iter(|| {
-            black_box(message_to_ingest_req(
-                &message,
-                INGESTION_CONFIG_ID,
-                RUN_ID.clone(),
-                &flow_configs,
-            ))
-        })
+        b.iter(|| black_box(message_to_ingest_req(&message, RUN_ID.clone(), &descriptor)))
     });
 }
 
 fn benchmark_message_to_ingest_req_varying_sizes(c: &mut Criterion) {
     let mut group = c.benchmark_group("message_to_ingest_req_varying_sizes");
 
     for &num_channels in &[5, 10, 100, 1000, 5000] {
+        let flow_name = format!("flow_{num_channels}");
+
         // Create flow configs with varying channel counts
-        let mut flow_configs = Vec::with_capacity(NUM_FLOWS);
-        for i in 0..NUM_FLOWS {
-            flow_configs.push(flow_config(&format!("flow_{i}"), num_channels));
-        }
+        let flow = flow_config(&flow_name, num_channels);
+        let message_ordered = flow_ordered(&flow_name, &flow);
+        let message_randomized = flow_randomized(&flow_name, &flow);
+        let descriptor = FlowDescriptor::try_from((INGESTION_CONFIG_ID, flow)).unwrap();
 
         // Test direct scenario
-        let message_ordered = flow_ordered("flow_0", &flow_configs[FLOW_TO_RANDOMIZE]);
         group.bench_function(&format!("direct_{num_channels}_channels"), |b| {
             b.iter(|| {
                 black_box(message_to_ingest_req_direct(
@@ -209,22 +188,19 @@ fn benchmark_message_to_ingest_req_varying_sizes(c: &mut Criterion) {
             b.iter(|| {
                 black_box(message_to_ingest_req(
                     &message_ordered,
-                    INGESTION_CONFIG_ID,
                     RUN_ID.clone(),
-                    &flow_configs,
+                    &descriptor,
                 ))
             })
         });
 
         // Test randomized scenario
-        let message_randomized = flow_randomized("flow_0", &flow_configs[FLOW_TO_RANDOMIZE]);
         group.bench_function(&format!("randomized_{num_channels}_channels"), |b| {
             b.iter(|| {
                 black_box(message_to_ingest_req(
                     &message_randomized,
-                    INGESTION_CONFIG_ID,
                     RUN_ID.clone(),
-                    &flow_configs,
+                    &descriptor,
                 ))
             })
         });

rust/crates/sift_stream/src/stream/mode/bench.rs

@@ -1,21 +1,15 @@
+use crate::stream::flow::FlowDescriptor;
 use crate::stream::mode::ingestion_config::Flow;
 use crate::{IngestionConfigMode, SiftStream};
-use sift_rs::ingestion_configs::v2::FlowConfig;
 
 /// Unstable wrapper around [SiftStream::message_to_ingest_req] used for benchmarking purposes.
 #[inline]
 pub fn message_to_ingest_req(
     message: &Flow,
-    ingestion_config_id: &str,
     run_id: Option<String>,
-    flows: &[FlowConfig],
+    descriptor: &FlowDescriptor<String>,
 ) -> Option<sift_rs::ingest::v1::IngestWithConfigDataStreamRequest> {
-    SiftStream::<IngestionConfigMode>::message_to_ingest_req(
-        message,
-        ingestion_config_id,
-        run_id,
-        flows,
-    )
+    SiftStream::<IngestionConfigMode>::message_to_ingest_req(message, run_id, descriptor)
 }
 
 /// Unstable wrapper around [SiftStream::message_to_ingest_req_direct] used for benchmarking purposes.