use std::future::Future; use std::pin::Pin; use std::sync::Arc; use crate::content::PyZenDecisionContentJson; use anyhow::anyhow; use either::Either; use pyo3::{IntoPyObjectExt, Py, PyAny, PyResult, Python}; use pyo3_async_runtimes::TaskLocals; use zen_engine::loader::{DecisionLoader, LoaderError, LoaderResponse}; use zen_engine::model::DecisionContent; #[derive(Default, Debug)] pub(crate) struct PyDecisionLoader { callback: Option>, task_locals: Option, } impl PyDecisionLoader { pub fn new(callback: Option>, task_locals: Option) -> Self { Self { callback, task_locals, } } } impl PyDecisionLoader { async fn load_element(&self, key: &str) -> Result, anyhow::Error> { let Some(callable) = &self.callback else { return Err(anyhow!("Loader is not defined")); }; let maybe_result: PyResult<_> = Python::with_gil(|py| { let result = callable.call1(py, (key,))?; let is_coroutine = result.getattr(py, "__await__").is_ok(); if !is_coroutine { return Ok(Either::Left( result.extract::(py)?, )); } let Some(task_locals) = &self.task_locals else { Err(anyhow!("Task locals are required in async context"))? }; let result_future = pyo3_async_runtimes::into_future_with_locals( task_locals, result.into_bound_py_any(py)?, )?; Ok(Either::Right(result_future)) }); match maybe_result? { Either::Left(result) => Ok(result.0 .0), Either::Right(future) => { let result = future.await?; let content = Python::with_gil(|py| result.extract::(py))?; Ok(content.0 .0) } } } } impl DecisionLoader for PyDecisionLoader { fn load<'a>( &'a self, key: &'a str, ) -> Pin + 'a + Send>> { Box::pin(async move { self.load_element(key).await.map_err(|e| { LoaderError::Internal { source: e, key: key.to_string(), } .into() }) }) } }