diff --git a/bindings/python/src/engine.rs b/bindings/python/src/engine.rs index 081eb2d9..48837c60 100644 --- a/bindings/python/src/engine.rs +++ b/bindings/python/src/engine.rs @@ -3,6 +3,7 @@ use std::sync::Arc; use crate::content::PyZenDecisionContentJson; use crate::custom_node::PyCustomNode; use crate::decision::PyZenDecision; +use crate::http_handler::PyHttpHandler; use crate::loader::PyDecisionLoader; use crate::mt::{block_on, worker_pool}; use crate::value::PyValue; @@ -157,10 +158,18 @@ impl PyZenEngine { None => Arc::new(PyDecisionLoader::default()), }; - let engine = DecisionEngine::new( + let mut engine = DecisionEngine::new( loader, Arc::new(PyCustomNode::new(custom_node, make_locals())), ); + + if let Some(http_handler) = options.get_item("httpHandler")? { + engine = engine.with_http_handler(Some(Arc::new(PyHttpHandler::new( + http_handler.into_py_any(py)?, + make_locals(), + )))); + } + engine.compile(); Ok(Self { diff --git a/bindings/python/src/http_handler.rs b/bindings/python/src/http_handler.rs new file mode 100644 index 00000000..c3f7ed3d --- /dev/null +++ b/bindings/python/src/http_handler.rs @@ -0,0 +1,71 @@ +use anyhow::{anyhow, Context}; +use either::Either; +use pyo3::types::PyDict; +use pyo3::{Bound, IntoPyObjectExt, Py, PyAny, PyObject, Python}; +use pyo3_async_runtimes::TaskLocals; +use pythonize::{depythonize, pythonize}; +use std::future::Future; +use std::pin::Pin; +use zen_engine::nodes::http_handler::{HttpHandler, HttpHandlerRequest, HttpHandlerResponse}; + +#[derive(Debug)] +pub(crate) struct PyHttpHandler { + callback: Py, + task_locals: Option, +} + +impl PyHttpHandler { + pub fn new(callback: Py, task_locals: Option) -> Self { + Self { + callback, + task_locals, + } + } +} + +fn extract_http_response(py: Python<'_>, result: PyObject) -> anyhow::Result { + let dict = result + .extract::>(py) + .context("Failed to extract response")?; + let response: HttpHandlerResponse = + depythonize(&dict).context("Failed to depythonize response")?; + Ok(response) +} + +impl HttpHandler for PyHttpHandler { + fn handle( + &self, + request: HttpHandlerRequest, + ) -> Pin> + Send + '_>> { + Box::pin(async move { + let maybe_result: anyhow::Result<_> = Python::with_gil(|py| { + let request_obj = pythonize(py, &request).context("Failed to convert request")?; + let result = self.callback.call1(py, (request_obj,))?; + let is_coroutine = result.getattr(py, "__await__").is_ok(); + if !is_coroutine { + return Ok(Either::Left(extract_http_response(py, result))); + } + + 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.map_err(|err| err.to_string())? { + Either::Left(result) => result.map_err(|err| err.to_string()), + Either::Right(future) => { + let result = future.await.map_err(|err| err.to_string())?; + Python::with_gil(|py| extract_http_response(py, result)) + .map_err(|err| err.to_string()) + } + } + }) + } +} diff --git a/bindings/python/src/lib.rs b/bindings/python/src/lib.rs index 35d17853..bba75678 100644 --- a/bindings/python/src/lib.rs +++ b/bindings/python/src/lib.rs @@ -15,6 +15,7 @@ mod custom_node; mod decision; mod engine; mod expression; +mod http_handler; mod loader; mod mt; mod types; diff --git a/bindings/python/test_async.py b/bindings/python/test_async.py index 0efad8e7..cc2b68b5 100644 --- a/bindings/python/test_async.py +++ b/bindings/python/test_async.py @@ -35,7 +35,47 @@ async def custom_async_handler(request): } +def http_handler_decision_content(): + source = ( + "import http from 'http';\n" + "\n" + "export const handler = async (input) => {\n" + " const response = await http.get('https://example.com/products/1');\n" + " return { status: response.status, product: response.data.product };\n" + "};\n" + ) + return json.dumps({ + "contentType": "application/vnd.gorules.decision", + "nodes": [ + {"type": "inputNode", "id": "input1", "name": "request", "position": {"x": 0, "y": 0}}, + {"type": "functionNode", "id": "function1", "name": "function1", + "content": {"source": source}, "position": {"x": 100, "y": 0}}, + {"type": "outputNode", "id": "output1", "name": "response", "position": {"x": 200, "y": 0}}, + ], + "edges": [ + {"id": "edge1", "type": "edge", "sourceId": "input1", "targetId": "function1"}, + {"id": "edge2", "type": "edge", "sourceId": "function1", "targetId": "output1"}, + ], + }) + + class AsyncZenEngine(unittest.IsolatedAsyncioTestCase): + async def test_async_http_handler(self): + async def http_handler(request): + await asyncio.sleep(0.1) + return { + "status": 200, + "headers": {}, + "data": {"product": "notebook"}, + } + + engine = zen.ZenEngine({"httpHandler": http_handler}) + decision = engine.create_decision(http_handler_decision_content()) + r = await decision.async_evaluate({}) + + self.assertEqual(r["result"]["status"], 200) + self.assertEqual(r["result"]["product"], "notebook") + async def test_async_evaluate(self): engine = zen.ZenEngine({"loader": loader}) r1 = engine.async_evaluate("function.json", {"input": 5}) diff --git a/bindings/python/test_sync.py b/bindings/python/test_sync.py index 4d0acc0c..b351561a 100644 --- a/bindings/python/test_sync.py +++ b/bindings/python/test_sync.py @@ -22,6 +22,37 @@ def custom_handler(request): "output": {"sum": p1} } +def http_handler_decision_content(): + source = ( + "import http from 'http';\n" + "\n" + "export const handler = async (input) => {\n" + " const response = await http.get('https://example.com/products/1', {\n" + " headers: { 'x-request': 'ping' },\n" + " params: { page: '1' },\n" + " });\n" + "\n" + " return {\n" + " status: response.status,\n" + " product: response.data.product,\n" + " mockHeader: response.headers['x-mock'],\n" + " };\n" + "};\n" + ) + return json.dumps({ + "contentType": "application/vnd.gorules.decision", + "nodes": [ + {"type": "inputNode", "id": "input1", "name": "request", "position": {"x": 0, "y": 0}}, + {"type": "functionNode", "id": "function1", "name": "function1", + "content": {"source": source}, "position": {"x": 100, "y": 0}}, + {"type": "outputNode", "id": "output1", "name": "response", "position": {"x": 200, "y": 0}}, + ], + "edges": [ + {"id": "edge1", "type": "edge", "sourceId": "input1", "targetId": "function1"}, + {"id": "edge2", "type": "edge", "sourceId": "function1", "targetId": "output1"}, + ], + }) + # The test based on unittest module class ZenEngine(unittest.TestCase): @@ -68,6 +99,43 @@ def test_engine_custom_handler(self): self.assertEqual(r2["result"]["sum"], 30) self.assertEqual(r3["result"]["sum"], 40) + def test_engine_http_handler(self): + requests = [] + + def http_handler(request): + requests.append(request) + return { + "status": 200, + "headers": {"x-mock": "true"}, + "data": {"product": "notebook"}, + } + + engine = zen.ZenEngine({"httpHandler": http_handler}) + decision = engine.create_decision(http_handler_decision_content()) + r = decision.evaluate({}) + + self.assertEqual(r["result"]["status"], 200) + self.assertEqual(r["result"]["product"], "notebook") + self.assertEqual(r["result"]["mockHeader"], "true") + + self.assertEqual(len(requests), 1) + self.assertEqual(requests[0]["method"], "GET") + self.assertEqual(requests[0]["url"], "https://example.com/products/1") + self.assertEqual(requests[0]["headers"]["x-request"], "ping") + self.assertEqual(requests[0]["params"]["page"], "1") + + def test_engine_http_handler_error(self): + def http_handler(request): + raise PermissionError("domain not allowed") + + engine = zen.ZenEngine({"httpHandler": http_handler}) + decision = engine.create_decision(http_handler_decision_content()) + + with self.assertRaises(RuntimeError) as ctx: + decision.evaluate({}) + + self.assertIn("domain not allowed", str(ctx.exception)) + def test_static_loader_config(self): with open("../../test-data/table.json", "r") as f: table_content = json.loads(f.read()) diff --git a/bindings/python/zen.pyi b/bindings/python/zen.pyi index 0da522b3..2e29d563 100644 --- a/bindings/python/zen.pyi +++ b/bindings/python/zen.pyi @@ -36,9 +36,28 @@ ZenLoaderConfig: TypeAlias = Union[StaticLoaderConfig, FilesystemLoaderConfig, Z ZenLoaderCallback: TypeAlias = Callable[[str], Union[str, dict, ZenDecisionContent, Awaitable[Union[str, dict, ZenDecisionContent]]]] +class ZenHttpHandlerRequest(TypedDict): + method: str + url: str + body: Any + headers: dict[str, str] + params: dict[str, str] + auth: Any + + +class ZenHttpHandlerResponse(TypedDict): + status: int + headers: Any + data: Any + + +ZenHttpHandlerCallback: TypeAlias = Callable[[ZenHttpHandlerRequest], Union[ZenHttpHandlerResponse, Awaitable[ZenHttpHandlerResponse]]] + + class ZenEngineOptions(TypedDict, total=False): loader: Union[ZenLoaderCallback, ZenLoaderConfig] customHandler: Callable + httpHandler: ZenHttpHandlerCallback class EvaluateBatchRequest(TypedDict):