Load ops from .py files in the data directory.

examples/word2vec.py

@@ -0,0 +1,22 @@
+from lynxkite.core.ops import op
+import staticvectors
+import pandas as pd
+
+
+@op("LynxKite Graph Analytics", "Word2vec for the top 1000 words", cache=True)
+def word2vec_1000():
+    model = staticvectors.StaticVectors("neuml/word2vec-quantized")
+    with open("wordlist.txt") as f:
+        words = [w.strip() for w in f.read().strip().split("\n")]
+    df = pd.DataFrame(
+        {
+            "word": words,
+            "embedding": model.embeddings(words).tolist(),
+        }
+    )
+    return df
+
+
+@op("LynxKite Graph Analytics", "Take first N")
+def first_n(df: pd.DataFrame, *, n=10):
+    return df.head(n)

lynxkite-app/src/lynxkite_app/crdt.py

@@ -233,6 +233,7 @@ async def execute(name: str, ws_crdt: pycrdt.Map, ws_pyd: workspace.Workspace, d
     assert path.is_relative_to(cwd), "Provided workspace path is invalid"
     # Save user changes before executing, in case the execution fails.
     workspace.save(ws_pyd, path)
+    ops.load_user_scripts(name)
     ws_pyd._crdt = ws_crdt
     with ws_crdt.doc.transaction():
         for nc, np in zip(ws_crdt["nodes"], ws_pyd.nodes):

lynxkite-app/src/lynxkite_app/main.py

@@ -26,6 +26,7 @@ def detect_plugins():
 
 
 lynxkite_plugins = detect_plugins()
+ops.save_catalogs("plugins loaded")
 
 app = fastapi.FastAPI(lifespan=crdt.lifespan)
 app.include_router(crdt.router)
@@ -33,7 +34,8 @@ app.add_middleware(GZipMiddleware)
 
 
 @app.get("/api/catalog")
-def get_catalog():
+def get_catalog(workspace: str):
+    ops.load_user_scripts(workspace)
     return {k: {op.name: op.model_dump() for op in v.values()} for k, v in ops.CATALOGS.items()}
 
 

lynxkite-app/web/src/workspace/Workspace.tsx

@@ -151,7 +151,7 @@ function LynxKiteFlow() {
 
   const fetcher: Fetcher<Catalogs> = (resource: string, init?: RequestInit) =>
     fetch(resource, init).then((res) => res.json());
-  const catalog = useSWR("/api/catalog", fetcher);
+  const catalog = useSWR(`/api/catalog?workspace=${path}`, fetcher);
   const [suppressSearchUntil, setSuppressSearchUntil] = useState(0);
   const [nodeSearchSettings, setNodeSearchSettings] = useState(
     undefined as

lynxkite-core/src/lynxkite/core/ops.py

@@ -4,7 +4,11 @@ from __future__ import annotations
 import asyncio
 import enum
 import functools
+import importlib
 import inspect
+import pathlib
+import traceback
+import joblib
 import types
 import pydantic
 import typing
@@ -14,8 +18,11 @@ from typing_extensions import Annotated
 if typing.TYPE_CHECKING:
     from . import workspace
 
-CATALOGS: dict[str, dict[str, "Op"]] = {}
+Catalog = dict[str, "Op"]
+Catalogs = dict[str, Catalog]
+CATALOGS: Catalogs = {}
 EXECUTORS = {}
+mem = joblib.Memory(".joblib-cache")
 
 typeof = type  # We have some arguments called "type".
 
@@ -189,10 +196,12 @@ class Op(BaseConfig):
         return res
 
 
-def op(env: str, name: str, *, view="basic", outputs=None, params=None):
+def op(env: str, name: str, *, view="basic", outputs=None, params=None, cache=False):
     """Decorator for defining an operation."""
 
     def decorator(func):
+        if cache:
+            func = mem.cache(func)
         sig = inspect.signature(func)
         # Positional arguments are inputs.
         inputs = {
@@ -308,3 +317,41 @@ def slow(func):
         return await asyncio.to_thread(func, *args, **kwargs)
 
     return wrapper
+
+
+CATALOGS_SNAPSHOTS: dict[str, Catalogs] = {}
+
+
+def save_catalogs(snapshot_name: str):
+    CATALOGS_SNAPSHOTS[snapshot_name] = {k: dict(v) for k, v in CATALOGS.items()}
+
+
+def load_catalogs(snapshot_name: str):
+    global CATALOGS
+    snap = CATALOGS_SNAPSHOTS[snapshot_name]
+    CATALOGS = {k: dict(v) for k, v in snap.items()}
+
+
+def load_user_scripts(workspace: str):
+    """Reloads the *.py in the workspace's directory and higher-level directories."""
+    if "plugins loaded" in CATALOGS_SNAPSHOTS:
+        load_catalogs("plugins loaded")
+    cwd = pathlib.Path()
+    path = cwd / workspace
+    assert path.is_relative_to(cwd), "Provided workspace path is invalid"
+    for p in path.parents:
+        print("checking user scripts in", p)
+        for f in p.glob("*.py"):
+            try:
+                run_user_script(f)
+            except Exception:
+                traceback.print_exc()
+        if p == cwd:
+            break
+
+
+def run_user_script(script_path: pathlib.Path):
+    print(f"Running {script_path}...")
+    spec = importlib.util.spec_from_file_location(script_path.stem, str(script_path))
+    module = importlib.util.module_from_spec(spec)
+    spec.loader.exec_module(module)