Expose HTTP handler engine option in python bindings

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 {

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<PyAny>,
+    task_locals: Option<TaskLocals>,
+}
+
+impl PyHttpHandler {
+    pub fn new(callback: Py<PyAny>, task_locals: Option<TaskLocals>) -> Self {
+        Self {
+            callback,
+            task_locals,
+        }
+    }
+}
+
+fn extract_http_response(py: Python<'_>, result: PyObject) -> anyhow::Result<HttpHandlerResponse> {
+    let dict = result
+        .extract::<Bound<'_, PyDict>>(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<Box<dyn Future<Output = Result<HttpHandlerResponse, String>> + 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())
+                }
+            }
+        })
+    }
+}

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;

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})

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())

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):