add api + cli Interface

src/interface/api.py

@@ -0,0 +1,162 @@
+from __future__ import annotations
+
+import inspect
+import types
+import fastapi
+import pydantic
+import uvicorn
+
+from typing import Optional, Union
+
+
+def snake_to_camel_case(snake: str) -> str:
+    return "".join(word.title() for word in snake.split("_"))
+
+
+def get_path_param(route: str) -> Optional[str]:
+    last_path = route.split("/")[-1]
+    if last_path and "{" in last_path:
+        return last_path.strip("{}")
+    return None
+
+
+def alter_params_for_body(
+    parameters, as_body, as_body_except
+) -> list[Union[list[inspect.Parameter], inspect.Parameter]]:
+    if as_body_except:
+        body_params = []
+        rest = []
+        for param in parameters:
+            if param.name not in as_body_except:
+                body_params.append(param)
+            else:
+                rest.append(param)
+        return [body_params, *rest]
+
+    if as_body:
+        return [parameters]
+
+    return parameters
+
+
+def params_to_body(
+    params: list[inspect.Parameter], func_name: str
+) -> inspect.Parameter:
+    model = pydantic.create_model(
+        snake_to_camel_case(func_name),
+        **{
+            param.name: (
+                param.annotation,
+                param.default if param.default != param.empty else ...,
+            )
+            for param in params
+        },
+    )
+
+    return inspect.Parameter(
+        func_name.split("_")[0],
+        inspect.Parameter.POSITIONAL_ONLY,
+        default=...,
+        annotation=model,
+    )
+
+
+class Interface:
+    type = "api"
+
+    instance: Union[fastapi.FastAPI, fastapi.APIRouter]
+    name: str
+
+    def __init__(self, root: bool = False, name: Optional[str] = None):
+        self.instance = fastapi.FastAPI() if root else fastapi.APIRouter()
+        self.name = "root" if root else name or ""
+
+    def add(self, interface: Interface):
+        assert isinstance(interface.instance, fastapi.APIRouter)
+        self.instance.include_router(
+            interface.instance, prefix=f"/{interface.name}", tags=[interface.name]
+        )
+
+    def run(self, host="127.0.0.1", port=6787, debug=False, reload=False):
+        uvicorn.run(
+            self.instance,
+            host=host,
+            port=port,
+            log_level="info" if not debug else "debug",
+            reload=reload,
+            root_path="/yunohost/api",
+        )
+
+    def cli(self, *args, **kwargs):
+        def decorator(func):
+            return func
+
+        return decorator
+
+    def api(
+        self,
+        route: str,
+        method: str = "get",
+        as_body: bool = False,
+        as_body_except: Optional[list[str]] = None,
+        **kwargs,
+    ):
+        as_body = as_body if not as_body_except else True
+
+        def decorator(func):
+            signature = inspect.signature(func)
+            override_params = []
+            params = alter_params_for_body(
+                signature.parameters.values(), as_body, as_body_except
+            )
+            path_param = get_path_param(route)
+
+            for param in params:
+                if isinstance(param, list):
+                    override_params.append(params_to_body(param, func.__name__))
+                else:
+                    default_kwargs = kwargs.get(param.name, {})
+                    default_cls = (
+                        fastapi.Path if param.name == path_param else fastapi.Query
+                    )
+
+                    if param.default is None:
+                        default_value = default_cls(None, **default_kwargs)
+                    elif param.default is param.empty:
+                        default_value = default_cls(..., **default_kwargs)
+                    else:
+                        default_value = default_cls(param.default, **default_kwargs)
+
+                    override_params.append(param.replace(default=default_value))
+
+            route_func = getattr(self.instance, method)(route)
+            override_signature = signature.replace(parameters=tuple(override_params))
+
+            if as_body:
+
+                def body_to_args_back(*args, **kwargs):
+                    new_kwargs = {}
+                    for kwarg, value in kwargs.items():
+                        if issubclass(type(value), pydantic.BaseModel):
+                            new_kwargs = value.dict() | new_kwargs
+                        else:
+                            new_kwargs[kwarg] = value
+                    return func(*args, **new_kwargs)
+
+                body_to_args_back.__name__ = func.__name__
+                body_to_args_back.__signature__ = override_signature
+                route_func(body_to_args_back)
+            else:
+                func_copy = types.FunctionType(
+                    func.__code__,
+                    func.__globals__,
+                    func.__name__,
+                    func.__defaults__,
+                    func.__closure__,
+                )
+                func_copy.__signature__ = override_signature
+                route_func(func_copy)
+
+            return func
+
+        return decorator

src/interface/cli.py

@@ -0,0 +1,84 @@
+from __future__ import annotations
+
+import inspect
+import typer
+import yaml
+
+from typing import Any, Optional
+from rich import print as rprint
+from rich.syntax import Syntax
+
+
+def parse_cli_command(command: str) -> tuple[str, list[str]]:
+    command, *args = command.split(" ")
+    return (command, [arg.strip("{}") for arg in args])
+
+
+def print_as_yaml(data: Any):
+    data = yaml.dump(data, default_flow_style=False)
+    rprint(Syntax(data, "yaml", background_color="default"))
+
+
+class Interface:
+    type = "cli"
+
+    instance: typer.Typer
+    name: str
+
+    def __init__(self, root: bool = False, name: Optional[str] = None):
+        self.instance = typer.Typer()
+        self.name = "root" if root else name or ""
+
+    def add(self, interface: Interface):
+        self.instance.add_typer(interface.instance, name=interface.name)
+
+    def run(self):
+        self.instance()
+
+    def cli(self, command_def: str, **kwargs):
+        def decorator(func):
+            signature = inspect.signature(func)
+            override_params = []
+            command, args = parse_cli_command(command_def)
+
+            for param in signature.parameters.values():
+
+                # Auto setup typer Argument or Option kwargs
+                default_kwargs = kwargs.get(param.name, {})
+                # if param.name not in args and not default_kwargs.get("hidden", False):
+                #     default_kwargs["prompt"] = True
+                if param.name == "password":
+                    default_kwargs["confirmation_prompt"] = True
+                    default_kwargs["hide_input"] = True
+
+                # Define new default value for typer
+                default_cls = typer.Argument if param.name in args else typer.Option
+                if param.default is None:
+                    default_value = default_cls(None, **default_kwargs)
+                elif param.default is param.empty:
+                    default_value = default_cls(..., **default_kwargs)
+                else:
+                    default_value = default_cls(param.default, **default_kwargs)
+
+                override_params.append(param.replace(default=default_value))
+
+            def hook_results(*args, **kwargs):
+                results = func(*args, **kwargs)
+                print_as_yaml(results)
+                return results
+
+            hook_results.__name__ = func.__name__
+            hook_results.__signature__ = signature.replace(
+                parameters=tuple(override_params)
+            )
+            self.instance.command(command)(hook_results)
+
+            return func
+
+        return decorator
+
+    def api(self, *args, **kwargs):
+        def decorator(func):
+            return func
+
+        return decorator