From ecb1431617c852ca9b945bb4e842db465b214a75 Mon Sep 17 00:00:00 2001 From: Chris Sperandio Date: Mon, 31 Aug 2026 16:16:26 -0700 Subject: [PATCH] feat(python): expose httpHandler engine option Adds the httpHandler option to the Python bindings, mirroring the Node.js bindings. When provided, all http module calls from function nodes are routed to the Python callable (sync or async) instead of the native reqwest client, enabling request interception, domain allowlisting, custom auth, and mocking in tests. --- bindings/python/src/engine.rs | 11 ++++- bindings/python/src/http_handler.rs | 71 +++++++++++++++++++++++++++++ bindings/python/src/lib.rs | 1 + bindings/python/test_async.py | 40 ++++++++++++++++ bindings/python/test_sync.py | 68 +++++++++++++++++++++++++++ bindings/python/zen.pyi | 19 ++++++++ 6 files changed, 209 insertions(+), 1 deletion(-) create mode 100644 bindings/python/src/http_handler.rs 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):