fn-registry agent
101 lines
3.7 KiB
Python
101 lines
3.7 KiB
Python
"""
|
|
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()")
|