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

feat: mejoras notebook functions — discover multi-servidor, write batch ops

Browse Source 
jupyter_discover: soporte multi-servidor, detección de modo colaborativo mejorada.
jupyter_write: operaciones batch (insert, edit, delete), manejo robusto de Y.js.
jupyter_exec: mejoras en ejecución directa al kernel.

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
This commit is contained in:
Egutierrez
2026-04-05 17:11:50 +02:00
parent 5a324f6554
commit af1fa129f7
7 changed files with 668 additions and 28 deletions
Download Patch File Download Diff File Expand all files Collapse all files
python/functions/notebook/jupyter_discover.py
+132 -12
View File
@@ -30,9 +30,101 @@ def _is_collaborative(config: dict | list | None) -> bool:
return "ydocextension" in raw or "collaborative" in raw
def _query_instance(base_url: str) -> dict | None:
def _find_jupyter_pid_for_port(port: int) -> int | None:
"""Busca en /proc el PID del proceso jupyter que escucha en el puerto dado.
Solo funciona en Linux (donde /proc existe). Retorna None si no encuentra
el proceso o si /proc no esta disponible.
"""
proc_dir = Path("/proc")
if not proc_dir.is_dir():
return None
for pid_entry in proc_dir.iterdir():
if not pid_entry.name.isdigit():
continue
cmdline_path = pid_entry / "cmdline"
try:
raw = cmdline_path.read_bytes().decode("utf-8", errors="replace")
except OSError:
continue
if "jupyter" not in raw:
continue
# Los argumentos estan separados por \0 en /proc/pid/cmdline
parts = raw.split("\0")
port_found = False
for part in parts:
if part in (f"--port={port}", f"--ServerApp.port={port}"):
port_found = True
break
# Para el puerto default 8888, el proceso puede no tener --port explícito.
# En ese caso verificamos que sea un proceso jupyter y no tenga otro puerto.
if not port_found and port == 8888:
has_other_port = any(
p.startswith("--port=") or p.startswith("--ServerApp.port=")
for p in parts
)
if not has_other_port and any("jupyter" in p for p in parts):
port_found = True
if port_found:
try:
return int(pid_entry.name)
except ValueError:
continue
return None
def _get_root_dir_from_proc(pid: int) -> str:
"""Extrae el root_dir del proceso Jupyter a partir de su cmdline en /proc.
Busca --ServerApp.root_dir= o --notebook-dir= en los argumentos del proceso.
Si no los encuentra, usa el cwd del proceso como fallback.
Retorna cadena vacia si no puede leer /proc.
"""
try:
cmdline_path = f"/proc/{pid}/cmdline"
with open(cmdline_path, "rb") as f:
parts = f.read().decode("utf-8", errors="replace").split("\0")
for part in parts:
if part.startswith("--ServerApp.root_dir="):
return part.split("=", 1)[1].rstrip("/")
if part.startswith("--notebook-dir="):
return part.split("=", 1)[1].rstrip("/")
# Fallback: cwd del proceso
cwd = os.readlink(f"/proc/{pid}/cwd")
return cwd.rstrip("/")
except OSError:
return ""
def _extract_analysis_name(root_dir: str) -> str:
"""Extrae el nombre del analisis del root_dir.
Si root_dir contiene 'analysis/{nombre}', retorna '{nombre}'.
En caso contrario retorna el ultimo componente del path.
"""
if not root_dir:
return ""
parts = root_dir.replace("\\", "/").split("/")
# Buscar el segmento 'analysis' y tomar el siguiente
for i, part in enumerate(parts):
if part == "analysis" and i + 1 < len(parts) and parts[i + 1]:
return parts[i + 1]
# Fallback: ultimo componente del path
return parts[-1] if parts else ""
def _query_instance(base_url: str, port: int) -> dict | None:
"""Consulta la API REST de una instancia Jupyter y retorna su estado.
Detecta el root_dir real del proceso via /proc (Linux) para identificar
correctamente el analisis que esta sirviendo.
Retorna None si la instancia no responde o no es Jupyter.
"""
status = _get(f"{base_url}/api/status")
@@ -43,6 +135,12 @@ def _query_instance(base_url: str) -> dict | None:
kernels_raw = _get(f"{base_url}/api/kernels") or []
sessions_raw = _get(f"{base_url}/api/sessions") or []
# Detectar root_dir via /proc
root_dir = ""
pid = _find_jupyter_pid_for_port(port)
if pid is not None:
root_dir = _get_root_dir_from_proc(pid)
kernels = []
if isinstance(kernels_raw, list):
for k in kernels_raw:
@@ -67,6 +165,7 @@ def _query_instance(base_url: str) -> dict | None:
})
return {
"root_dir": root_dir,
"kernels": kernels,
"sessions": sessions,
"collaborative": _is_collaborative(config),
@@ -110,6 +209,10 @@ def jupyter_discover(
comunes (8888-8892). Para cada instancia que responde consulta /api/status,
/api/config, /api/kernels y /api/sessions.
En Linux detecta el root_dir real del proceso Jupyter via /proc/pid/cmdline,
lo que permite identificar correctamente el analisis en escenarios
multi-instancia donde varios Jupyter corren en puertos distintos.
Args:
registry_root: Raiz del fn_registry. Si vacio usa el directorio actual
o la variable de entorno FN_REGISTRY_ROOT.
@@ -117,8 +220,9 @@ def jupyter_discover(
en .jupyter-port mas los defaults (8888-8892).
Returns:
Lista de dicts con: url, port, analysis, collaborative, kernels, sessions.
Cada sesion incluye: notebook, kernel_id, kernel_state.
Lista de dicts con: url, port, analysis, root_dir, collaborative,
kernels, sessions. Cada sesion incluye: notebook, kernel_id,
kernel_state.
"""
if not registry_root:
registry_root = os.environ.get("FN_REGISTRY_ROOT", "")
@@ -139,15 +243,25 @@ def jupyter_discover(
port_analysis[p] = ""
results = []
for port, analysis_name in port_analysis.items():
for port, analysis_hint in port_analysis.items():
base_url = f"http://localhost:{port}"
info = _query_instance(base_url)
info = _query_instance(base_url, port)
if info is None:
continue
# Determinar analysis name: preferir deteccion via /proc sobre .jupyter-port
root_dir = info["root_dir"]
if root_dir:
analysis_name = _extract_analysis_name(root_dir)
else:
# Fallback al hint del .jupyter-port
analysis_name = analysis_hint
results.append({
"url": base_url,
"port": port,
"analysis": analysis_name,
"root_dir": root_dir,
"collaborative": info["collaborative"],
"kernels": info["kernels"],
"sessions": info["sessions"],
@@ -201,19 +315,25 @@ if __name__ == "__main__":
sys.exit(0)
for inst in instances:
label = f" analysis: {inst['analysis']}" if inst["analysis"] else ""
collab = "colaborativo" if inst["collaborative"] else "estandar"
print(f"Jupyter Lab en {inst['url']} [{collab}]{label}")
analysis = inst["analysis"] or "(desconocido)"
root_dir = inst["root_dir"] or "(no detectado)"
print(f"Puerto {inst['port']} [{collab}]")
print(f" url: {inst['url']}")
print(f" analysis: {analysis}")
print(f" root_dir: {root_dir}")
kernel_count = len(inst["kernels"])
if inst["kernels"]:
print(f" Kernels ({len(inst['kernels'])}):")
print(f" kernels ({kernel_count}):")
for k in inst["kernels"]:
print(f" - {k['name']} estado={k['execution_state']} id={k['id'][:8]}...")
else:
print(" Kernels: ninguno")
print(" kernels: ninguno")
if inst["sessions"]:
print(f" Sesiones ({len(inst['sessions'])}):")
print(f" sesiones ({len(inst['sessions'])}):")
for s in inst["sessions"]:
print(f" - {s['notebook']} kernel={s['kernel_id'][:8]}... estado={s['kernel_state']}")
kid = s["kernel_id"][:8] + "..." if s["kernel_id"] else "(sin kernel)"
print(f" - {s['notebook']} kernel={kid} estado={s['kernel_state']}")
else:
print(" Sesiones: ninguna")
print(" sesiones: ninguna")
print()