Function stream_from_async_iter

crates/python/src/py_callable.rs:167–183

(
    async_iter: Py<PyAny>,
)

}

fn stream_from_async_iter(
    async_iter: Py<PyAny>,
) -> FlowResult<Pin<Box<dyn Stream<Item = FlowResult<Json>> + Send>>> {
    let (tx, rx) = tokio::sync::mpsc::channel::<FlowResult<Json>>(32);
    let task_locals = Python::attach(|py| {
        pyo3_async_runtimes::tokio::get_current_locals(py)
            .map_err(|e: pyo3::PyErr| FlowError::Internal(e.to_string()))
    })?;

    let async_iter = Arc::new(async_iter);
    tokio::spawn(pyo3_async_runtimes::tokio::scope(task_locals, async move {
        forward_async_iter(async_iter, tx).await;
    }));

    let stream = tokio_stream::wrappers::ReceiverStream::new(rx);
    Ok(Box::pin(stream) as Pin<Box<dyn Stream<Item = FlowResult<Json>> + Send>>)
}

/// Wrap a Python callable `(str, Json) -> Json` for tool sanitize/intercept fns.
pub fn wrap_py_tool_fn(py_fn: Py<PyAny>) -> ToolSanitizeFn {