Create notebook_runner.py

server/notebook_runner.py

@@ -0,0 +1,68 @@
+"""
+notebook_runner.py
+
+Executes a Jupyter notebook (.ipynb) in-process and returns
+cell outputs as a list of dicts.  Used by POST /notebook/run.
+"""
+
+import asyncio
+import json
+import time
+from pathlib import Path
+from typing import AsyncIterator
+
+import nbformat
+from nbconvert.preprocessors import ExecutePreprocessor
+from nbformat.v4 import new_notebook, new_code_cell
+
+
+async def run_notebook_source(
+    cells_source: list[str],
+    timeout: int = 60,
+) -> AsyncIterator[str]:
+    """
+    Given a list of code-cell source strings, execute them and
+    yield Server-Sent Event lines with cell outputs.
+    """
+    nb = new_notebook()
+    nb.cells = [new_code_cell(src) for src in cells_source]
+
+    def _execute():
+        ep = ExecutePreprocessor(
+            timeout=timeout,
+            kernel_name="python3",
+            allow_errors=True,         # send errors as output, don't abort
+        )
+        ep.preprocess(nb)
+        return nb
+
+    loop = asyncio.get_event_loop()
+    executed_nb = await loop.run_in_executor(None, _execute)
+
+    for i, cell in enumerate(executed_nb.cells):
+        for output in cell.get("outputs", []):
+            otype = output.get("output_type", "")
+            if otype in ("stream",):
+                text = "".join(output.get("text", []))
+                yield _sse({"cell": i, "type": "stream", "text": text})
+            elif otype in ("execute_result", "display_data"):
+                data = output.get("data", {})
+                yield _sse({
+                    "cell": i,
+                    "type": otype,
+                    "text": data.get("text/plain", ""),
+                    "html": data.get("text/html", ""),
+                })
+            elif otype == "error":
+                yield _sse({
+                    "cell": i,
+                    "type": "error",
+                    "ename": output.get("ename", ""),
+                    "evalue": output.get("evalue", ""),
+                    "traceback": output.get("traceback", []),
+                })
+    yield _sse({"type": "done"})
+
+
+def _sse(obj: dict) -> str:
+    return f"data: {json.dumps(obj)}\n\n"