chore: sync from fn-registry agent
This commit is contained in:
Binary file not shown.
@@ -0,0 +1,100 @@
|
||||
"""
|
||||
fn_registry kernel startup
|
||||
Autoconfigura acceso al registry en cada notebook.
|
||||
Generado por write_jupyter_registry_kernel (fn_registry).
|
||||
"""
|
||||
import os
|
||||
import sys
|
||||
import sqlite3
|
||||
from pathlib import Path
|
||||
|
||||
# ── FN_REGISTRY_ROOT ────────────────────────────────────────
|
||||
# Prioridad: env var > path hardcoded > descubrimiento automatico
|
||||
def _discover_registry_root():
|
||||
if os.environ.get("FN_REGISTRY_ROOT"):
|
||||
return Path(os.environ["FN_REGISTRY_ROOT"]).resolve()
|
||||
hardcoded = Path("/home/egutierrez/fn_registry")
|
||||
if (hardcoded / "registry.db").exists():
|
||||
return hardcoded
|
||||
# Subir desde CWD hasta encontrar registry.db
|
||||
p = Path.cwd()
|
||||
for _ in range(10):
|
||||
if (p / "registry.db").exists():
|
||||
return p
|
||||
if p.parent == p:
|
||||
break
|
||||
p = p.parent
|
||||
return hardcoded
|
||||
|
||||
FN_REGISTRY_ROOT = _discover_registry_root()
|
||||
os.environ["FN_REGISTRY_ROOT"] = str(FN_REGISTRY_ROOT)
|
||||
|
||||
# ── sys.path: importar funciones Python del registry ────────
|
||||
_python_functions = FN_REGISTRY_ROOT / "python" / "functions"
|
||||
for _domain in sorted(_python_functions.iterdir()) if _python_functions.exists() else []:
|
||||
if _domain.is_dir() and not _domain.name.startswith("_"):
|
||||
_path = str(_domain)
|
||||
if _path not in sys.path:
|
||||
sys.path.insert(0, _path)
|
||||
|
||||
# Tambien el directorio padre para imports por dominio: from core import filter_list
|
||||
_pf = str(_python_functions)
|
||||
if _pf not in sys.path:
|
||||
sys.path.insert(0, _pf)
|
||||
|
||||
# ── fn_query: consultar registry.db desde el notebook ───────
|
||||
_REGISTRY_DB = FN_REGISTRY_ROOT / "registry.db"
|
||||
|
||||
def fn_query(sql, params=()):
|
||||
"""Ejecuta una consulta SQL sobre registry.db y retorna las filas.
|
||||
|
||||
Ejemplos:
|
||||
fn_query("SELECT id, description FROM functions WHERE domain = ?", ("finance",))
|
||||
fn_query("SELECT id FROM functions_fts WHERE functions_fts MATCH ?", ("slice*",))
|
||||
"""
|
||||
if not _REGISTRY_DB.exists():
|
||||
raise FileNotFoundError(f"registry.db no encontrado en {_REGISTRY_DB}")
|
||||
con = sqlite3.connect(str(_REGISTRY_DB))
|
||||
con.row_factory = sqlite3.Row
|
||||
try:
|
||||
rows = con.execute(sql, params).fetchall()
|
||||
return [dict(r) for r in rows]
|
||||
finally:
|
||||
con.close()
|
||||
|
||||
def fn_search(term):
|
||||
"""Busca funciones y tipos en el registry por nombre o descripcion.
|
||||
|
||||
Ejemplo:
|
||||
fn_search("slice")
|
||||
fn_search("finance")
|
||||
"""
|
||||
fts_term = f"name:{term}* OR description:{term}*"
|
||||
functions = fn_query(
|
||||
"SELECT id, kind, purity, lang, description FROM functions "
|
||||
"WHERE id IN (SELECT id FROM functions_fts WHERE functions_fts MATCH ?) "
|
||||
"ORDER BY name", (fts_term,)
|
||||
)
|
||||
types = fn_query(
|
||||
"SELECT id, algebraic, lang, description FROM types "
|
||||
"WHERE id IN (SELECT id FROM types_fts WHERE types_fts MATCH ?) "
|
||||
"ORDER BY name", (fts_term,)
|
||||
)
|
||||
return {"functions": functions, "types": types}
|
||||
|
||||
def fn_code(function_id):
|
||||
"""Retorna el codigo fuente de una funcion del registry.
|
||||
|
||||
Ejemplo:
|
||||
print(fn_code("filter_list_py_core"))
|
||||
"""
|
||||
rows = fn_query("SELECT code FROM functions WHERE id = ?", (function_id,))
|
||||
if not rows:
|
||||
raise KeyError(f"Funcion no encontrada: {function_id}")
|
||||
return rows[0]["code"]
|
||||
|
||||
# ── Mensaje de bienvenida ───────────────────────────────────
|
||||
print(f"fn_registry conectado: {FN_REGISTRY_ROOT}")
|
||||
print(f" registry.db: {'OK' if _REGISTRY_DB.exists() else 'NO ENCONTRADO'}")
|
||||
print(f" Python functions: {_pf}")
|
||||
print(f" Helpers: fn_query(), fn_search(), fn_code()")
|
||||
Reference in New Issue
Block a user