diff --git a/src/kernel/__init__.py b/src/kernel/__init__.py index 1093761..10b14f7 100644 --- a/src/kernel/__init__.py +++ b/src/kernel/__init__.py @@ -26,6 +26,18 @@ ) from ._base_client import DefaultHttpxClient, DefaultAsyncHttpxClient from ._utils._logs import setup_logging as _setup_logging +from .app_framework import ( + App, + KernelApp, + KernelJson, + KernelAction, + KernelAppJson, + KernelContext, + KernelActionJson, + KernelAppRegistry, + app_registry, + export_registry, +) __all__ = [ "types", @@ -66,6 +78,16 @@ "DEFAULT_CONNECTION_LIMITS", "DefaultHttpxClient", "DefaultAsyncHttpxClient", + "KernelContext", + "KernelAction", + "KernelActionJson", + "KernelAppJson", + "KernelJson", + "KernelApp", + "KernelAppRegistry", + "App", + "app_registry", + "export_registry", ] _setup_logging() diff --git a/src/kernel/app_framework.py b/src/kernel/app_framework.py new file mode 100644 index 0000000..efad54f --- /dev/null +++ b/src/kernel/app_framework.py @@ -0,0 +1,155 @@ +import json +import inspect +import functools +from typing import Any, Dict, List, Union, TypeVar, Callable, Optional +from dataclasses import dataclass + +T = TypeVar("T") + +# Context definition +@dataclass +class KernelContext: + """Context object passed to action handlers""" + invocation_id: str + +# Action definition +@dataclass +class KernelAction: + """Action that can be invoked on a Kernel app""" + name: str + handler: Callable[..., Any] + +# JSON interfaces +@dataclass +class KernelActionJson: + """JSON representation of a Kernel action""" + name: str + +@dataclass +class KernelAppJson: + """JSON representation of a Kernel app""" + name: str + actions: List[KernelActionJson] + +@dataclass +class KernelJson: + """JSON representation of Kernel manifest""" + apps: List[KernelAppJson] + entrypoint: str + +# App class +class KernelApp: + def __init__(self, name: str): + self.name = name + self.actions: Dict[str, KernelAction] = {} + # Register this app in the global registry + _app_registry.register_app(self) + + def action(self, name_or_handler: Optional[Union[str, Callable[..., Any]]] = None) -> Callable[..., Any]: + """Decorator to register an action with the app""" + if name_or_handler is None: + # This is the @app.action() case, which should return the decorator + def decorator(f: Callable[..., Any]) -> Callable[..., Any]: + return self._register_action(f.__name__, f) + return decorator + elif callable(name_or_handler): + # This is the @app.action case (handler passed directly) + return self._register_action(name_or_handler.__name__, name_or_handler) + else: + # This is the @app.action("name") case (name_or_handler is a string) + def decorator(f: Callable[..., Any]) -> Callable[..., Any]: + return self._register_action(name_or_handler, f) # name_or_handler is the name string here + return decorator + + def _register_action(self, name: str, handler: Callable[..., Any]) -> Callable[..., Any]: + """Internal method to register an action""" + + @functools.wraps(handler) + def wrapper(*args: Any, **kwargs: Any) -> Any: + # Determine if the original handler accepts context as first argument + sig = inspect.signature(handler) + param_names = list(sig.parameters.keys()) + param_count = len(param_names) + + if param_count == 1: + actual_input = None + # The handler only takes input + if len(args) > 0: # Prioritize args if context was implicitly passed + # If context (args[0]) and input (args[1]) were provided, or just input (args[0]) + actual_input = args[1] if len(args) > 1 else args[0] + elif kwargs: + # Attempt to find the single expected kwarg + if param_names: # Should always be true if param_count == 1 + param_name = param_names[0] + if param_name in kwargs: + actual_input = kwargs[param_name] + elif kwargs: # Fallback if name doesn't match but kwargs exist + actual_input = next(iter(kwargs.values())) + elif kwargs: # param_names is empty but kwargs exist (unlikely for param_count==1) + actual_input = next(iter(kwargs.values())) + # If no args/kwargs, actual_input remains None, handler might raise error or accept None + return handler(actual_input) + else: # param_count == 0 or param_count > 1 + # Handler takes context and input (or more), or no args + return handler(*args, **kwargs) + + action = KernelAction(name=name, handler=wrapper) + self.actions[name] = action + return wrapper + + def get_actions(self) -> List[KernelAction]: + """Get all actions for this app""" + return list(self.actions.values()) + + def get_action(self, name: str) -> Optional[KernelAction]: + """Get an action by name""" + return self.actions.get(name) + + def to_dict(self) -> KernelAppJson: + """Export app information without handlers""" + return KernelAppJson( + name=self.name, + actions=[KernelActionJson(name=action.name) for action in self.get_actions()] + ) + + +# Registry for storing Kernel apps +class KernelAppRegistry: + def __init__(self) -> None: + self.apps: Dict[str, KernelApp] = {} + + def register_app(self, app: KernelApp) -> None: + self.apps[app.name] = app + + def get_apps(self) -> List[KernelApp]: + return list(self.apps.values()) + + def get_app_by_name(self, name: str) -> Optional[KernelApp]: + return self.apps.get(name) + + def export(self, entrypoint_relpath: str) -> KernelJson: + """Export the registry as a KernelJson object""" + apps = [app.to_dict() for app in self.get_apps()] + return KernelJson(apps=apps, entrypoint=entrypoint_relpath) + + def export_json(self, entrypoint_relpath: str) -> str: + """Export the registry as JSON""" + kernel_json = self.export(entrypoint_relpath) + return json.dumps(kernel_json.__dict__, indent=2) + + +# Create singleton registry for apps +_app_registry = KernelAppRegistry() + +# Create a simple function for creating apps +def App(name: str) -> KernelApp: + """Create a new Kernel app""" + return KernelApp(name) + +# Export the app registry for boot loader +app_registry = _app_registry + +# Function to export registry as JSON +def export_registry(entrypoint_relpath: str) -> str: + """Export the registry as JSON""" + return _app_registry.export_json(entrypoint_relpath)