API: init op_key, improve seralisation & signature typing
This commit is contained in:
parent
a89fd31844
commit
cb847cab82
|
@ -38,8 +38,10 @@ def dataclass_to_dict(obj: Any) -> Any:
|
|||
return [dataclass_to_dict(item) for item in obj]
|
||||
elif isinstance(obj, dict):
|
||||
return {k: dataclass_to_dict(v) for k, v in obj.items()}
|
||||
else:
|
||||
elif isinstance(obj, Path):
|
||||
return str(obj)
|
||||
else:
|
||||
return obj
|
||||
|
||||
|
||||
# Implement the abstract open_file function
|
||||
|
@ -265,6 +267,7 @@ class WebView:
|
|||
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.
|
||||
|
@ -276,8 +279,13 @@ class WebView:
|
|||
else:
|
||||
reconciled_arguments[k] = v
|
||||
|
||||
result = handler_fn(**reconciled_arguments)
|
||||
serialized = json.dumps(dataclass_to_dict(result))
|
||||
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)
|
||||
|
|
|
@ -1,6 +1,7 @@
|
|||
from collections.abc import Callable
|
||||
from dataclasses import dataclass
|
||||
from functools import wraps
|
||||
from inspect import Parameter, signature
|
||||
from typing import Annotated, Any, Generic, Literal, TypeVar, get_type_hints
|
||||
|
||||
from clan_cli.errors import ClanError
|
||||
|
@ -21,17 +22,34 @@ class ApiError:
|
|||
class SuccessDataClass(Generic[ResponseDataType]):
|
||||
status: Annotated[Literal["success"], "The status of the response."]
|
||||
data: ResponseDataType
|
||||
op_key: str | None
|
||||
|
||||
|
||||
@dataclass
|
||||
class ErrorDataClass:
|
||||
status: Literal["error"]
|
||||
errors: list[ApiError]
|
||||
op_key: str | None
|
||||
|
||||
|
||||
ApiResponse = SuccessDataClass[ResponseDataType] | ErrorDataClass
|
||||
|
||||
|
||||
def update_wrapper_signature(wrapper: Callable, wrapped: Callable) -> None:
|
||||
sig = signature(wrapped)
|
||||
params = list(sig.parameters.values())
|
||||
|
||||
# Add 'op_key' parameter
|
||||
op_key_param = Parameter(
|
||||
"op_key", Parameter.KEYWORD_ONLY, default=None, annotation=str | None
|
||||
)
|
||||
params.append(op_key_param)
|
||||
|
||||
# Create a new signature
|
||||
new_sig = sig.replace(parameters=params)
|
||||
wrapper.__signature__ = new_sig # type: ignore
|
||||
|
||||
|
||||
class _MethodRegistry:
|
||||
def __init__(self) -> None:
|
||||
self._orig: dict[str, Callable[[Any], Any]] = {}
|
||||
|
@ -41,13 +59,16 @@ class _MethodRegistry:
|
|||
self._orig[fn.__name__] = fn
|
||||
|
||||
@wraps(fn)
|
||||
def wrapper(*args: Any, **kwargs: Any) -> ApiResponse[T]:
|
||||
def wrapper(
|
||||
*args: Any, op_key: str | None = None, **kwargs: Any
|
||||
) -> ApiResponse[T]:
|
||||
try:
|
||||
data: T = fn(*args, **kwargs)
|
||||
return SuccessDataClass(status="success", data=data)
|
||||
return SuccessDataClass(status="success", data=data, op_key=op_key)
|
||||
except ClanError as e:
|
||||
return ErrorDataClass(
|
||||
status="error",
|
||||
op_key=op_key,
|
||||
errors=[
|
||||
ApiError(
|
||||
message=e.msg,
|
||||
|
@ -63,6 +84,11 @@ class _MethodRegistry:
|
|||
orig_return_type = get_type_hints(fn).get("return")
|
||||
wrapper.__annotations__["return"] = ApiResponse[orig_return_type] # type: ignore
|
||||
|
||||
# Add additional argument for the operation key
|
||||
wrapper.__annotations__["op_key"] = str | None # type: ignore
|
||||
|
||||
update_wrapper_signature(wrapper, fn)
|
||||
|
||||
self._registry[fn.__name__] = wrapper
|
||||
return fn
|
||||
|
||||
|
@ -91,6 +117,12 @@ class _MethodRegistry:
|
|||
|
||||
return_type = serialized_hints.pop("return")
|
||||
|
||||
sig = signature(func)
|
||||
required_args = []
|
||||
for n, param in sig.parameters.items():
|
||||
if param.default == Parameter.empty:
|
||||
required_args.append(n)
|
||||
|
||||
api_schema["properties"][name] = {
|
||||
"type": "object",
|
||||
"required": ["arguments", "return"],
|
||||
|
@ -99,7 +131,7 @@ class _MethodRegistry:
|
|||
"return": return_type,
|
||||
"arguments": {
|
||||
"type": "object",
|
||||
"required": [k for k in serialized_hints.keys()],
|
||||
"required": required_args,
|
||||
"additionalProperties": False,
|
||||
"properties": serialized_hints,
|
||||
},
|
||||
|
|
Loading…
Reference in New Issue
Block a user