rust(feat): Add get_flows to SiftStream (#382)

Author: nathan-sift

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

@@ -309,6 +309,24 @@ impl SiftStream<IngestionConfigMode> {
         Ok(())
     }
 
+    /// Get a copy of the current flow configs known to SiftStream as a HashMap keyed to the flow name.
+    /// This includes flows provided at initialization, and any existing configs
+    /// previously registered in Sift
+    pub fn get_flows(&self) -> HashMap<String, FlowConfig> {
+        // Currently we get the first FlowConfig provided in the Vec to match how send() validates flows
+        self.mode
+            .flows_by_name
+            .iter()
+            .filter_map(|(k, v)| {
+                if v.is_empty() {
+                    None
+                } else {
+                    Some((k.clone(), v[0].clone()))
+                }
+            })
+            .collect()
+    }
+
     /// Attach a run to the stream. Any data provided through [SiftStream::send] after return
     /// of this function will be associated with the run.
     pub async fn attach_run(&mut self, run_selector: RunSelector) -> Result<()> {

rust/crates/sift_stream_bindings/Cargo.toml

@@ -21,9 +21,9 @@ path = "src/bin/stub_gen.rs"
 [dependencies]
 pyo3 = { version = "0.25.0" }
 pyo3-stub-gen = { version = "0.10.0" }
-sift_rs = { version = "0.7.0-rc.2" }
-sift_stream = { version = "0.7.0-rc.2", features = ["metrics-unstable"] }
-sift_error = { version = "0.7.0-rc.2" }
+sift_rs = { workspace = true }
+sift_stream = { workspace = true, features = ["metrics-unstable"] }
+sift_error = { workspace = true }
 tokio = { version = "1.43.0", features = ["rt", "sync", "time", "macros"] }
 pyo3-async-runtimes = { version = "0.25.0", features = ["tokio-runtime"] }
 pbjson-types = { workspace = true }

rust/crates/sift_stream_bindings/sift_stream_bindings.pyi

@@ -273,6 +273,7 @@ class SiftStreamPy:
     def send_requests(self, requests:typing.Sequence[IngestWithConfigDataStreamRequestPy]) -> typing.Any: ...
     def get_metrics_snapshot(self) -> SiftStreamMetricsSnapshotPy: ...
     def add_new_flows(self, flow_configs:typing.Sequence[FlowConfigPy]) -> typing.Any: ...
+    def get_flows(self) -> builtins.dict[builtins.str, FlowConfigPy]: ...
     def attach_run(self, run_selector:RunSelectorPy) -> typing.Any: ...
     def detach_run(self) -> None: ...
     def run(self) -> typing.Optional[builtins.str]: ...

rust/crates/sift_stream_bindings/src/stream/channel.rs

@@ -83,12 +83,33 @@ impl From<ChannelBitFieldElementPy> for ChannelBitFieldElement {
     }
 }
 
+impl From<ChannelBitFieldElement> for ChannelBitFieldElementPy {
+    fn from(value: ChannelBitFieldElement) -> Self {
+        ChannelBitFieldElementPy {
+            inner: value.clone(),
+            name: value.name,
+            index: value.index,
+            bit_count: value.bit_count,
+        }
+    }
+}
+
 impl From<ChannelEnumTypePy> for ChannelEnumType {
     fn from(value: ChannelEnumTypePy) -> Self {
         value.inner
     }
 }
 
+impl From<ChannelEnumType> for ChannelEnumTypePy {
+    fn from(value: ChannelEnumType) -> Self {
+        ChannelEnumTypePy {
+            inner: value.clone(),
+            name: value.name,
+            key: value.key,
+        }
+    }
+}
+
 impl From<ChannelDataType> for ChannelDataTypePy {
     fn from(data_type: ChannelDataType) -> Self {
         match data_type {

rust/crates/sift_stream_bindings/src/stream/config.rs

@@ -127,6 +127,35 @@ impl From<FlowConfigPy> for sift_rs::ingestion_configs::v2::FlowConfig {
     }
 }
 
+impl From<sift_rs::ingestion_configs::v2::FlowConfig> for FlowConfigPy {
+    fn from(config: sift_rs::ingestion_configs::v2::FlowConfig) -> Self {
+        FlowConfigPy {
+            inner: config.clone(),
+            name: config.name,
+            channels: config.channels.into_iter().map(|c| c.into()).collect(),
+        }
+    }
+}
+
+impl From<ChannelConfig> for ChannelConfigPy {
+    fn from(config: ChannelConfig) -> Self {
+        let data_type_py = config.data_type().into();
+        ChannelConfigPy {
+            inner: config.clone(),
+            name: config.name,
+            unit: config.unit,
+            description: config.description,
+            data_type: data_type_py,
+            enum_types: config.enum_types.into_iter().map(|ce| ce.into()).collect(),
+            bit_field_elements: config
+                .bit_field_elements
+                .into_iter()
+                .map(|bfe| bfe.into())
+                .collect(),
+        }
+    }
+}
+
 // PyO3 Method Implementations
 #[gen_stub_pymethods]
 #[pymethods]

rust/crates/sift_stream_bindings/src/stream/mod.rs

@@ -14,6 +14,7 @@ use pyo3::prelude::*;
 use pyo3_async_runtimes::tokio::future_into_py;
 use pyo3_stub_gen::derive::*;
 use sift_stream::{Flow, FlowConfig, IngestionConfigMode, SiftStream};
+use std::collections::HashMap;
 use std::sync::Arc;
 use tokio::sync::Mutex;
 
@@ -135,6 +136,20 @@ impl SiftStreamPy {
         Ok(awaitable.into())
     }
 
+    pub fn get_flows(&self) -> PyResult<HashMap<String, FlowConfigPy>> {
+        let inner_guard = self.inner.blocking_lock();
+        let sift_stream = inner_guard.as_ref().ok_or_else(|| {
+            PyErr::new::<pyo3::exceptions::PyRuntimeError, _>(
+                "Stream has been consumed by finish()",
+            )
+        })?;
+        Ok(sift_stream
+            .get_flows()
+            .into_iter()
+            .map(|(k, v)| (k, v.into()))
+            .collect())
+    }
+
     pub fn attach_run(&self, py: Python, run_selector: RunSelectorPy) -> PyResult<Py<PyAny>> {
         let inner = self.inner.clone();