improve jsonrpc

pyflowlauncher/__init__.py

@@ -1,18 +1,21 @@
 import logging
 
 from .plugin import Plugin
-from .result import JsonRPCAction, Result, send_results, ResultResponse
+from .result import Result, send_results
 from .method import Method
+from .jsonrpc.models import JsonRPCRequest, JsonRPCResult
+from .jsonrpc.client import send_request
 
 
 logger = logging.getLogger(__name__)
 
 
 __all__ = [
     "Plugin",
-    "ResultResponse",
     "send_results",
     "Result",
-    "JsonRPCAction",
+    "JsonRPCRequest",
+    "JsonRPCResult",
     "Method",
+    "send_request",
 ]

pyflowlauncher/api.py

@@ -1,79 +1,80 @@
 from typing import Optional
 
-from .result import JsonRPCAction
+from .jsonrpc.models import JsonRPCRequest
+from .jsonrpc.client import create_request
 
 NAME_SPACE = 'Flow.Launcher'
 
 
-def _send_action(method: str, *parameters) -> JsonRPCAction:
-    return {"method": f"{NAME_SPACE}.{method}", "parameters": parameters}
+def _get_namespace(method: str) -> str:
+    return f"{NAME_SPACE}.{method}"
 
 
-def change_query(query: str, requery: bool = False) -> JsonRPCAction:
+def change_query(query: str, requery: bool = False) -> JsonRPCRequest:
     """Change the query in Flow Launcher."""
-    return _send_action("ChangeQuery", query, requery)
+    return create_request(_get_namespace("ChangeQuery"), [query, requery])
 
 
-def shell_run(command: str, filename: str = 'cmd.exe') -> JsonRPCAction:
+def shell_run(command: str, filename: str = 'cmd.exe') -> JsonRPCRequest:
     """Run a shell command."""
-    return _send_action("ShellRun", command, filename)
+    return create_request(_get_namespace("ShellRun"), [command, filename])
 
 
-def close_app() -> JsonRPCAction:
+def close_app() -> JsonRPCRequest:
     """Close Flow Launcher."""
-    return _send_action("CloseApp")
+    return create_request(_get_namespace("CloseApp"))
 
 
-def hide_app() -> JsonRPCAction:
+def hide_app() -> JsonRPCRequest:
     """Hide Flow Launcher."""
-    return _send_action("HideApp")
+    return create_request(_get_namespace("HideApp"))
 
 
-def show_app() -> JsonRPCAction:
+def show_app() -> JsonRPCRequest:
     """Show Flow Launcher."""
-    return _send_action("ShowApp")
+    return create_request(_get_namespace("ShowApp"))
 
 
-def show_msg(title: str, sub_title: str, ico_path: str = "") -> JsonRPCAction:
+def show_msg(title: str, sub_title: str, ico_path: str = "") -> JsonRPCRequest:
     """Show a message in Flow Launcher."""
-    return _send_action("ShowMsg", title, sub_title, ico_path)
+    return create_request(_get_namespace("ShowMsg"), [title, sub_title, ico_path])
 
 
-def open_setting_dialog() -> JsonRPCAction:
+def open_setting_dialog() -> JsonRPCRequest:
     """Open the settings window in Flow Launcher."""
-    return _send_action("OpenSettingDialog")
+    return create_request(_get_namespace("OpenSettingDialog"))
 
 
-def start_loading_bar() -> JsonRPCAction:
+def start_loading_bar() -> JsonRPCRequest:
     """Start the loading bar in Flow Launcher."""
-    return _send_action("StartLoadingBar")
+    return create_request(_get_namespace("StartLoadingBar"))
 
 
-def stop_loading_bar() -> JsonRPCAction:
+def stop_loading_bar() -> JsonRPCRequest:
     """Stop the loading bar in Flow Launcher."""
-    return _send_action("StopLoadingBar")
+    return create_request(_get_namespace("StopLoadingBar"))
 
 
-def reload_plugins() -> JsonRPCAction:
+def reload_plugins() -> JsonRPCRequest:
     """Reload the plugins in Flow Launcher."""
-    return _send_action("ReloadPlugins")
+    return create_request(_get_namespace("ReloadPlugins"))
 
 
-def copy_to_clipboard(text: str, direct_copy: bool = False, show_default_notification=True) -> JsonRPCAction:
+def copy_to_clipboard(text: str, direct_copy: bool = False, show_default_notification=True) -> JsonRPCRequest:
     """Copy text to the clipboard."""
-    return _send_action("CopyToClipboard", text, direct_copy, show_default_notification)
+    return create_request(_get_namespace("CopyToClipboard"), [text, direct_copy, show_default_notification])
 
 
-def open_directory(directory_path: str, filename_or_filepath: Optional[str] = None) -> JsonRPCAction:
+def open_directory(directory_path: str, filename_or_filepath: Optional[str] = None) -> JsonRPCRequest:
     """Open a directory."""
-    return _send_action("OpenDirectory", directory_path, filename_or_filepath)
+    return create_request(_get_namespace("OpenDirectory"), [directory_path, filename_or_filepath])
 
 
-def open_url(url: str, in_private: bool = False) -> JsonRPCAction:
+def open_url(url: str, in_private: bool = False) -> JsonRPCRequest:
     """Open a URL."""
-    return _send_action("OpenUrl", url, in_private)
+    return create_request(_get_namespace("OpenUrl"), [url, in_private])
 
 
-def open_uri(uri: str) -> JsonRPCAction:
+def open_uri(uri: str) -> JsonRPCRequest:
     """Open a URI."""
-    return _send_action("OpenAppUri", uri)
+    return create_request(_get_namespace("OpenUri"), [uri])

pyflowlauncher/jsonrpc/__init__.py

@@ -0,0 +1,5 @@
+from .id_generation import incremental_int
+
+JSONRPC_VER = "2.0"
+
+ids = incremental_int()

pyflowlauncher/jsonrpc/client.py

@@ -0,0 +1,33 @@
+import json
+from typing import List, Optional
+
+from . import JSONRPC_VER, ids
+from .models import JsonRPCRequest
+
+
+def create_request(
+        method: str,
+        parameters: Optional[List] = None,
+        id: Optional[int] = None,
+        jsonrpc: str = JSONRPC_VER
+) -> JsonRPCRequest:
+    return {
+        "jsonrpc": jsonrpc,
+        "method": method,
+        "parameters": parameters or [],
+        "id": id or next(ids)
+    }
+
+
+def request_from_string(
+    method: str,
+    parameters: Optional[List] = None,
+    id: Optional[int] = None,
+) -> str:
+    return json.dumps(
+        create_request(method, parameters, id)
+    )
+
+
+def send_request(request: JsonRPCRequest) -> None:
+    print(json.dumps(request))

pyflowlauncher/jsonrpc/id_generation.py

@@ -0,0 +1,6 @@
+import itertools
+from typing import Iterator
+
+
+def incremental_int(start: int = 1) -> Iterator[int]:
+    return itertools.count(start)

pyflowlauncher/jsonrpc/models.py

@@ -0,0 +1,35 @@
+from __future__ import annotations
+
+import sys
+from typing import Any, List, Optional
+
+if sys.version_info < (3, 11):
+    from typing_extensions import NotRequired, TypedDict
+else:
+    from typing import NotRequired, TypedDict
+
+
+class BaseJsonRPCRequest(TypedDict):
+    """Standard JsonRPC Request"""
+    id: NotRequired[int]
+    jsonrpc: NotRequired[str]
+    method: str
+    parameters: List
+
+
+class JsonRPCRequest(BaseJsonRPCRequest):
+    """Flow Launcher JsonRPC Request"""
+    dontHideAfterAction: NotRequired[bool]
+    settings: NotRequired[dict]
+
+
+class BaseJsonRPCResult(TypedDict):
+    """Standard JsonRPC Result"""
+    id: NotRequired[int]
+    jsonrpc: str
+    result: Any
+
+
+class JsonRPCResult(BaseJsonRPCResult):
+    """Flow Launcher JsonRPC Result"""
+    SettingsChange: NotRequired[Optional[dict]]

pyflowlauncher/jsonrpc/server.py

@@ -0,0 +1,28 @@
+from __future__ import annotations
+
+import json
+from typing import Any, Dict, Optional
+
+from .models import JsonRPCRequest, JsonRPCResult
+
+from . import JSONRPC_VER, ids
+
+
+def parse_request(message: str) -> JsonRPCRequest:
+    request = json.loads(message)
+    if "id" not in request:
+        request["id"] = next(ids)
+    return request
+
+
+def create_response(result: Any, id: int, SettingsChange: Optional[Dict] = None) -> JsonRPCResult:
+    return {
+        "jsonrpc": JSONRPC_VER,
+        "result": result,
+        "id": id,
+        "SettingsChange": SettingsChange
+    }
+
+
+def response(result: Any, id: int, SettingsChange: Optional[Dict] = None) -> str:
+    return json.dumps(create_response(result, id, SettingsChange))

pyflowlauncher/method.py

@@ -1,10 +1,11 @@
 from __future__ import annotations
 
 from abc import ABC, abstractmethod
-from typing import Any, Dict, Optional
+from typing import Any, Dict, List, Optional
 
-from .result import JsonRPCAction, Result, ResultResponse, send_results
+from .result import Result, send_results
 from .shared import logger
+from .jsonrpc.models import JsonRPCResult
 
 
 class Method(ABC):
@@ -16,9 +17,9 @@ def __init__(self) -> None:
     def add_result(self, result: Result) -> None:
         self._results.append(result)
 
-    def return_results(self, settings: Optional[Dict[str, Any]] = None) -> ResultResponse:
-        return send_results(self._results, settings)
+    def return_results(self) -> List[Dict[str, Any]]:
+        return send_results(self._results)
 
     @abstractmethod
-    def __call__(self, *args, **kwargs) -> ResultResponse | JsonRPCAction:
+    def __call__(self, *args, **kwargs) -> Optional[JsonRPCResult]:
         pass

pyflowlauncher/plugin.py

@@ -10,18 +10,17 @@
 from pyflowlauncher.shared import logger
 
 from .event import EventHandler
-from .jsonrpc import JsonRPCClient
-from .result import JsonRPCAction, ResultResponse
+from .jsonrpc import server
+from .jsonrpc.models import JsonRPCRequest, JsonRPCResult
 from .manifest import PluginManifestSchema, MANIFEST_FILE
 
-Method = Callable[..., Union[ResultResponse, JsonRPCAction, None]]
+Method = Callable[..., Union[JsonRPCResult, None]]
 
 
 class Plugin:
 
     def __init__(self, methods: list[Method] | None = None) -> None:
         self._logger = logger(self)
-        self._client = JsonRPCClient()
         self._event_handler = EventHandler()
         self._settings: dict[str, Any] = {}
         if methods:
@@ -56,30 +55,22 @@ def wrapper(handler: Callable[..., Any]) -> Callable[..., Any]:
             return handler
         return wrapper
 
-    def action(self, method: Method, parameters: Optional[Iterable] = None) -> JsonRPCAction:
+    def action(self, method: Method, parameters: Optional[Iterable] = None) -> JsonRPCRequest:
         """Register a method and return a JsonRPCAction that calls it."""
         method_name = self.add_method(method)
-        return {"method": method_name, "parameters": parameters or []}
+        return {"method": method_name, "parameters": list(parameters or [])}
 
-    @property
-    def settings(self) -> dict:
-        if self._settings is None:
-            self._settings = {}
-        self._settings = self._client.recieve().get('settings', {})
-        return self._settings
+    async def run_async(self) -> None:
+        request = server.parse_request(sys.argv[1])
+        feedback = await self._event_handler.trigger_event(request["method"], *request["parameters"])
+        print(server.response(feedback, request["id"]))
 
     def run(self) -> None:
-        request = self._client.recieve()
-        method = request["method"]
-        parameters = request.get('parameters', [])
         if sys.version_info >= (3, 10, 0):
-            feedback = asyncio.run(self._event_handler.trigger_event(method, *parameters))
+            asyncio.run(self.run_async())
         else:
             loop = asyncio.get_event_loop()
-            feedback = loop.run_until_complete(self._event_handler.trigger_event(method, *parameters))
-        if not feedback:
-            return
-        self._client.send(feedback)
+            loop.run_until_complete(self.run_async())
 
     @property
     def run_dir(self) -> Path: