Single import box for multiple file formats (#80)

* Add the ability to override automatic parameter creation

* Add a new ParameterGroup class

* Correctly process kwargs in operations as parameters, not inputs

---------

Co-authored-by: JMLizano <[email protected]>

biome.json

@@ -10,6 +10,9 @@
         "noExplicitAny": "off",
         "noArrayIndexKey": "off"
       },
+      "correctness": {
+        "useExhaustiveDependencies": "off"
+      },
       "style": {
         "noNonNullAssertion": "off"
       },

lynxkite-app/.gitignore

@@ -2,3 +2,4 @@
 !/src/lynxkite_app/web_assets/__init__.py
 !/src/lynxkite_app/web_assets/assets/__init__.py
 data/
+!/web/tests/data

lynxkite-app/web/src/workspace/nodes/NodeGroupParameter.tsx

@@ -0,0 +1,67 @@
+import { useEffect, useState } from "react";
+import NodeParameter from "./NodeParameter";
+
+interface SelectorType {
+  name: string;
+  default: string;
+  type: {
+    enum: string[];
+  };
+}
+
+interface ParameterType {
+  name: string;
+  default: string;
+  type: {
+    type: string;
+  };
+}
+
+interface GroupsType {
+  [key: string]: ParameterType[];
+}
+
+interface NodeGroupParameterProps {
+  meta: { selector: SelectorType; groups: GroupsType };
+  value: any;
+  setParam: (name: string, value: any, options?: { delay: number }) => void;
+  deleteParam: (name: string, options?: { delay: number }) => void;
+}
+
+export default function NodeGroupParameter({
+  meta,
+  value,
+  setParam,
+  deleteParam,
+}: NodeGroupParameterProps) {
+  const selector = meta.selector;
+  const groups = meta.groups;
+  const [selectedValue, setSelectedValue] = useState<string>(
+    value || selector.default,
+  );
+
+  const handleSelectorChange = (value: any, opts?: { delay: number }) => {
+    setSelectedValue(value);
+    setParam(selector.name, value, opts);
+  };
+
+  useEffect(() => {
+    // Clean possible previous parameters first
+    Object.values(groups).flatMap((group) =>
+      group.map((entry) => deleteParam(entry.name)),
+    );
+    for (const param of groups[selectedValue]) {
+      setParam(param.name, param.default);
+    }
+  }, [selectedValue]);
+
+  return (
+    <NodeParameter
+      name={selector.name}
+      key={selector.name}
+      value={selectedValue}
+      meta={selector}
+      onChange={handleSelectorChange}
+    />
+  );
+}

lynxkite-app/web/src/workspace/nodes/NodeWithParams.tsx

@@ -1,5 +1,6 @@
 import { useReactFlow } from "@xyflow/react";
 import LynxKiteNode from "./LynxKiteNode";
+import NodeGroupParameter from "./NodeGroupParameter";
 import NodeParameter from "./NodeParameter";
 
 export type UpdateOptions = { delay?: number };
@@ -7,9 +8,22 @@ export type UpdateOptions = { delay?: number };
 function NodeWithParams(props: any) {
   const reactFlow = useReactFlow();
   const metaParams = props.data.meta?.params;
+
   function setParam(name: string, newValue: any, opts: UpdateOptions) {
+    reactFlow.updateNodeData(props.id, (prevData: any) => ({
+      ...prevData,
+      params: { ...prevData.data.params, [name]: newValue },
+      __execution_delay: opts.delay || 0,
+    }));
+  }
+
+  function deleteParam(name: string, opts: UpdateOptions) {
+    if (props.data.params[name] === undefined) {
+      return;
+    }
+    delete props.data.params[name];
     reactFlow.updateNodeData(props.id, {
-      params: { ...props.data.params, [name]: newValue },
+      params: { ...props.data.params },
       __execution_delay: opts.delay || 0,
     });
   }
@@ -17,17 +31,31 @@ function NodeWithParams(props: any) {
 
   return (
     <LynxKiteNode {...props}>
-      {params.map(([name, value]) => (
-        <NodeParameter
-          name={name}
-          key={name}
-          value={value}
-          meta={metaParams?.[name]}
-          onChange={(value: any, opts?: UpdateOptions) =>
-            setParam(name, value, opts || {})
-          }
-        />
-      ))}
+      {params.map(([name, value]) =>
+        metaParams?.[name]?.type === "group" ? (
+          <NodeGroupParameter
+            key={name}
+            value={value}
+            meta={metaParams?.[name]}
+            setParam={(name: string, value: any, opts?: UpdateOptions) =>
+              setParam(name, value, opts || {})
+            }
+            deleteParam={(name: string, opts?: UpdateOptions) =>
+              deleteParam(name, opts || {})
+            }
+          />
+        ) : (
+          <NodeParameter
+            name={name}
+            key={name}
+            value={value}
+            meta={metaParams?.[name]}
+            onChange={(value: any, opts?: UpdateOptions) =>
+              setParam(name, value, opts || {})
+            }
+          />
+        ),
+      )}
       {props.children}
     </LynxKiteNode>
   );

lynxkite-app/web/tests/data/import_test.json

@@ -0,0 +1 @@
+{ "name": { "0": "Adam", "1": "Eve", "2": "Bob", "3": "Isolated Joe" } }

lynxkite-app/web/tests/{upload.spec.ts → import.spec.ts}

@@ -9,7 +9,7 @@ let workspace: Workspace;
 test.beforeEach(async ({ browser }) => {
   workspace = await Workspace.empty(
     await browser.newPage(),
-    "upload_spec_test",
+    "import_spec_test",
   );
 });
 
@@ -19,25 +19,57 @@ test.afterEach(async () => {
   splash.page.on("dialog", async (dialog) => {
     await dialog.accept();
   });
-  await splash.deleteEntry("upload_spec_test");
+  await splash.deleteEntry("import_spec_test");
 });
 
-test("can upload and import a simple CSV", async () => {
+async function validateImport(
+  workspace: Workspace,
+  fileName: string,
+  fileFormat: string,
+) {
   const __filename = fileURLToPath(import.meta.url);
   const __dirname = dirname(__filename);
-  const csvPath = join(__dirname, "data", "upload_test.csv");
+  const filePath = join(__dirname, "data", fileName);
 
-  await workspace.addBox("Import CSV");
-  const csvBox = workspace.getBox("Import CSV 1");
-  const filenameInput = csvBox.locator("input.input-bordered").nth(0);
-  await filenameInput.click();
-  await filenameInput.fill(csvPath);
-  await filenameInput.press("Enter");
+  await workspace.addBox("Import file");
+  const importBox = workspace.getBox("Import file 1");
+  const fileFormatSelect = await importBox
+    .locator("label.param", { hasText: "file format" })
+    .locator("select");
+  await fileFormatSelect.selectOption(fileFormat);
+  const filePathInput = await importBox
+    .locator("label.param", { hasText: "file path" })
+    .locator("input");
+  await filePathInput.click();
+  await filePathInput.fill(filePath);
+  await filePathInput.press("Enter");
+  const tableNameInput = await importBox
+    .locator("label.param", { hasText: "table name" })
+    .locator("input");
+  await tableNameInput.click();
+  await tableNameInput.fill("table");
+  await tableNameInput.press("Enter");
 
   await workspace.addBox("View tables");
   const tableBox = workspace.getBox("View tables 1");
-  await workspace.connectBoxes("Import CSV 1", "View tables 1");
+  await workspace.connectBoxes("Import file 1", "View tables 1");
 
   const tableRows = tableBox.locator("table tbody tr");
   await expect(tableRows).toHaveCount(4);
+}
+
+test("Can import a CSV file", async () => {
+  await validateImport(workspace, "import_test.csv", "csv");
+});
+
+test("Can import a parquet file", async () => {
+  await validateImport(workspace, "import_test.parquet", "parquet");
+});
+
+test("Can import a JSON file", async () => {
+  await validateImport(workspace, "import_test.json", "json");
+});
+
+test("Can import an Excel file", async () => {
+  await validateImport(workspace, "import_test.xlsx", "excel");
 });

lynxkite-app/web/tests/lynxkite.ts

@@ -155,7 +155,7 @@ export class Workspace {
     await new Promise((resolve) =>
       setTimeout(resolve, executionWaitTime ? executionWaitTime : 500),
     );
-    await expect(this.getBoxes().locator(".error")).not.toBeVisible();
+    await expect(this.getBoxes().locator(".error").first()).not.toBeVisible();
   }
 
   async close() {

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

@@ -75,6 +75,16 @@ class Parameter(BaseConfig):
         return Parameter(name=name, default=default, type=type)
 
 
+class ParameterGroup(BaseConfig):
+    """Defines a group of parameters for an operation."""
+
+    name: str
+    selector: Parameter
+    default: typing.Any
+    groups: dict[str, list[Parameter]]
+    type: str = "group"
+
+
 class Input(BaseConfig):
     name: str
     type: Type
@@ -116,7 +126,7 @@ def basic_outputs(*names):
 class Op(BaseConfig):
     func: typing.Callable = pydantic.Field(exclude=True)
     name: str
-    params: dict[str, Parameter]
+    params: dict[str, Parameter | ParameterGroup]
     inputs: dict[str, Input]
     outputs: dict[str, Output]
     # TODO: Make type an enum with the possible values.
@@ -148,7 +158,7 @@ class Op(BaseConfig):
         return res
 
 
-def op(env: str, name: str, *, view="basic", outputs=None):
+def op(env: str, name: str, *, view="basic", outputs=None, params=None):
     """Decorator for defining an operation."""
 
     def decorator(func):
@@ -157,12 +167,14 @@ def op(env: str, name: str, *, view="basic", outputs=None):
         inputs = {
             name: Input(name=name, type=param.annotation)
             for name, param in sig.parameters.items()
-            if param.kind != param.KEYWORD_ONLY
+            if param.kind not in (param.KEYWORD_ONLY, param.VAR_KEYWORD)
         }
-        params = {}
+        _params = {}
         for n, param in sig.parameters.items():
             if param.kind == param.KEYWORD_ONLY and not n.startswith("_"):
-                params[n] = Parameter.basic(n, param.default, param.annotation)
+                _params[n] = Parameter.basic(n, param.default, param.annotation)
+        if params:
+            _params.update(params)
         if outputs:
             _outputs = {name: Output(name=name, type=None) for name in outputs}
         else:
@@ -172,7 +184,7 @@ def op(env: str, name: str, *, view="basic", outputs=None):
         op = Op(
             func=func,
             name=name,
-            params=params,
+            params=_params,
             inputs=inputs,
             outputs=_outputs,
             type=view,

lynxkite-graph-analytics/src/lynxkite_graph_analytics/lynxkite_ops.py

@@ -1,5 +1,6 @@
 """Graph analytics operations."""
 
+import enum
 import os
 import fsspec
 from lynxkite.core import ops
@@ -18,6 +19,68 @@ mem = joblib.Memory("../joblib-cache")
 op = ops.op_registration(core.ENV)
 
 
+class FileFormat(enum.StrEnum):
+    csv = "csv"
+    parquet = "parquet"
+    json = "json"
+    excel = "excel"
+
+
+@op(
+    "Import file",
+    params={
+        "file_format": ops.ParameterGroup(
+            name="file_format",
+            selector=ops.Parameter(
+                name="file_format", type=FileFormat, default=FileFormat.csv
+            ),
+            groups={
+                "csv": [
+                    ops.Parameter.basic("columns", type=str, default="<from file>"),
+                    ops.Parameter.basic("separator", type=str, default="<auto>"),
+                ],
+                "parquet": [],
+                "json": [],
+                "excel": [
+                    ops.Parameter.basic("sheet_name", type=str, default="Sheet1")
+                ],
+            },
+            default=FileFormat.csv,
+        ),
+    },
+)
+def import_file(
+    *, file_path: str, table_name: str, file_format: FileFormat, **kwargs
+) -> core.Bundle:
+    """Read the contents of the a file into a `Bundle`.
+
+    Args:
+        file_path: Path to the file to import.
+        table_name: Name to use for identifying the table in the bundle.
+        file_format: Format of the file. Has to be one of the values in the `FileFormat` enum.
+
+    Returns:
+        Bundle: Bundle with a single table with the contents of the file.
+    """
+    if file_format == "csv":
+        names = kwargs.pop("columns", "<from file>")
+        names = (
+            pd.api.extensions.no_default if names == "<from file>" else names.split(",")
+        )
+        sep = kwargs.pop("separator", "<auto>")
+        sep = pd.api.extensions.no_default if sep == "<auto>" else sep
+        df = pd.read_csv(file_path, names=names, sep=sep, **kwargs)
+    elif file_format == "json":
+        df = pd.read_json(file_path, **kwargs)
+    elif file_format == "parquet":
+        df = pd.read_parquet(file_path, **kwargs)
+    elif file_format == "excel":
+        df = pd.read_excel(file_path, **kwargs)
+    else:
+        df = ValueError(f"Unsupported file format: {file_format}")
+    return core.Bundle(dfs={table_name: df})
+
+
 @op("Import Parquet")
 def import_parquet(*, filename: str):
     """Imports a Parquet file."""