dataforge/fn_registry
1
0
Fork 0
Code Issues Pull Requests 3 Actions Packages Projects Releases Wiki Activity

chore: auto-commit (43 archivos)

Browse Source 
- .mcp.json
- bash/functions/infra/write_mcp_jupyter_config.md
- bash/functions/infra/write_mcp_jupyter_config.sh
- cpp/CMakeLists.txt
- cpp/apps/chart_demo
- cpp/apps/shaders_lab
- cpp/functions/gfx/gl_framebuffer.cpp
- cpp/functions/gfx/gl_framebuffer.h
- cpp/functions/gfx/gl_framebuffer.md
- cpp/functions/gfx/mesh_gpu.md
- ...

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
This commit is contained in:
Egutierrez
2026-05-30 17:28:47 +02:00
parent a2efdcf003
commit fd5787c55f
44 changed files with 3924 additions and 64 deletions
Download Patch File Download Diff File Expand all files Collapse all files
python/functions/notebook/jupyter_run_all.md
+105
View File
@@ -0,0 +1,105 @@
---
name: jupyter_run_all
kind: function
lang: py
domain: notebook
version: "1.0.0"
purity: impure
signature: "jupyter_run_all(notebook_path: str, server_url: str, token: str, restart_kernel: bool, stop_on_error: bool, timeout_per_cell_s: int) -> dict"
description: "Ejecuta todas las celdas de codigo de un notebook Jupyter en orden, con reinicio opcional del kernel antes de empezar. Equivalente al boton 'Run All' del UI pero invocable desde CLI/MCP/agente. Persiste outputs a disco via REST."
tags: [jupyter, notebook, kernel, run-all, smoke-test, ci, execution, notebook]
uses_functions: [jupyter_run_cells_py_notebook]
uses_types: []
returns: []
returns_optional: false
error_type: "error_go_core"
imports: [jupyter_kernel_client, urllib, json, time]
params:
- name: notebook_path
desc: "Ruta relativa al notebook desde la raiz del servidor Jupyter (ej: 'notebooks/analisis.ipynb')"
- name: server_url
desc: "URL base del servidor Jupyter (default http://localhost:8888)"
- name: token
desc: "Token de autenticacion del servidor. Vacio si no se requiere auth"
- name: restart_kernel
desc: "Si True, reinicia el kernel antes de ejecutar para garantizar estado limpio. Default True"
- name: stop_on_error
desc: "Si True, detiene la ejecucion cuando una celda produce error. Default True"
- name: timeout_per_cell_s
desc: "Timeout en segundos por celda. Default 600 (10 minutos)"
output: "Dict con notebook, code_cell_indices, executed (lista de resultados por celda con cell_index/execution_count/outputs/error/duration_s), stopped_at (indice de la celda donde se detuvo si hubo error), kernel_id y total_duration_s"
tested: false
tests: []
test_file_path: ""
file_path: "python/functions/notebook/jupyter_run_all.py"
---
## Ejemplo
```bash
# CLI: ejecutar todas las celdas con kernel limpio
python -m notebook.jupyter_run_all notebooks/analisis.ipynb
# CLI: ejecutar sin reiniciar kernel y continuar si hay errores
python -m notebook.jupyter_run_all notebooks/analisis.ipynb --no-restart --continue-on-error
# CLI: salida JSON completa
python -m notebook.jupyter_run_all notebooks/analisis.ipynb --server http://localhost:8888
```
```python
# Importar y usar desde Python
import sys
sys.path.insert(0, "python/functions")
from notebook.jupyter_run_all import jupyter_run_all
result = jupyter_run_all(
notebook_path="notebooks/analisis.ipynb",
server_url="http://localhost:8888",
token="",
restart_kernel=True,
stop_on_error=True,
)
print(f"Ejecutadas: {len(result['executed'])} celdas en {result['total_duration_s']}s")
if result["stopped_at"] is not None:
print(f"ERROR en celda {result['stopped_at']}")
failed = next(e for e in result["executed"] if e["cell_index"] == result["stopped_at"])
print(failed["error"])
```
Salida ejemplo:
```json
{
"notebook": "notebooks/analisis.ipynb",
"code_cell_indices": [0, 1, 2, 4, 6],
"executed": [
{"cell_index": 0, "execution_count": 1, "outputs": ["pandas 2.2.1"], "error": null, "duration_s": 0.312},
{"cell_index": 1, "execution_count": 2, "outputs": ["(1500, 12)"], "error": null, "duration_s": 0.085}
],
"stopped_at": null,
"kernel_id": "a1b2c3d4-...",
"total_duration_s": 4.217
}
```
## Cuando usarla
Usar `jupyter_run_all` cuando necesitas:
- Smoke test de un notebook despues de cambiar dependencias (confirma que ejecuta de principio a fin).
- Validar un notebook en CI/CD o desde un agente sin abrir el UI de Jupyter Lab.
- Regenerar todos los outputs del notebook con estado limpio (restart_kernel=True).
- Detectar celdas que fallan antes de compartir o publicar el notebook.
No usar para ejecutar una sola celda — usar `jupyter_exec` (modo `cell` o `kernel`) para eso.
## Gotchas
- **Requiere sesion activa**: el kernel del notebook debe estar corriendo. Si el notebook no esta abierto en Jupyter Lab, llamar antes a `jupyter_exec` para crear la sesion, o abrir el notebook manualmente. Error: `RuntimeError: No hay sesion activa`.
- **restart_kernel=True limpia TODO el estado**: variables, imports, estado de modulos. Si el notebook depende de estado previo (interactivo), usar `restart_kernel=False`.
- **stop_on_error=True es el default**: una celda con error detiene el resto. Para runs de diagnostico donde quieres ver todos los errores, pasar `stop_on_error=False`.
- **Timeout por celda**: `timeout_per_cell_s=600` (10 min) es el maximo por celda individual. Celdas con operaciones largas (entrenamiento ML, queries pesadas) pueden necesitar valor mayor.
- **Outputs se persisten a disco**: al terminar, el notebook se guarda via REST con los nuevos outputs. Jupyter Lab puede pedir "Revert to disk" si el usuario tiene cambios no guardados en el browser.
- **Celdas vacias se saltan**: una celda de codigo cuyo `source` es solo espacios o saltos de linea se omite (execution_count queda None, outputs=[]).
- **`jupyter_run_cells` como dependencia futura**: cuando `jupyter_run_cells_py_notebook` este disponible, el batch de ejecucion puede delegarse a esa funcion. Hoy la logica es autonoma.
python/functions/notebook/jupyter_run_all.py
+315
View File
@@ -0,0 +1,315 @@
"""Ejecuta todas las celdas de codigo de un notebook en orden (Run All).
Equivalente al boton "Run All" del UI de Jupyter Lab, pero invocable desde
CLI/MCP/agente. Opcionalmente reinicia el kernel antes de empezar para
garantizar un estado limpio.
Depende de jupyter_run_cells cuando esta disponible; si no, ejecuta la logica
de batch internamente reutilizando los helpers de jupyter_exec.
"""
from __future__ import annotations
import json
import time
from typing import Any
from urllib.error import HTTPError, URLError
from urllib.request import Request, urlopen
from jupyter_kernel_client import KernelClient
# ---------------------------------------------------------------------------
# Helpers REST (minimos, alineados con jupyter_exec)
# ---------------------------------------------------------------------------
def _auth_headers(token: str, content_type: bool = False) -> dict[str, str]:
headers = {"Accept": "application/json"}
if content_type:
headers["Content-Type"] = "application/json"
if token:
headers["Authorization"] = f"token {token}"
return headers
def _api_get(url: str, token: str = "") -> dict | list | None:
try:
req = Request(url, headers=_auth_headers(token))
with urlopen(req, timeout=10) as resp:
return json.loads(resp.read())
except (URLError, OSError, json.JSONDecodeError):
return None
def _api_post(url: str, token: str = "", body: dict | None = None) -> dict | None:
data = json.dumps(body or {}).encode("utf-8")
req = Request(url, data=data, headers=_auth_headers(token, content_type=True), method="POST")
try:
with urlopen(req, timeout=30) as resp:
raw = resp.read()
return json.loads(raw) if raw else {}
except (URLError, OSError, json.JSONDecodeError):
return None
def _resolve_kernel_id(server_url: str, token: str, notebook_path: str) -> str | None:
"""Busca el kernel_id activo para el notebook via /api/sessions."""
sessions = _api_get(f"{server_url}/api/sessions", token) or []
for session in sessions:
nb = session.get("notebook", session.get("path", {}))
nb_path = nb.get("path", nb) if isinstance(nb, dict) else str(nb)
if nb_path == notebook_path:
kernel = session.get("kernel", {})
return kernel.get("id")
return None
def _get_notebook_content(notebook_path: str, server_url: str, token: str) -> dict:
"""Lee el notebook completo via /api/contents."""
url = f"{server_url}/api/contents/{notebook_path}?content=1&type=notebook"
req = Request(url, headers=_auth_headers(token))
with urlopen(req, timeout=15) as resp:
return json.loads(resp.read())
def _put_notebook_content(notebook_path: str, server_url: str, token: str, content: dict) -> None:
"""Persiste el notebook via PUT /api/contents."""
body = json.dumps({"type": "notebook", "format": "json", "content": content}).encode("utf-8")
url = f"{server_url}/api/contents/{notebook_path}"
req = Request(url, data=body, headers=_auth_headers(token, content_type=True), method="PUT")
with urlopen(req, timeout=15) as resp:
resp.read()
def _extract_outputs(raw_outputs: list[dict]) -> list[str]:
"""Convierte outputs nbformat a strings legibles."""
result: list[str] = []
for output in raw_outputs:
output_type = output.get("output_type", "")
if output_type == "stream":
text = output.get("text", "")
if isinstance(text, list):
text = "".join(text)
result.append(text.rstrip("\n"))
elif output_type in ("display_data", "execute_result"):
data = output.get("data", {})
text = data.get("text/plain", "")
if isinstance(text, list):
text = "".join(text)
result.append(text.rstrip("\n"))
elif output_type == "error":
traceback = output.get("traceback", [])
result.append("\n".join(traceback))
return result
def _kernel_outputs_to_nbformat(outputs: list[dict]) -> list[dict]:
return [dict(o) for o in outputs]
def _restart_kernel_and_wait(server_url: str, token: str, kernel_id: str, poll_timeout_s: int = 30) -> None:
"""Reinicia el kernel y espera hasta que vuelva a estado idle."""
url = f"{server_url}/api/kernels/{kernel_id}/restart"
_api_post(url, token)
deadline = time.monotonic() + poll_timeout_s
while time.monotonic() < deadline:
kernels = _api_get(f"{server_url}/api/kernels", token) or []
for k in kernels:
if k.get("id") == kernel_id:
state = k.get("execution_state", "")
if state == "idle":
return
break
time.sleep(0.5)
# ---------------------------------------------------------------------------
# API publica
# ---------------------------------------------------------------------------
def jupyter_run_all(
notebook_path: str,
server_url: str = "http://localhost:8888",
token: str = "",
restart_kernel: bool = True,
stop_on_error: bool = True,
timeout_per_cell_s: int = 600,
) -> dict[str, Any]:
"""Ejecuta todas las celdas de codigo del notebook en orden.
Equivalente al boton "Run All" del UI de Jupyter Lab. Si restart_kernel
es True, reinicia el kernel del notebook ANTES de empezar para garantizar
un estado completamente limpio (sin variables residuales de ejecuciones
anteriores).
Args:
notebook_path: Ruta relativa al notebook desde la raiz del servidor
(ej: "notebooks/analisis.ipynb").
server_url: URL base del servidor Jupyter (default http://localhost:8888).
token: Token de autenticacion. Vacio si el servidor no requiere auth.
restart_kernel: Si True, reinicia el kernel antes de empezar.
Garantiza estado limpio. Default True.
stop_on_error: Si True, detiene la ejecucion cuando una celda produce
un error (output_type == "error"). Default True.
timeout_per_cell_s: Timeout en segundos por celda. Default 600 (10 min).
Returns:
{
"notebook": str, # ruta del notebook
"code_cell_indices": [int], # indices de celdas de codigo ejecutadas
"executed": [ # resultado de cada celda ejecutada
{
"cell_index": int,
"execution_count": int | None,
"outputs": [str | dict],
"error": str | None, # mensaje del error si hubo, sino None
"duration_s": float,
}
],
"stopped_at": int | None, # cell_index donde se detuvo por error
"kernel_id": str,
"total_duration_s": float,
}
Raises:
RuntimeError: Si no existe sesion activa para el notebook (el kernel
no esta corriendo). Usar jupyter_exec para crear sesion.
HTTPError: Si el servidor Jupyter devuelve un error HTTP.
URLError: Si no se puede conectar al servidor.
"""
t0 = time.monotonic()
# 1. Obtener kernel_id del notebook
kernel_id = _resolve_kernel_id(server_url, token, notebook_path)
if not kernel_id:
raise RuntimeError(
f"No hay sesion activa para '{notebook_path}'. "
"Abre el notebook en Jupyter Lab o usa jupyter_exec para crear sesion."
)
# 2. Reiniciar kernel si se solicita
if restart_kernel:
_restart_kernel_and_wait(server_url, token, kernel_id)
# 3. Leer notebook y filtrar indices de celdas de codigo
file_node = _get_notebook_content(notebook_path, server_url, token)
nb = file_node["content"]
cells = nb.get("cells", [])
code_cell_indices = [
i for i, cell in enumerate(cells)
if cell.get("cell_type") == "code"
]
# 4. Ejecutar cada celda en orden
executed: list[dict] = []
stopped_at: int | None = None
with KernelClient(server_url=server_url, token=token, kernel_id=kernel_id) as kernel:
for cell_index in code_cell_indices:
cell = cells[cell_index]
source = cell.get("source", "")
if isinstance(source, list):
source = "".join(source)
# Saltar celdas vacias
if not source.strip():
executed.append({
"cell_index": cell_index,
"execution_count": None,
"outputs": [],
"error": None,
"duration_s": 0.0,
})
continue
cell_t0 = time.monotonic()
result = kernel.execute(source)
cell_duration = time.monotonic() - cell_t0
raw_outputs = result.get("outputs", [])
cell["outputs"] = _kernel_outputs_to_nbformat(raw_outputs)
cell["execution_count"] = result.get("execution_count")
# Detectar error en outputs
error_msg: str | None = None
for out in raw_outputs:
if out.get("output_type") == "error":
ename = out.get("ename", "Error")
evalue = out.get("evalue", "")
error_msg = f"{ename}: {evalue}"
break
executed.append({
"cell_index": cell_index,
"execution_count": result.get("execution_count"),
"outputs": _extract_outputs(raw_outputs),
"error": error_msg,
"duration_s": round(cell_duration, 3),
})
if error_msg and stop_on_error:
stopped_at = cell_index
break
# 5. Persistir notebook con outputs actualizados
_put_notebook_content(notebook_path, server_url, token, nb)
total_duration = time.monotonic() - t0
return {
"notebook": notebook_path,
"code_cell_indices": code_cell_indices,
"executed": executed,
"stopped_at": stopped_at,
"kernel_id": kernel_id,
"total_duration_s": round(total_duration, 3),
}
# ---------------------------------------------------------------------------
# CLI
# ---------------------------------------------------------------------------
if __name__ == "__main__":
import argparse
import sys
parser = argparse.ArgumentParser(
description="Ejecuta todas las celdas de codigo de un notebook en orden (Run All)"
)
parser.add_argument("notebook", help="Ruta del notebook relativa al servidor Jupyter")
parser.add_argument("--server", default="http://localhost:8888", help="URL del servidor Jupyter")
parser.add_argument("--token", default="", help="Token de autenticacion")
parser.add_argument(
"--no-restart",
action="store_true",
help="No reiniciar el kernel antes de ejecutar",
)
parser.add_argument(
"--continue-on-error",
action="store_true",
help="Continuar ejecucion aunque una celda produzca error",
)
parser.add_argument(
"--timeout",
type=int,
default=600,
help="Timeout en segundos por celda (default: 600)",
)
args = parser.parse_args()
try:
result = jupyter_run_all(
notebook_path=args.notebook,
server_url=args.server,
token=args.token,
restart_kernel=not args.no_restart,
stop_on_error=not args.continue_on_error,
timeout_per_cell_s=args.timeout,
)
print(json.dumps(result, ensure_ascii=False, indent=2))
except Exception as exc:
print(json.dumps({"error": str(exc)}, ensure_ascii=False), file=sys.stderr)
sys.exit(1)
python/functions/notebook/jupyter_run_cells.md
+94
View File
@@ -0,0 +1,94 @@
---
name: jupyter_run_cells
kind: function
lang: py
domain: notebook
version: "1.0.0"
purity: impure
signature: "jupyter_run_cells(notebook_path: str, cell_indices: list[int], server_url: str, token: str, stop_on_error: bool, timeout_per_cell_s: int) -> dict"
description: "Ejecuta un lote de celdas existentes (por indice) en una sola conexion WebSocket. Un GET inicial + un PUT final. Latencia fija ~3s en vez de ~3s * N de jupyter_execute_cell individual."
tags: [jupyter, notebook, kernel, websocket, execution, cells, batch, notebook]
uses_functions: [jupyter_exec_py_notebook]
uses_types: []
returns: []
returns_optional: false
error_type: "error_go_core"
imports: [jupyter_kernel_client, notebook.jupyter_exec]
params:
- name: notebook_path
desc: "Ruta relativa al notebook (relativa a la raiz del servidor Jupyter)"
- name: cell_indices
desc: "Lista de indices de celdas a ejecutar (0-based, en orden). Solo celdas de tipo code."
- name: server_url
desc: "URL del servidor Jupyter (default http://localhost:8888)"
- name: token
desc: "Token de autenticacion (default vacio = sin auth)"
- name: stop_on_error
desc: "Si True, para al primer output de tipo error. El PUT se hace con lo ejecutado hasta ese punto."
- name: timeout_per_cell_s
desc: "Timeout en segundos por cada ejecucion individual de celda (default 600)"
output: "Dict con notebook, executed (lista de resultados por celda), stopped_at, kernel_id y total_duration_s"
tested: true
tests:
- "test_run_cells_single_cell_returns_output"
- "test_run_cells_stops_on_error_by_default"
- "test_run_cells_no_stop_on_error_continues"
- "test_run_cells_invalid_index_raises"
- "test_run_cells_non_code_cell_raises"
- "e2e: test_e2e_run_cells_batch"
- "e2e: test_e2e_run_cells_stop_on_error"
test_file_path: "python/functions/notebook/tests/test_jupyter_run_cells.py"
file_path: "python/functions/notebook/jupyter_run_cells.py"
---
## Ejemplo
```python
from notebook.jupyter_run_cells import jupyter_run_cells
result = jupyter_run_cells(
notebook_path="notebooks/analisis.ipynb",
cell_indices=[0, 1, 2, 5],
server_url="http://localhost:8888",
token="",
stop_on_error=True,
)
# {
# "notebook": "notebooks/analisis.ipynb",
# "executed": [
# {"cell_index": 0, "execution_count": 1, "outputs": ["import ok"], "error": None, "duration_s": 1.2},
# {"cell_index": 1, "execution_count": 2, "outputs": ["42"], "error": None, "duration_s": 0.1},
# ...
# ],
# "stopped_at": None,
# "kernel_id": "abc-123",
# "total_duration_s": 4.8,
# }
```
CLI:
```bash
# Indices como argumentos posicionales
python python/functions/notebook/jupyter_run_cells.py notebooks/analisis.ipynb 0 1 2 5
# Indices via stdin JSON (util desde scripts)
echo '[0, 1, 2, 5]' | python python/functions/notebook/jupyter_run_cells.py notebooks/analisis.ipynb
# No parar en error + timeout custom
python python/functions/notebook/jupyter_run_cells.py notebooks/analisis.ipynb 0 1 2 \
--no-stop-on-error --timeout 120
```
## Cuando usarla
Cuando necesites re-ejecutar varias celdas de un notebook existente y el overhead de abrir/cerrar N conexiones WS sea inaceptable (>3 celdas, celdas pesadas, o en pipelines automatizados). Sustituye a llamar `jupyter_execute_cell` N veces en bucle desde el agente.
## Gotchas
- Las celdas deben existir previamente en el notebook. Para anadir y ejecutar celdas nuevas, usar `jupyter_append_execute` de `jupyter_exec`.
- Solo ejecuta celdas de tipo `code`. Pasar el indice de una celda markdown lanza `ValueError` antes de abrir el WS.
- El `PUT` final se hace siempre, incluso si `stop_on_error` detiene el lote. El notebook queda con los outputs de las celdas ejecutadas hasta el punto de parada.
- `KernelClient` ignora `timeout_per_cell_s` si la implementacion subyacente no lo soporta (depende de la version de `jupyter-kernel-client`). En ese caso el timeout global del proceso es el unico limite.
- Si el servidor Jupyter no esta corriendo, `_ensure_session` lanza `URLError` inmediatamente (no hay retry incorporado).
- El WS se abre con el `kernel_id` de la sesion activa del notebook. Si el kernel muere entre el `_ensure_session` y la ejecucion, `KernelClient` lanzara una excepcion.
python/functions/notebook/jupyter_run_cells.py
+192
View File
@@ -0,0 +1,192 @@
"""Ejecuta un lote de celdas existentes en una sola conexion WebSocket.
A diferencia de `jupyter_execute_cell` (que abre/cierra un WS por celda),
esta funcion comparte una unica sesion WS para todas las celdas del lote.
Un solo GET /api/contents al inicio + un solo PUT al final.
Latencia total: ~3s fija (overhead) + tiempo de ejecucion real de las celdas,
en lugar de ~3s * N (una conexion por celda con `jupyter_execute_cell`).
"""
import json
import time
from typing import Any
from jupyter_kernel_client import KernelClient
from notebook.jupyter_exec import (
_ensure_session,
_extract_outputs,
_get_notebook_content,
_kernel_outputs_to_nbformat,
_put_notebook_content,
)
def jupyter_run_cells(
notebook_path: str,
cell_indices: list[int],
server_url: str = "http://localhost:8888",
token: str = "",
stop_on_error: bool = True,
timeout_per_cell_s: int = 600,
) -> dict[str, Any]:
"""Ejecuta una lista de celdas existentes (por indice) en un solo paso.
Abre UNA conexion WebSocket para todo el lote. Lee el notebook una vez al
inicio y lo persiste una vez al final. Si `stop_on_error` es True y una
celda produce un output de tipo 'error', el lote para en esa celda y el
PUT se hace con los outputs ejecutados hasta ese momento.
Args:
notebook_path: Ruta relativa al notebook (relativa a la raiz del
servidor Jupyter).
cell_indices: Lista de indices de celdas a ejecutar (0-based, en
orden). Deben ser celdas de tipo 'code'.
server_url: URL del servidor Jupyter (default 'http://localhost:8888').
token: Token de autenticacion (default vacio = sin auth).
stop_on_error: Si True, para al encontrar el primer error en outputs.
timeout_per_cell_s: Timeout en segundos por cada ejecucion individual
de celda (pasado a KernelClient.execute).
Returns:
{
"notebook": str, # notebook_path recibido
"executed": [ # celdas ejecutadas (en orden)
{
"cell_index": int,
"execution_count": int | None,
"outputs": list[str], # strings legibles (via _extract_outputs)
"error": str | None, # traceback si hubo error, else None
"duration_s": float,
},
...
],
"stopped_at": int | None, # indice donde paro si stop_on_error
"kernel_id": str,
"total_duration_s": float,
}
"""
t_total_start = time.monotonic()
kernel_id = _ensure_session(server_url, token, notebook_path)
file_node = _get_notebook_content(notebook_path, server_url, token)
nb = file_node["content"]
cells = nb.get("cells", [])
# Validacion anticipada: todos los indices deben estar en rango y ser code
for idx in cell_indices:
if idx < 0 or idx >= len(cells):
raise IndexError(
f"cell_index {idx} fuera de rango (notebook tiene {len(cells)} celdas)"
)
if cells[idx].get("cell_type") != "code":
raise ValueError(
f"La celda {idx} no es de codigo (cell_type={cells[idx].get('cell_type')!r})"
)
executed: list[dict[str, Any]] = []
stopped_at: int | None = None
with KernelClient(server_url=server_url, token=token, kernel_id=kernel_id) as kernel:
for idx in cell_indices:
cell = cells[idx]
source = cell.get("source", "")
if isinstance(source, list):
source = "".join(source)
t_cell_start = time.monotonic()
result = kernel.execute(source, timeout=timeout_per_cell_s)
duration_s = round(time.monotonic() - t_cell_start, 3)
raw_outputs = result.get("outputs", [])
cell["outputs"] = _kernel_outputs_to_nbformat(raw_outputs)
cell["execution_count"] = result.get("execution_count")
readable_outputs = _extract_outputs(raw_outputs)
# Detectar si hubo error
error_text: str | None = None
for out in raw_outputs:
if out.get("output_type") == "error":
tb = out.get("traceback", [])
error_text = "\n".join(tb) if isinstance(tb, list) else str(tb)
break
executed.append({
"cell_index": idx,
"execution_count": result.get("execution_count"),
"outputs": readable_outputs,
"error": error_text,
"duration_s": duration_s,
})
if stop_on_error and error_text is not None:
stopped_at = idx
break
# Persiste notebook con todos los outputs actualizados de una vez
_put_notebook_content(notebook_path, server_url, token, nb)
return {
"notebook": notebook_path,
"executed": executed,
"stopped_at": stopped_at,
"kernel_id": kernel_id,
"total_duration_s": round(time.monotonic() - t_total_start, 3),
}
# ---------------------------------------------------------------------------
# CLI
# ---------------------------------------------------------------------------
if __name__ == "__main__":
import argparse
import sys
parser = argparse.ArgumentParser(
description="Ejecuta un lote de celdas en un solo paso WS"
)
parser.add_argument("notebook", help="Ruta al notebook relativa al servidor")
parser.add_argument(
"indices",
nargs="*",
type=int,
help="Indices de celdas a ejecutar (0-based). Si se omiten, lee JSON de stdin.",
)
parser.add_argument("--server", default="http://localhost:8888")
parser.add_argument("--token", default="")
parser.add_argument(
"--no-stop-on-error",
action="store_true",
help="Continuar aunque una celda emita error",
)
parser.add_argument("--timeout", type=int, default=600, help="Timeout por celda en segundos")
args = parser.parse_args()
if args.indices:
indices = args.indices
else:
raw = sys.stdin.read().strip()
indices = json.loads(raw) if raw else []
if not indices:
print(json.dumps({"error": "No se especificaron indices de celdas"}), file=sys.stderr)
sys.exit(1)
try:
result = jupyter_run_cells(
notebook_path=args.notebook,
cell_indices=indices,
server_url=args.server,
token=args.token,
stop_on_error=not args.no_stop_on_error,
timeout_per_cell_s=args.timeout,
)
print(json.dumps(result, ensure_ascii=False, indent=2))
except Exception as exc:
print(json.dumps({"error": str(exc)}, ensure_ascii=False), file=sys.stderr)
sys.exit(1)
python/functions/notebook/tests/test_jupyter_run_cells.py
+319
View File
@@ -0,0 +1,319 @@
"""Tests para jupyter_run_cells.
Cubre:
- Validacion de indices (rango y tipo de celda).
- Comportamiento stop_on_error (True/False).
- E2E contra un Jupyter Lab vivo si esta disponible (skip si no).
"""
from __future__ import annotations
import json
import os
import socket
import subprocess
import sys
import time
import urllib.request
from pathlib import Path
from unittest.mock import MagicMock, call, patch
import pytest
sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..", "..", "..", ".."))
from python.functions.notebook import jupyter_exec as jx
from python.functions.notebook import jupyter_run_cells as jrc
# ---------------------------------------------------------------------------
# Helpers para mocks
# ---------------------------------------------------------------------------
def _make_nb(cells: list[dict]) -> dict:
"""Construye un dict de notebook minimo con las celdas dadas."""
return {
"nbformat": 4,
"nbformat_minor": 5,
"metadata": {},
"cells": cells,
}
def _code_cell(source: str) -> dict:
return {
"cell_type": "code",
"source": source,
"outputs": [],
"execution_count": None,
"metadata": {},
"id": "test-id",
}
def _markdown_cell(source: str) -> dict:
return {
"cell_type": "markdown",
"source": source,
"metadata": {},
"id": "md-id",
}
def _kernel_result(outputs: list[dict], execution_count: int = 1) -> dict:
return {"outputs": outputs, "execution_count": execution_count, "status": "ok"}
def _stream_output(text: str) -> dict:
return {"output_type": "stream", "name": "stdout", "text": text}
def _error_output(ename: str = "ValueError", traceback: list[str] | None = None) -> dict:
return {
"output_type": "error",
"ename": ename,
"evalue": "bad value",
"traceback": traceback or [f"{ename}: bad value"],
}
class _FakeKernel:
"""Simula KernelClient con una lista de resultados predefinidos."""
def __init__(self, results: list[dict]):
self._results = iter(results)
def execute(self, source: str, **kwargs) -> dict:
return next(self._results)
def __enter__(self):
return self
def __exit__(self, *args):
pass
# ---------------------------------------------------------------------------
# Tests unitarios
# ---------------------------------------------------------------------------
def _patch_infra(nb: dict, kernel: _FakeKernel):
"""Context managers que parchean _ensure_session, _get/put_notebook y KernelClient."""
file_node = {"content": nb}
return [
patch.object(jrc, "_ensure_session", return_value="kernel-abc"),
patch.object(jrc, "_get_notebook_content", return_value=file_node),
patch.object(jrc, "_put_notebook_content"),
patch("python.functions.notebook.jupyter_run_cells.KernelClient", return_value=kernel),
]
def test_run_cells_single_cell_returns_output():
nb = _make_nb([_code_cell("print(42)")])
fake_kernel = _FakeKernel([_kernel_result([_stream_output("42\n")], execution_count=1)])
with (
patch.object(jrc, "_ensure_session", return_value="k1"),
patch.object(jrc, "_get_notebook_content", return_value={"content": nb}),
patch.object(jrc, "_put_notebook_content") as mock_put,
patch("python.functions.notebook.jupyter_run_cells.KernelClient", return_value=fake_kernel),
):
result = jrc.jupyter_run_cells(
"nb.ipynb", [0], server_url="http://localhost:8888", token=""
)
assert result["kernel_id"] == "k1"
assert result["stopped_at"] is None
assert len(result["executed"]) == 1
entry = result["executed"][0]
assert entry["cell_index"] == 0
assert entry["execution_count"] == 1
assert entry["outputs"] == ["42"]
assert entry["error"] is None
assert entry["duration_s"] >= 0
mock_put.assert_called_once()
def test_run_cells_stops_on_error_by_default():
nb = _make_nb([_code_cell("x=1"), _code_cell("bad"), _code_cell("ok")])
fake_kernel = _FakeKernel([
_kernel_result([_stream_output("1")], 1),
_kernel_result([_error_output("RuntimeError", ["RuntimeError: bad"])], 2),
# la celda 2 nunca deberia ejecutarse
])
with (
patch.object(jrc, "_ensure_session", return_value="k2"),
patch.object(jrc, "_get_notebook_content", return_value={"content": nb}),
patch.object(jrc, "_put_notebook_content"),
patch("python.functions.notebook.jupyter_run_cells.KernelClient", return_value=fake_kernel),
):
result = jrc.jupyter_run_cells("nb.ipynb", [0, 1, 2], stop_on_error=True)
assert result["stopped_at"] == 1
assert len(result["executed"]) == 2
assert result["executed"][1]["error"] is not None
assert "RuntimeError" in result["executed"][1]["error"]
def test_run_cells_no_stop_on_error_continues():
nb = _make_nb([_code_cell("bad"), _code_cell("print('after')")])
fake_kernel = _FakeKernel([
_kernel_result([_error_output()], 1),
_kernel_result([_stream_output("after")], 2),
])
with (
patch.object(jrc, "_ensure_session", return_value="k3"),
patch.object(jrc, "_get_notebook_content", return_value={"content": nb}),
patch.object(jrc, "_put_notebook_content"),
patch("python.functions.notebook.jupyter_run_cells.KernelClient", return_value=fake_kernel),
):
result = jrc.jupyter_run_cells("nb.ipynb", [0, 1], stop_on_error=False)
assert result["stopped_at"] is None
assert len(result["executed"]) == 2
assert result["executed"][0]["error"] is not None
assert result["executed"][1]["outputs"] == ["after"]
def test_run_cells_invalid_index_raises():
nb = _make_nb([_code_cell("x=1")])
with (
patch.object(jrc, "_ensure_session", return_value="k4"),
patch.object(jrc, "_get_notebook_content", return_value={"content": nb}),
patch.object(jrc, "_put_notebook_content"),
patch("python.functions.notebook.jupyter_run_cells.KernelClient"),
):
with pytest.raises(IndexError, match="fuera de rango"):
jrc.jupyter_run_cells("nb.ipynb", [5])
def test_run_cells_non_code_cell_raises():
nb = _make_nb([_markdown_cell("# titulo")])
with (
patch.object(jrc, "_ensure_session", return_value="k5"),
patch.object(jrc, "_get_notebook_content", return_value={"content": nb}),
patch.object(jrc, "_put_notebook_content"),
patch("python.functions.notebook.jupyter_run_cells.KernelClient"),
):
with pytest.raises(ValueError, match="no es de codigo"):
jrc.jupyter_run_cells("nb.ipynb", [0])
# ---------------------------------------------------------------------------
# E2E (requiere Jupyter Lab corriendo)
# ---------------------------------------------------------------------------
JUPYTER_VENV_BIN = Path("/home/lucas/fn_registry/analysis/pruebas_jupyter/.venv/bin")
def _free_port() -> int:
with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
s.bind(("127.0.0.1", 0))
return s.getsockname()[1]
def _wait_http(url: str, timeout: float = 10.0) -> bool:
end = time.time() + timeout
while time.time() < end:
try:
with urllib.request.urlopen(url, timeout=1):
return True
except OSError:
time.sleep(0.3)
return False
@pytest.fixture(scope="module")
def jupyter_server(tmp_path_factory):
"""Arranca un Jupyter Lab en puerto libre. Skip si las deps no estan."""
if not (JUPYTER_VENV_BIN / "jupyter-lab").exists():
pytest.skip("Jupyter Lab no disponible en pruebas_jupyter venv")
workdir = tmp_path_factory.mktemp("jupyter_run_cells_e2e")
(workdir / "notebooks").mkdir()
port = _free_port()
proc = subprocess.Popen(
[
str(JUPYTER_VENV_BIN / "jupyter-lab"),
f"--port={port}",
"--no-browser",
"--ServerApp.token=",
"--ServerApp.password=",
"--ServerApp.disable_check_xsrf=True",
f"--ServerApp.root_dir={workdir}",
"--collaborative",
],
stdout=subprocess.DEVNULL,
stderr=subprocess.DEVNULL,
)
server_url = f"http://localhost:{port}"
if not _wait_http(f"{server_url}/api"):
proc.terminate()
pytest.skip("Jupyter Lab no levanto a tiempo")
yield server_url, workdir
proc.terminate()
try:
proc.wait(timeout=5)
except subprocess.TimeoutExpired:
proc.kill()
def test_e2e_run_cells_batch(jupyter_server):
"""Ejecuta 3 celdas en lote y verifica outputs y persistence."""
server_url, workdir = jupyter_server
# Prepara el notebook con 3 celdas via jupyter_exec
jx.jupyter_append_execute("notebooks/batch.ipynb", "x = 10", server_url=server_url)
jx.jupyter_append_execute("notebooks/batch.ipynb", "y = x * 3", server_url=server_url)
jx.jupyter_append_execute("notebooks/batch.ipynb", "print(x + y)", server_url=server_url)
# Ejecuta el lote
result = jrc.jupyter_run_cells(
"notebooks/batch.ipynb",
[0, 1, 2],
server_url=server_url,
)
assert result["stopped_at"] is None
assert len(result["executed"]) == 3
assert result["executed"][2]["outputs"] == ["40"]
assert result["total_duration_s"] > 0
assert result["kernel_id"] != ""
# Verifica persistencia en disco
nb = json.loads((workdir / "notebooks" / "batch.ipynb").read_text())
assert nb["cells"][2]["execution_count"] is not None
def test_e2e_run_cells_stop_on_error(jupyter_server):
"""Verifica que stop_on_error detiene el lote en la celda con error."""
server_url, workdir = jupyter_server
jx.jupyter_append_execute("notebooks/stopper.ipynb", "a = 1", server_url=server_url)
jx.jupyter_append_execute("notebooks/stopper.ipynb", "raise ValueError('boom')", server_url=server_url)
jx.jupyter_append_execute("notebooks/stopper.ipynb", "print('no llego')", server_url=server_url)
result = jrc.jupyter_run_cells(
"notebooks/stopper.ipynb",
[0, 1, 2],
server_url=server_url,
stop_on_error=True,
)
assert result["stopped_at"] == 1
assert len(result["executed"]) == 2
assert result["executed"][1]["error"] is not None
assert "ValueError" in result["executed"][1]["error"]
# La celda 2 no debe aparecer en executed
assert all(e["cell_index"] != 2 for e in result["executed"])