From be10d99015dbfff82ec8078895b193e1c921e41d Mon Sep 17 00:00:00 2001 From: Qubasa Date: Tue, 8 Jul 2025 15:45:40 +0700 Subject: [PATCH] clan-app: Add plug and play middleware interface --- .../clan_app/deps/webview/api_bridge.py | 86 ++++++++ .../clan_app/deps/webview/middleware.py | 201 ++++++++++++++++++ .../clan_app/deps/webview/webview_bridge.py | 105 +++++++++ 3 files changed, 392 insertions(+) create mode 100644 pkgs/clan-app/clan_app/deps/webview/api_bridge.py create mode 100644 pkgs/clan-app/clan_app/deps/webview/middleware.py create mode 100644 pkgs/clan-app/clan_app/deps/webview/webview_bridge.py diff --git a/pkgs/clan-app/clan_app/deps/webview/api_bridge.py b/pkgs/clan-app/clan_app/deps/webview/api_bridge.py new file mode 100644 index 000000000..e9479edc7 --- /dev/null +++ b/pkgs/clan-app/clan_app/deps/webview/api_bridge.py @@ -0,0 +1,86 @@ +import logging +from abc import ABC, abstractmethod +from contextlib import ExitStack +from dataclasses import dataclass +from typing import TYPE_CHECKING, Any + +if TYPE_CHECKING: + from .middleware import Middleware + +log = logging.getLogger(__name__) + + +@dataclass(frozen=True) +class ApiRequest: + method_name: str + args: dict[str, Any] + header: dict[str, Any] + op_key: str + + +@dataclass(frozen=True) +class ApiResponse: + op_key: str + success: bool + data: Any + error: str | None = None + + +@dataclass +class ApiBridge(ABC): + """Generic interface for API bridges that can handle method calls from different sources.""" + + middleware_chain: tuple["Middleware", ...] + + @abstractmethod + def send_response(self, response: ApiResponse) -> None: + """Send response back to the client.""" + + def process_request(self, request: ApiRequest) -> None: + """Process an API request through the middleware chain.""" + from .middleware import MiddlewareContext + + with ExitStack() as stack: + context = MiddlewareContext( + request=request, + bridge=self, + exit_stack=stack, + ) + + # Process through middleware chain + for middleware in self.middleware_chain: + try: + log.debug( + f"{middleware.__class__.__name__} => {request.method_name}" + ) + middleware.process(context) + except Exception as e: + # If middleware fails, handle error + self.send_error_response( + request.op_key, str(e), ["middleware_error"] + ) + return + + def send_error_response( + self, op_key: str, error_message: str, location: list[str] + ) -> None: + """Send an error response.""" + from clan_lib.api import ApiError, ErrorDataClass + + error_data = ErrorDataClass( + op_key=op_key, + status="error", + errors=[ + ApiError( + message="An internal error occured", + description=error_message, + location=location, + ) + ], + ) + + response = ApiResponse( + op_key=op_key, success=False, data=error_data, error=error_message + ) + + self.send_response(response) diff --git a/pkgs/clan-app/clan_app/deps/webview/middleware.py b/pkgs/clan-app/clan_app/deps/webview/middleware.py new file mode 100644 index 000000000..3a5832a0a --- /dev/null +++ b/pkgs/clan-app/clan_app/deps/webview/middleware.py @@ -0,0 +1,201 @@ +import io +import json +import logging +from abc import ABC, abstractmethod +from contextlib import ExitStack +from dataclasses import dataclass +from typing import TYPE_CHECKING, Any, ContextManager + +from clan_lib.api import MethodRegistry, dataclass_to_dict, from_dict +from clan_lib.async_run import AsyncContext, get_async_ctx, set_async_ctx +from clan_lib.custom_logger import setup_logging +from clan_lib.log_manager import LogManager + +from .api_bridge import ApiRequest, ApiResponse + +if TYPE_CHECKING: + from .api_bridge import ApiBridge, ApiRequest + +log = logging.getLogger(__name__) + + +@dataclass +class MiddlewareContext: + request: "ApiRequest" + bridge: "ApiBridge" + exit_stack: ExitStack + + +@dataclass(frozen=True) +class Middleware(ABC): + """Abstract base class for middleware components.""" + + @abstractmethod + def process(self, context: MiddlewareContext) -> None: + """Process the request through this middleware.""" + + def register_context_manager( + self, context: MiddlewareContext, cm: ContextManager[Any] + ) -> Any: + """Register a context manager with the exit stack.""" + return context.exit_stack.enter_context(cm) + + +@dataclass(frozen=True) +class ArgumentParsingMiddleware(Middleware): + """Middleware that handles argument parsing and dataclass construction.""" + + api: MethodRegistry + + def process(self, context: MiddlewareContext) -> None: + try: + # Convert dictionary arguments to dataclass instances + reconciled_arguments = {} + for k, v in context.request.args.items(): + if k == "op_key": + continue + + # Get the expected argument type from the API + arg_class = self.api.get_method_argtype(context.request.method_name, k) + + # Convert dictionary to dataclass instance + reconciled_arguments[k] = from_dict(arg_class, v) + + # Add op_key to arguments + reconciled_arguments["op_key"] = context.request.op_key + + # Create a new request with reconciled arguments + + updated_request = ApiRequest( + method_name=context.request.method_name, + args=reconciled_arguments, + header=context.request.header, + op_key=context.request.op_key, + ) + context.request = updated_request + + except Exception as e: + log.exception( + f"Error while parsing arguments for {context.request.method_name}" + ) + context.bridge.send_error_response( + context.request.op_key, + str(e), + ["argument_parsing", context.request.method_name], + ) + raise + + +@dataclass(frozen=True) +class LoggingMiddleware(Middleware): + """Middleware that sets up logging context without executing methods.""" + + log_manager: LogManager + + def process(self, context: MiddlewareContext) -> None: + method = context.request.method_name + + try: + # Handle log group configuration + log_group: list[str] | None = context.request.header.get("logging", {}).get( + "group_path", None + ) + if log_group is not None: + if not isinstance(log_group, list): + msg = f"Expected log_group to be a list, got {type(log_group)}" + raise TypeError(msg) + log.warning( + f"Using log group {log_group} for {context.request.method_name} with op_key {context.request.op_key}" + ) + # Create log file + log_file = self.log_manager.create_log_file( + method, op_key=context.request.op_key, group_path=log_group + ).get_file_path() + + except Exception as e: + log.exception( + f"Error while handling request header of {context.request.method_name}" + ) + context.bridge.send_error_response( + context.request.op_key, + str(e), + ["header_middleware", context.request.method_name], + ) + return + + # Register logging context manager + class LoggingContextManager: + def __init__(self, log_file) -> None: + self.log_file = log_file + self.log_f = None + self.handler = None + self.original_ctx = None + + def __enter__(self): + self.log_f = self.log_file.open("ab") + self.original_ctx = get_async_ctx() + + # Set up async context for logging + ctx = AsyncContext(**self.original_ctx.__dict__) + ctx.stderr = self.log_f + ctx.stdout = self.log_f + set_async_ctx(ctx) + + # Set up logging handler + handler_stream = io.TextIOWrapper( + self.log_f, + encoding="utf-8", + write_through=True, + line_buffering=True, + ) + self.handler = setup_logging( + log.getEffectiveLevel(), log_file=handler_stream + ) + + return self + + def __exit__(self, exc_type, exc_val, exc_tb): + if self.handler: + self.handler.root_logger.removeHandler(self.handler.new_handler) + self.handler.new_handler.close() + if self.log_f: + self.log_f.close() + if self.original_ctx: + set_async_ctx(self.original_ctx) + + # Register the logging context manager + self.register_context_manager(context, LoggingContextManager(log_file)) + + +@dataclass(frozen=True) +class MethodExecutionMiddleware(Middleware): + """Middleware that handles actual method execution.""" + + api: MethodRegistry + + def process(self, context: MiddlewareContext) -> None: + method = self.api.functions[context.request.method_name] + + try: + # Execute the actual method + result = method(**context.request.args) + wrapped_result = {"body": dataclass_to_dict(result), "header": {}} + + log.debug( + f"Result for {context.request.method_name}: {json.dumps(dataclass_to_dict(wrapped_result), indent=4)}" + ) + + response = ApiResponse( + op_key=context.request.op_key, success=True, data=wrapped_result + ) + context.bridge.send_response(response) + + except Exception as e: + log.exception( + f"Error while handling result of {context.request.method_name}" + ) + context.bridge.send_error_response( + context.request.op_key, + str(e), + ["method_execution", context.request.method_name], + ) diff --git a/pkgs/clan-app/clan_app/deps/webview/webview_bridge.py b/pkgs/clan-app/clan_app/deps/webview/webview_bridge.py new file mode 100644 index 000000000..8cef32a76 --- /dev/null +++ b/pkgs/clan-app/clan_app/deps/webview/webview_bridge.py @@ -0,0 +1,105 @@ +import json +import logging +import threading +from dataclasses import dataclass, field +from typing import TYPE_CHECKING + +from clan_lib.api import dataclass_to_dict +from clan_lib.api.tasks import WebThread +from clan_lib.async_run import set_should_cancel + +from .api_bridge import ApiBridge, ApiRequest, ApiResponse +from .webview import FuncStatus + +if TYPE_CHECKING: + from .webview import Webview + +log = logging.getLogger(__name__) + + +@dataclass +class WebviewBridge(ApiBridge): + """Webview-specific implementation of the API bridge.""" + + webview: "Webview" + threads: dict[str, WebThread] = field(default_factory=dict) + + def send_response(self, response: ApiResponse) -> None: + """Send response back to the webview client.""" + + if response.success: + serialized = json.dumps( + dataclass_to_dict(response.data), indent=4, ensure_ascii=False + ) + status = FuncStatus.SUCCESS + else: + serialized = json.dumps( + dataclass_to_dict(response.data), indent=4, ensure_ascii=False + ) + status = FuncStatus.SUCCESS # Even errors are sent as SUCCESS to webview + + log.debug( + f"Sending response for op_key {response.op_key} with status {status.name} and data: {serialized}" + ) + self.webview.return_(response.op_key, status, serialized) + + def handle_webview_call( + self, + method_name: str, + op_key_bytes: bytes, + request_data: bytes, + arg: int, + ) -> None: + """Handle a call from webview's JavaScript bridge.""" + op_key = op_key_bytes.decode() + raw_args = json.loads(request_data.decode()) + + try: + # Parse the webview-specific request format + header = {} + args = {} + + if len(raw_args) == 1: + request = raw_args[0] + header = request.get("header", {}) + if not isinstance(header, dict): + msg = f"Expected header to be a dict, got {type(header)}" + raise TypeError(msg) + + body = request.get("body", {}) + if not isinstance(body, dict): + msg = f"Expected body to be a dict, got {type(body)}" + raise TypeError(msg) + + args = body + elif len(raw_args) > 1: + msg = "Expected a single argument, got multiple arguments" + raise ValueError(msg) + + # Create API request + api_request = ApiRequest( + method_name=method_name, args=args, header=header, op_key=op_key + ) + + except Exception as e: + self.send_error_response(op_key, str(e), ["webview_bridge", method_name]) + return + + # Process in a separate thread + def thread_task(stop_event: threading.Event) -> None: + set_should_cancel(lambda: stop_event.is_set()) + + try: + log.debug( + f"Calling {method_name}({json.dumps(api_request.args, indent=4)}) with header {json.dumps(api_request.header, indent=4)} and op_key {op_key}" + ) + self.process_request(api_request) + finally: + self.threads.pop(op_key, None) + + stop_event = threading.Event() + thread = threading.Thread( + target=thread_task, args=(stop_event,), name="WebviewThread" + ) + thread.start() + self.threads[op_key] = WebThread(thread=thread, stop_event=stop_event)