clan-app: Add Webview to python async backend
This commit is contained in:
Qubasa
122
pkgs/clan-app/clan_app/api/__init__.py
Normal file
122
pkgs/clan-app/clan_app/api/__init__.py
Normal file
@@ -0,0 +1,122 @@
|
||||
|
||||
import importlib
|
||||
import inspect
|
||||
import logging
|
||||
import pkgutil
|
||||
from collections.abc import Callable
|
||||
from types import ModuleType
|
||||
from typing import Any, ClassVar, Generic, ParamSpec, TypeVar
|
||||
|
||||
from gi.repository import GLib, GObject
|
||||
|
||||
log = logging.getLogger(__name__)
|
||||
|
||||
|
||||
class GResult(GObject.Object):
|
||||
result: Any
|
||||
op_key: str
|
||||
method_name: str
|
||||
|
||||
def __init__(self, result: Any, method_name: str, op_key: str) -> None:
|
||||
super().__init__()
|
||||
self.op_key = op_key
|
||||
self.result = result
|
||||
self.method_name = method_name
|
||||
|
||||
|
||||
B = TypeVar('B')
|
||||
P = ParamSpec('P')
|
||||
class ImplFunc(GObject.Object, Generic[P, B]):
|
||||
op_key: str | None = None
|
||||
__gsignals__: ClassVar = {
|
||||
"returns": (GObject.SignalFlags.RUN_FIRST, None, [GObject.Object]),
|
||||
}
|
||||
def returns(self, result: B) -> None:
|
||||
self.emit("returns", GResult(result, self.__class__.__name__, self.op_key))
|
||||
|
||||
def await_result(self, fn: Callable[[GObject.Object, B], None]) -> None:
|
||||
self.connect("returns", fn)
|
||||
|
||||
def async_run(self, *args: P.args, **kwargs: P.kwargs) -> bool:
|
||||
raise NotImplementedError("Method 'async_run' must be implemented")
|
||||
|
||||
def _async_run(self, data: Any, op_key: str) -> bool:
|
||||
self.op_key = op_key
|
||||
result = GLib.SOURCE_REMOVE
|
||||
try:
|
||||
result = self.async_run(**data)
|
||||
except Exception as e:
|
||||
log.exception(e)
|
||||
# TODO: send error to js
|
||||
finally:
|
||||
return result
|
||||
|
||||
|
||||
def is_gobject_subclass(obj: object) -> bool:
|
||||
return inspect.isclass(obj) and issubclass(obj, ImplFunc) and obj is not ImplFunc
|
||||
|
||||
def check_module_for_gobject_classes(module: ModuleType, found_classes: list[type[GObject.Object]] | None = None) -> list[type[GObject.Object]]:
|
||||
if found_classes is None:
|
||||
found_classes = []
|
||||
|
||||
for name, obj in inspect.getmembers(module):
|
||||
if is_gobject_subclass(obj):
|
||||
found_classes.append(obj)
|
||||
|
||||
if hasattr(module, '__path__'): # Check if the module has submodules
|
||||
for _, submodule_name, _ in pkgutil.iter_modules(module.__path__, module.__name__ + '.'):
|
||||
submodule = importlib.import_module(submodule_name)
|
||||
check_module_for_gobject_classes(submodule, found_classes)
|
||||
|
||||
return found_classes
|
||||
|
||||
class ImplApi:
|
||||
def __init__(self) -> None:
|
||||
self._obj_registry: dict[str, type[ImplFunc]] = {}
|
||||
|
||||
def register_all(self, module: ModuleType) -> None:
|
||||
objects = check_module_for_gobject_classes(module)
|
||||
for obj in objects:
|
||||
self.register(obj)
|
||||
|
||||
def register(self, obj: type[ImplFunc]) -> None:
|
||||
fn_name = obj.__name__
|
||||
if fn_name in self._obj_registry:
|
||||
raise ValueError(f"Function '{fn_name}' already registered")
|
||||
self._obj_registry[fn_name] = obj
|
||||
|
||||
def validate(self,
|
||||
abstr_methods: dict[str, dict[str, Any]]
|
||||
) -> None:
|
||||
impl_fns = self._obj_registry
|
||||
|
||||
# iterate over the methods and check if all are implemented
|
||||
for abstr_name, abstr_annotations in abstr_methods.items():
|
||||
if abstr_name not in impl_fns:
|
||||
raise NotImplementedError(
|
||||
f"Abstract method '{abstr_name}' is not implemented"
|
||||
)
|
||||
else:
|
||||
# check if the signature of the abstract method matches the implementation
|
||||
# abstract signature
|
||||
values = list(abstr_annotations.values())
|
||||
expected_signature = (tuple(values[:-1]), values[-1:][0])
|
||||
|
||||
# implementation signature
|
||||
obj = dict(impl_fns[abstr_name].__dict__)
|
||||
obj_type = obj["__orig_bases__"][0]
|
||||
got_signature = obj_type.__args__
|
||||
|
||||
if expected_signature != got_signature:
|
||||
log.error(f"Expected signature: {expected_signature}")
|
||||
log.error(f"Actual signature: {got_signature}")
|
||||
raise ValueError(
|
||||
f"Abstract method '{abstr_name}' has different signature than the implementation"
|
||||
)
|
||||
|
||||
def get_obj(self, name: str) -> type[ImplFunc] | None:
|
||||
return self._obj_registry.get(name)
|
||||
|
||||
|
||||
|
||||
|
||||
@@ -1,103 +1,88 @@
|
||||
# ruff: noqa: N801
|
||||
import gi
|
||||
|
||||
gi.require_version("WebKit", "6.0")
|
||||
|
||||
from gi.repository import Gio, GLib, Gtk, WebKit
|
||||
gi.require_version("Gtk", "4.0")
|
||||
|
||||
import logging
|
||||
|
||||
from clan_cli.api.directory import FileRequest
|
||||
from gi.repository import Gio, GLib, Gtk
|
||||
|
||||
from clan_app.api import ImplFunc
|
||||
|
||||
log = logging.getLogger(__name__)
|
||||
|
||||
|
||||
# Implement the abstract open_file function
|
||||
def open_file(file_request: FileRequest) -> str | None:
|
||||
# Function to handle the response and stop the loop
|
||||
selected_path = None
|
||||
# This implements the abstract function open_file with one argument, file_request,
|
||||
# which is a FileRequest object and returns a string or None.
|
||||
class open_file(ImplFunc[[FileRequest], str | None]):
|
||||
def __init__(self) -> None:
|
||||
super().__init__()
|
||||
|
||||
def on_file_select(
|
||||
file_dialog: Gtk.FileDialog, task: Gio.Task, main_loop: GLib.MainLoop
|
||||
) -> 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}")
|
||||
finally:
|
||||
main_loop.quit()
|
||||
def async_run(self, file_request: FileRequest) -> bool:
|
||||
def on_file_select(file_dialog: Gtk.FileDialog, task: Gio.Task) -> None:
|
||||
try:
|
||||
gfile = file_dialog.open_finish(task)
|
||||
if gfile:
|
||||
selected_path = gfile.get_path()
|
||||
self.returns(selected_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, main_loop: GLib.MainLoop
|
||||
) -> 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}")
|
||||
finally:
|
||||
main_loop.quit()
|
||||
def on_folder_select(file_dialog: Gtk.FileDialog, task: Gio.Task) -> None:
|
||||
try:
|
||||
gfile = file_dialog.select_folder_finish(task)
|
||||
if gfile:
|
||||
selected_path = gfile.get_path()
|
||||
self.returns(selected_path)
|
||||
except Exception as e:
|
||||
print(f"Error getting selected directory: {e}")
|
||||
|
||||
def on_save_finish(
|
||||
file_dialog: Gtk.FileDialog, task: Gio.Task, main_loop: GLib.MainLoop
|
||||
) -> 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}")
|
||||
finally:
|
||||
main_loop.quit()
|
||||
def on_save_finish(file_dialog: Gtk.FileDialog, task: Gio.Task) -> None:
|
||||
try:
|
||||
gfile = file_dialog.save_finish(task)
|
||||
if gfile:
|
||||
selected_path = gfile.get_path()
|
||||
self.returns(selected_path)
|
||||
except Exception as e:
|
||||
print(f"Error getting selected file: {e}")
|
||||
|
||||
dialog = Gtk.FileDialog()
|
||||
dialog = Gtk.FileDialog()
|
||||
|
||||
if file_request.title:
|
||||
dialog.set_title(file_request.title)
|
||||
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:
|
||||
filters = Gio.ListStore.new(Gtk.FileFilter)
|
||||
file_filters = Gtk.FileFilter()
|
||||
|
||||
if file_request.filters.title:
|
||||
file_filters.set_name(file_request.filters.title)
|
||||
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.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.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)
|
||||
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)
|
||||
filters.append(file_filters)
|
||||
dialog.set_filters(filters)
|
||||
|
||||
main_loop = GLib.MainLoop()
|
||||
# if select_folder
|
||||
if file_request.mode == "select_folder":
|
||||
dialog.select_folder(callback=on_folder_select)
|
||||
elif file_request.mode == "open_file":
|
||||
dialog.open(callback=on_file_select)
|
||||
elif file_request.mode == "save":
|
||||
dialog.save(callback=on_save_finish)
|
||||
|
||||
# if select_folder
|
||||
if file_request.mode == "select_folder":
|
||||
dialog.select_folder(
|
||||
callback=lambda dialog, task: on_folder_select(dialog, task, main_loop),
|
||||
)
|
||||
elif file_request.mode == "open_file":
|
||||
dialog.open(
|
||||
callback=lambda dialog, task: on_file_select(dialog, task, main_loop)
|
||||
)
|
||||
elif file_request.mode == "save":
|
||||
dialog.save(
|
||||
callback=lambda dialog, task: on_save_finish(dialog, task, main_loop)
|
||||
)
|
||||
return GLib.SOURCE_REMOVE
|
||||
|
||||
# Wait for the user to select a file or directory
|
||||
main_loop.run() # type: ignore
|
||||
|
||||
return selected_path
|
||||
|
||||
Reference in New Issue
Block a user