clan-app: stash

pkgs/clan-app/clan_app/api/__init__.py

@@ -2,7 +2,7 @@
 import logging
 from collections.abc import Callable
 from typing import Any, T
-
+from functools import wraps
 from gi.repository import GObject
 
 log = logging.getLogger(__name__)
@@ -13,7 +13,8 @@ class _ImplApi:
         self._registry: dict[str, Callable[[Any], Any]] = {}
 
 
-    def register(self, fn: Callable[..., T]) -> Callable[..., T]:
+    def register(self, fn_name: str, fn: object) -> object:
+        breakpoint()
         name = fn.__name__
         if name in self._registry:
             raise ValueError(f"Function {name} already registered")
@@ -27,10 +28,11 @@ class _ImplApi:
             raise ValueError(f"Function {name} must have a 'source' annotation of type GObject")
 
         del fn.__annotations__["signal_source"]
+
         self._registry[name] = fn
 
 
-        return fn
+        return wrapper
 
     def validate(self,
         abstr_methods: dict[str, dict[str, Any]]

pkgs/clan-app/clan_app/api/file.py

@@ -5,84 +5,23 @@ gi.require_version("Gtk", "4.0")
 import logging
 
 from clan_cli.api.directory import FileRequest
-from gi.repository import Gio, Gtk
-
+from gi.repository import Gio, Gtk, GLib
 from clan_app.api import ImplApi
 from clan_app.views.webview import WebExecutor
 
 log = logging.getLogger(__name__)
 
+@ImplApi.register("open_file")
+class OpenFileObj(GObject.Object):
+    __gsignals__: ClassVar = {
+        "func_returned": (GObject.SignalFlags.RUN_FIRST, None, []),
+    }
 
-@ImplApi.register
-def open_file(signal_source: WebExecutor, file_request: FileRequest) -> str | None:
-    # Function to handle the response and stop the loop
-    selected_path = None
+    def __init__(self, file_request: FileRequest) -> None:
+        super().__init__()
 
-    def on_file_select(file_dialog: Gtk.FileDialog, task: Gio.Task) -> None:
-        try:
-            gfile = file_dialog.open_finish(task)
-            if gfile:
-                nonlocal selected_path
-                selected_path = gfile.get_path()
-        except Exception as e:
-            print(f"Error getting selected file or directory: {e}")
 
-    def on_folder_select(file_dialog: Gtk.FileDialog, task: Gio.Task) -> None:
-        try:
-            gfile = file_dialog.select_folder_finish(task)
-            if gfile:
-                nonlocal selected_path
-                selected_path = gfile.get_path()
-        except Exception as e:
-            print(f"Error getting selected directory: {e}")
+    def run(self) -> bool:
+        return GLib.SOURCE_REMOVE
 
-    def on_save_finish(file_dialog: Gtk.FileDialog, task: Gio.Task) -> None:
-        try:
-            gfile = file_dialog.save_finish(task)
-            if gfile:
-                nonlocal selected_path
-                selected_path = gfile.get_path()
-        except Exception as e:
-            print(f"Error getting selected file: {e}")
 
-    dialog = Gtk.FileDialog()
-
-    if file_request.title:
-        dialog.set_title(file_request.title)
-
-    if file_request.filters:
-        filters = Gio.ListStore.new(Gtk.FileFilter)
-        file_filters = Gtk.FileFilter()
-
-        if file_request.filters.title:
-            file_filters.set_name(file_request.filters.title)
-
-        # Create and configure a filter for image files
-        if file_request.filters.mime_types:
-            for mime in file_request.filters.mime_types:
-                file_filters.add_mime_type(mime)
-                filters.append(file_filters)
-
-        if file_request.filters.patterns:
-            for pattern in file_request.filters.patterns:
-                file_filters.add_pattern(pattern)
-
-        if file_request.filters.suffixes:
-            for suffix in file_request.filters.suffixes:
-                file_filters.add_suffix(suffix)
-
-        filters.append(file_filters)
-        dialog.set_filters(filters)
-
-    # if select_folder
-    if file_request.mode == "select_folder":
-        dialog.select_folder(
-            callback=lambda dialog, task: on_folder_select(dialog, task),
-        )
-    elif file_request.mode == "open_file":
-        dialog.open(callback=lambda dialog, task: on_file_select(dialog, task))
-    elif file_request.mode == "save":
-        dialog.save(callback=lambda dialog, task: on_save_finish(dialog, task))
-
-    log.debug(f"Selected path: {selected_path}")
-    return selected_path

pkgs/clan-app/clan_app/views/webview.py

@@ -103,10 +103,6 @@ def from_dict(t: type, data: dict[str, Any] | None) -> Any:
 
 
 class WebExecutor(GObject.Object):
-    __gsignals__: ClassVar = {
-        "func_returned": (GObject.SignalFlags.RUN_FIRST, None, []),
-    }
-
     def __init__(self, content_uri: str, abstr_methods: dict[str, Callable]) -> None:
         super().__init__()
 
@@ -138,11 +134,53 @@ class WebExecutor(GObject.Object):
         log.debug(f"Queue size: {self.queue_size} (Wait)")
 
         GLib.idle_add(
+            self.func_executor,
             handler_fn,
             payload.get("data"),
             method_name,
         )
 
+    def func_executor(
+        self,
+        handler_fn: Callable[
+            ...,
+            Any,
+        ],
+        data: dict[str, Any] | None,
+        method_name: str,
+    ) -> None:
+        with self.mutex_lock:
+            log.debug("Executing... ", method_name)
+            log.debug(f"{data}")
+            if data is None:
+                result = handler_fn()
+            else:
+                reconciled_arguments = {}
+                op_key = data.pop("op_key", None)
+                for k, v in data.items():
+                    # Some functions expect to be called with dataclass instances
+                    # But the js api returns dictionaries.
+                    # Introspect the function and create the expected dataclass from dict dynamically
+                    # Depending on the introspected argument_type
+                    arg_class = API.get_method_argtype(method_name, k)
+                    if dataclasses.is_dataclass(arg_class):
+                        reconciled_arguments[k] = from_dict(arg_class, v)
+                    else:
+                        reconciled_arguments[k] = v
+
+                r = handler_fn(**reconciled_arguments)
+                # Parse the result to a serializable dictionary
+                # Echo back the "op_key" to the js api
+                result = dataclass_to_dict(r)
+                result["op_key"] = op_key
+
+            serialized = json.dumps(result)
+
+            # Use idle_add to queue the response call to js on the main GTK thread
+            GLib.idle_add(self.return_data_to_js, method_name, serialized)
+
+
+
     def return_data_to_js(self, method_name: str, serialized: str) -> bool:
         # This function must be run on the main GTK thread to interact with the webview
         # result = method_fn(data) # takes very long