rust(feat): Add batch_send to sift_stream_bindings (#384)

Author: nathan-sift

rust/crates/sift_stream_bindings/sift_stream_bindings.pyi

@@ -270,6 +270,7 @@ class SiftStreamMetricsSnapshotPy:
 @typing.final
 class SiftStreamPy:
     def send(self, flow:FlowPy) -> typing.Any: ...
+    def batch_send(self, flows:typing.Any) -> typing.Any: ...
     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: ...

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

@@ -10,7 +10,7 @@ use crate::metrics::SiftStreamMetricsSnapshotPy;
 use crate::stream::channel::ChannelValuePy;
 use crate::stream::config::{FlowConfigPy, RunSelectorPy};
 use crate::stream::time::TimeValuePy;
-use pyo3::prelude::*;
+use pyo3::{prelude::*, types::PyIterator};
 use pyo3_async_runtimes::tokio::future_into_py;
 use pyo3_stub_gen::derive::*;
 use sift_stream::{Flow, FlowConfig, IngestionConfigMode, SiftStream};
@@ -71,6 +71,41 @@ impl SiftStreamPy {
         Ok(awaitable.into())
     }
 
+    // Function to take in a python iterable of PyFlow and call send on each item
+    // Can allow more performant sending of data from python to SiftStream
+    pub fn batch_send<'py>(
+        &self,
+        py: Python<'py>,
+        flows: &Bound<'_, PyAny>,
+    ) -> PyResult<Py<PyAny>> {
+        let flow_iter = PyIterator::from_object(flows)?;
+        let mut flows_vec = Vec::new();
+        for item in flow_iter {
+            flows_vec.push(item?.extract::<FlowPy>()?);
+        }
+
+        let inner = self.inner.clone();
+
+        let awaitable = future_into_py(py, async move {
+            let mut guard = inner.lock().await;
+            let stream = guard.as_mut().ok_or_else(|| {
+                PyErr::new::<pyo3::exceptions::PyRuntimeError, _>(
+                    "Stream has been consumed by finish()",
+                )
+            })?;
+
+            for flow in flows_vec {
+                match stream.send(flow.into()).await {
+                    Ok(_) => (),
+                    Err(e) => return Err(SiftErrorWrapper(e).into()),
+                }
+            }
+            Ok(())
+        })?;
+
+        Ok(awaitable.into())
+    }
+
     pub fn send_requests(
         &self,
         py: Python,