[DAS-Dashboard#1225] Update usage of query params on command router.

das-dashboard/backend/controllers/query_controllers.py

@@ -39,6 +39,7 @@ def create_execution_on_proxy(body: QueryExecutionDto):
     response = QUERY_SERVICES.execute_proxy_command(
         body.command_type,
         body.command_text,
+        body.parameters,
     )
 
     return JSONResponse(
@@ -134,6 +135,11 @@ async def _safe_send_error(
 
 def _proxy_json_content(response: Response) -> Any:
     try:
-        return response.json()
+        content = response.json()
     except ValueError:
         return {"content": response.text}
+
+    if isinstance(content, dict) and "error" in content and "message" not in content:
+        content = {**content, "message": content["error"]}
+
+    return content

das-dashboard/backend/services/query_services.py

@@ -1,7 +1,6 @@
 import json
 import logging
 from collections.abc import AsyncIterator
-from concurrent.futures import ThreadPoolExecutor, as_completed
 
 import requests
 from requests import Response
@@ -13,9 +12,9 @@
 from shared.exceptions.custom_exceptions import CommandRouterConnectionError, CustomValueError
 from shared.internal.constants import LOCAL_HOSTS
 from shared.internal.web_configuration import WebConfiguration
+from shared.utils.command_router_payload import build_query_execution_payload
 from shared.utils.parse_query_answer import transform_stream_event
 
-VALID_COMMAND_TYPES = ("get", "set", "query")  # Evolution will be disconsidered for now.
 ROUTE_PREFIX = "/command-router"
 
 logger = logging.getLogger(__name__)
@@ -64,56 +63,18 @@ def get_execution_answers(
     def cancel_query_execution(self, execution_id: str) -> Response:
         return self._call_http_proxy("POST", f"{ROUTE_PREFIX}/executions/{execution_id}/cancel")
 
-    def execute_proxy_command(self, command_type: str, command_text: str) -> Response:
-        handlers = {
-            "get": self._get_query_parameters,
-            "set": self._set_query_parameters,
-            "query": self._create_query_execution,
-        }
-
-        if command_type not in VALID_COMMAND_TYPES:
+    def execute_proxy_command(
+        self,
+        command_type: str,
+        command_text: str,
+        parameters: dict | None = None,
+    ) -> Response:
+        if command_type != "query":
             raise CustomValueError(
-                "This proxy command does not exist/is not permitted in the current context."
+                "Only query executions are allowed for this method."
             )
 
-        return handlers[command_type](command_type, command_text)
-
-    def _get_query_parameters(self, command_type: str, command_text: str) -> Response:
-        return self._post_execution(command_type, command_text)
-
-    def _set_query_parameters(self, command_type: str, command_text: str) -> Response:
-        try:
-            custom_parameters_dict = json.loads(command_text)
-        except json.JSONDecodeError as error:
-            raise CustomValueError(f"Invalid query parameters payload: {error}") from error
-
-        if not custom_parameters_dict:
-            raise CustomValueError("No query parameters were provided.")
-
-        responses = []
-        max_workers = min(len(custom_parameters_dict), 8)
-
-        with ThreadPoolExecutor(max_workers=max_workers) as executor:
-            futures = [
-                executor.submit(
-                    self._post_execution,
-                    command_type,
-                    f"param {key} {self._format_param_value(value)}",
-                )
-                for key, value in custom_parameters_dict.items()
-            ]
-
-            for future in as_completed(futures):
-                responses.append(future.result())
-
-        failed_response = next(
-            (response for response in responses if response.status_code >= 400),
-            None,
-        )
-        if failed_response is not None:
-            return failed_response
-
-        return responses[-1]
+        return self._create_query_execution(command_text, parameters)
 
     def get_default_params_from_config(self) -> dict:
         config = self.web_config.load_raw_configuration()
@@ -131,20 +92,20 @@ def get_default_params_from_config(self) -> dict:
 
         return defaults
 
-    @staticmethod
-    def _format_param_value(value) -> str:
-        if isinstance(value, bool):
-            return "true" if value else "false"
-        return str(value)
-
-    def _create_query_execution(self, command_type: str, command_text: str) -> Response:
-        return self._post_execution(command_type, command_text)
+    def _create_query_execution(
+        self,
+        command_text: str,
+        parameters: dict | None = None,
+    ) -> Response:
+        try:
+            payload = build_query_execution_payload(command_text, parameters)
+        except ValueError as error:
+            raise CustomValueError(str(error)) from error
 
-    def _post_execution(self, command_type: str, command_text: str) -> Response:
         return self._call_http_proxy(
             "POST",
             f"{ROUTE_PREFIX}/executions",
-            json={"command_type": command_type, "command_text": command_text},
+            json=payload,
         )
 
     def _call_http_proxy(self, method: str, path: str, **request_kwargs) -> Response:

das-dashboard/backend/shared/dtos/query_execution_dto.py

@@ -1,5 +1,8 @@
+from typing import Any
+
 from pydantic import BaseModel
 
 class QueryExecutionDto(BaseModel):
     command_type: str
     command_text: str
+    parameters: dict[str, Any] | None = None

das-dashboard/backend/shared/utils/command_router_payload.py

@@ -0,0 +1,40 @@
+from typing import Any
+
+
+def build_query_execution_payload(
+    command_text: str,
+    parameters: dict[str, Any] | None = None,
+) -> dict[str, Any]:
+    trimmed = command_text.strip()
+    if not trimmed:
+        raise ValueError("Query text must not be empty.")
+
+    params: dict[str, Any] = {
+        "query": {
+            "syntax": "metta",
+            "tokens": [trimmed],
+        }
+    }
+
+    if parameters:
+        params.update(normalize_router_parameters(parameters))
+
+    return {"command": "query", "params": params}
+
+
+def normalize_router_parameters(parameters: dict[str, Any]) -> dict[str, Any]:
+    normalized: dict[str, Any] = {}
+
+    for key, value in parameters.items():
+        if isinstance(value, bool):
+            normalized[key] = value
+        elif isinstance(value, int):
+            normalized[key] = value
+        elif isinstance(value, float):
+            normalized[key] = value
+        elif isinstance(value, str):
+            normalized[key] = value
+        else:
+            normalized[key] = value
+
+    return normalized

das-dashboard/backend/shared/utils/parse_query_answer.py

@@ -37,6 +37,43 @@ def normalize_query_answer(item: Any) -> dict[str, Any] | None:
 
 
 def transform_stream_event(event: dict[str, Any]) -> dict[str, Any]:
+    command = event.get("command")
+    params = event.get("params")
+
+    if command == "query_answers" and isinstance(params, dict):
+        raw_items = params.get("answers")
+        if not isinstance(raw_items, list):
+            return event
+
+        data = []
+        for item in raw_items:
+            normalized = normalize_query_answer(item)
+            if normalized is not None:
+                data.append(normalized)
+
+        return {
+            "execution_id": params.get("execution_id") or event.get("execution_id"),
+            "type": "chunk",
+            "seq": params.get("seq"),
+            "received_count": params.get("received_count"),
+            "data": data,
+        }
+
+    if command == "execution_status" and isinstance(params, dict):
+        transformed: dict[str, Any] = {
+            "execution_id": params.get("execution_id"),
+            "status": params.get("status"),
+        }
+
+        if params.get("message"):
+            transformed["message"] = params["message"]
+        if params.get("total_items") is not None:
+            transformed["received_count"] = params["total_items"]
+        elif params.get("received_count") is not None:
+            transformed["received_count"] = params["received_count"]
+
+        return transformed
+
     if event.get("type") != "chunk":
         return event
 

das-dashboard/src/api/APIUtils.js

@@ -3,7 +3,7 @@ export function extractErrorMessage(err, fallback = "An unexpected error occurre
     return fallback;
   }
 
-  if (err.response) {
+    if (err.response) {
     const data = err.response.data;
 
     if (typeof data === "string") {
@@ -13,6 +13,10 @@ export function extractErrorMessage(err, fallback = "An unexpected error occurre
     if (data?.message) {
       return data.message;
     }
+
+    if (data?.error) {
+      return data.error;
+    }
   }
 
   if (err.message) {
@@ -38,6 +42,10 @@ export function extractErrorDetails(err) {
       return data.exceptionMessage;
     }
 
+    if (data?.error) {
+      return typeof data.error === "string" ? data.error : JSON.stringify(data.error);
+    }
+
     if (data?.detail) {
       return typeof data.detail === "string" ? data.detail : JSON.stringify(data.detail);
     }

das-dashboard/src/api/QueryAPI.js

@@ -5,19 +5,17 @@ export async function getQueryParamDefaults() {
   return response.data;
 }
 
-export async function setQueryParameters(params) {
-  const response = await api.post("/query/executions", {
-    command_type: "set",
-    command_text: JSON.stringify(params)
-  });
-  return response.data;
-}
-
-export async function startQueryExecution(queryText) {
-  const response = await api.post("/query/executions", {
+export async function startQueryExecution(queryText, parameters = null) {
+  const payload = {
     command_type: "query",
     command_text: queryText
-  });
+  };
+
+  if (parameters && Object.keys(parameters).length > 0) {
+    payload.parameters = parameters;
+  }
+
+  const response = await api.post("/query/executions", payload);
   return response.data;
 }
 

das-dashboard/src/hooks/useQueryExecution.js

@@ -1,7 +1,6 @@
 import { useCallback, useEffect, useMemo, useRef, useState } from "react";
 import {
   cancelQueryExecution,
-  setQueryParameters,
   startQueryExecution
 } from "../api/QueryAPI";
 import { extractApiError } from "../api/APIUtils";
@@ -179,12 +178,11 @@ export function useQueryExecution(parameters) {
       startedAtRef.current = Date.now();
 
       try {
-        const pendingParams = parameters.consumeQueryRunParameters();
-        if (Object.keys(pendingParams).length > 0) {
-          await setQueryParameters(pendingParams);
-        }
-
-        const { execution_id: nextExecutionId } = await startQueryExecution(trimmedQuery);
+        const runParameters = parameters.collectParameters();
+        const { execution_id: nextExecutionId } = await startQueryExecution(
+          trimmedQuery,
+          runParameters
+        );
         if (!nextExecutionId) {
           throw new Error("Query execution did not return an execution id.");
         }