yadunut/clan-core
1
0
Fork 0
Code Issues Projects Releases Wiki Activity

clan-app: Fix mypy lints and add GenericFnRuntime

Browse Source
This commit is contained in:
Qubasa
2024-07-15 19:48:20 +02:00
parent 660ac92c29
commit 77dc90d3b4
8 changed files with 83 additions and 85 deletions
Download Patch File Download Diff File Expand all files Collapse all files

97
pkgs/clan-app/clan_app/api/__init__.py
View File

@@ -1,11 +1,6 @@
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 typing import Any, ClassVar, Generic, ParamSpec, TypeVar, cast
from gi.repository import GLib, GObject
@@ -24,17 +19,23 @@ class GResult(GObject.Object):
self.method_name = method_name
B = TypeVar('B')
P = ParamSpec('P')
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]),
"returns": (GObject.SignalFlags.RUN_FIRST, None, [GResult]),
}
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:
def returns(self, result: B) -> None:
method_name = self.__class__.__name__
if self.op_key is None:
raise ValueError(f"op_key is not set for the function {method_name}")
self.emit("returns", GResult(result, method_name, self.op_key))
def await_result(self, fn: Callable[["ImplFunc[..., Any]", B], None]) -> None:
self.connect("returns", fn)
def async_run(self, *args: P.args, **kwargs: P.kwargs) -> bool:
@@ -52,58 +53,36 @@ class ImplFunc(GObject.Object, Generic[P, B]):
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:
class GObjApi:
def __init__(self, methods: dict[str, Callable[..., Any]]) -> None:
self._methods: dict[str, Callable[..., Any]] = methods
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:
def register_overwrite(self, obj: type[ImplFunc]) -> None:
fn_name = obj.__name__
if not isinstance(obj, type(ImplFunc)):
raise ValueError(f"Object '{fn_name}' is not an instance of ImplFunc")
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
def check_signature(self, method_annotations: dict[str, dict[str, Any]]) -> None:
overwrite_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"
)
for m_name, m_annotations in method_annotations.items():
if m_name not in overwrite_fns:
continue
else:
# check if the signature of the abstract method matches the implementation
# abstract signature
values = list(abstr_annotations.values())
values = list(m_annotations.values())
expected_signature = (tuple(values[:-1]), values[-1:][0])
# implementation signature
obj = dict(impl_fns[abstr_name].__dict__)
obj = dict(overwrite_fns[m_name].__dict__)
obj_type = obj["__orig_bases__"][0]
got_signature = obj_type.__args__
@@ -111,12 +90,26 @@ class ImplApi:
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"
f"Overwritten method '{m_name}' has different signature than the implementation"
)
def get_obj(self, name: str) -> type[ImplFunc] | None:
return self._obj_registry.get(name)
def get_obj(self, name: str) -> type[ImplFunc]:
result = self._obj_registry.get(name, None)
if result is not None:
return result
plain_method = self._methods.get(name, None)
if plain_method is None:
raise ValueError(f"Method '{name}' not found in Api")
class GenericFnRuntime(ImplFunc[..., Any]):
def __init__(self) -> None:
super().__init__()
def async_run(self, *args: Any, **kwargs: dict[str, Any]) -> bool:
assert plain_method is not None
result = plain_method(*args, **kwargs)
self.returns(result)
return GLib.SOURCE_REMOVE
return cast(type[ImplFunc], GenericFnRuntime)

2
pkgs/clan-app/clan_app/api/file.py
View File

@@ -84,5 +84,3 @@ class open_file(ImplFunc[[FileRequest], str | None]):
dialog.save(callback=on_save_finish)
return GLib.SOURCE_REMOVE

3
pkgs/clan-app/clan_app/components/serializer.py
View File

@@ -1,4 +1,3 @@
import dataclasses
import logging
from dataclasses import fields, is_dataclass
@@ -90,4 +89,4 @@ def from_dict(t: type, data: dict[str, Any] | None) -> Any:
except (TypeError, ValueError) as e:
print(f"Failed to instantiate {t.__name__}: {e}")
return None
return None

33
pkgs/clan-app/clan_app/views/webview.py
View File

@@ -1,14 +1,13 @@
import dataclasses
import json
import logging
from collections.abc import Callable
from typing import Any
import gi
from clan_cli.api import API
from clan_cli.api import MethodRegistry
import clan_app
from clan_app.api import GResult, ImplApi, ImplFunc
from clan_app.api import GObjApi, GResult, ImplFunc
from clan_app.api.file import open_file
from clan_app.components.serializer import dataclass_to_dict, from_dict
gi.require_version("WebKit", "6.0")
@@ -18,19 +17,21 @@ log = logging.getLogger(__name__)
class WebExecutor(GObject.Object):
def __init__(self, content_uri: str, abstr_methods: dict[str, Callable]) -> None:
def __init__(self, content_uri: str, plain_api: MethodRegistry) -> None:
super().__init__()
self.plain_api: MethodRegistry = plain_api
self.webview: WebKit.WebView = WebKit.WebView()
self.webview = WebKit.WebView()
settings = self.webview.get_settings()
settings: WebKit.Settings = self.webview.get_settings()
# settings.
settings.set_property("enable-developer-extras", True)
self.webview.set_settings(settings)
# Fixme. This filtering is incomplete, it only triggers if a user clicks a link
self.webview.connect("decide-policy", self.on_decide_policy)
self.manager = self.webview.get_user_content_manager()
self.manager: WebKit.UserContentManager = (
self.webview.get_user_content_manager()
)
# Can be called with: window.webkit.messageHandlers.gtk.postMessage("...")
# Important: it seems postMessage must be given some payload, otherwise it won't trigger the event
self.manager.register_script_message_handler("gtk")
@@ -39,10 +40,10 @@ class WebExecutor(GObject.Object):
self.webview.load_uri(content_uri)
self.content_uri = content_uri
self.api = ImplApi()
self.api.register_all(clan_app.api)
#self.api.validate(abstr_methods)
self.api: GObjApi = GObjApi(self.plain_api.functions)
self.api.register_overwrite(open_file)
self.api.check_signature(self.plain_api.annotations)
def on_decide_policy(
self,
@@ -73,16 +74,13 @@ class WebExecutor(GObject.Object):
def on_message_received(
self, user_content_manager: WebKit.UserContentManager, message: Any
) -> None:
json_msg = message.to_json(4)
json_msg = message.to_json(4) # 4 is num of indents
log.debug(f"Webview Request: {json_msg}")
payload = json.loads(json_msg)
method_name = payload["method"]
# Get the function gobject from the api
function_obj = self.api.get_obj(method_name)
if function_obj is None:
log.error(f"Method '{method_name}' not found in api")
return
# Create an instance of the function gobject
fn_instance = function_obj()
@@ -102,7 +100,7 @@ class WebExecutor(GObject.Object):
# 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)
arg_class = self.plain_api.get_method_argtype(method_name, k)
if dataclasses.is_dataclass(arg_class):
reconciled_arguments[k] = from_dict(arg_class, v)
else:
@@ -114,7 +112,6 @@ class WebExecutor(GObject.Object):
op_key,
)
def on_result(self, source: ImplFunc, data: GResult) -> None:
result = dict()
result["result"] = dataclass_to_dict(data.result)

7
pkgs/clan-app/clan_app/windows/main_window.py
View File

@@ -36,12 +36,9 @@ class MainWindow(Adw.ApplicationWindow):
stack_view = ViewStack.use().view
webexec = WebExecutor(plain_api=API, content_uri=config.content_uri)
webview = WebExecutor(
abstr_methods=API._orig_annotations, content_uri=config.content_uri
)
stack_view.add_named(webview.get_webview(), "webview")
stack_view.add_named(webexec.get_webview(), "webview")
stack_view.set_visible_child_name(config.initial_view)
view.set_content(stack_view)

3
pkgs/clan-app/pyproject.toml
View File

@@ -37,9 +37,10 @@ disallow_untyped_defs = true
no_implicit_optional = true
[[tool.mypy.overrides]]
module = "clan_cli.*"
module = "argcomplete.*"
ignore_missing_imports = true
[tool.ruff]
target-version = "py311"
line-length = 88

21
pkgs/clan-cli/clan_cli/api/__init__.py
View File

@@ -50,10 +50,22 @@ def update_wrapper_signature(wrapper: Callable, wrapped: Callable) -> None:
wrapper.__signature__ = new_sig # type: ignore
class _MethodRegistry:
class MethodRegistry:
def __init__(self) -> None:
self._orig_annotations: dict[str, Callable[[Any], Any]] = {}
self._registry: dict[str, Callable[[Any], Any]] = {}
self._orig_annotations: dict[str, dict[str, Any]] = {}
self._registry: dict[str, Callable[..., Any]] = {}
@property
def annotations(self) -> dict[str, dict[str, Any]]:
return self._orig_annotations
@property
def functions(self) -> dict[str, Callable[..., Any]]:
return self._registry
def reset(self) -> None:
self._orig_annotations.clear()
self._registry.clear()
def register_abstract(self, fn: Callable[..., T]) -> Callable[..., T]:
@wraps(fn)
@@ -84,7 +96,6 @@ API.register(open_file)
return fn
def register(self, fn: Callable[..., T]) -> Callable[..., T]:
if fn.__name__ in self._registry:
raise ValueError(f"Function {fn.__name__} already registered")
if fn.__name__ in self._orig_annotations:
@@ -189,4 +200,4 @@ API.register(open_file)
return None
API = _MethodRegistry()
API = MethodRegistry()

2
pkgs/clan-cli/tests/test_api_dataclass_compat.py
View File

@@ -5,6 +5,7 @@ import sys
from dataclasses import is_dataclass
from pathlib import Path
from clan_cli.api import API
from clan_cli.api.util import JSchemaTypeError, type_to_dict
from clan_cli.errors import ClanError
@@ -121,6 +122,7 @@ def test_all_dataclasses() -> None:
for file, dataclass in dataclasses:
print(f"checking dataclass {dataclass} in file: {file}")
try:
API.reset()
dclass = load_dataclass_from_file(file, dataclass, str(cli_path.parent))
type_to_dict(dclass)
except JSchemaTypeError as e: