CliHelper facilitates create commands that require a class instance and command on its method. It uses Typer as underlying CLI library. Typer provides infer arguments & options from Python Typing Annotaion. However, there is no good way to get rid of the object initialization and self instance pass. It will be redundant to write all arguments tw…

cli_helper.py

import functools
import inspect
from collections import OrderedDict, defaultdict
from collections.abc import Mapping
from inspect import Parameter
from typing import Callable, get_type_hints

import typer


def debug_anno(f):
    print(f.__name__)
    print("__annotations__:", f.__annotations__)

    # sigature is not affeted by __annotations__
    # CPython implementation detail: If the passed object has a __signature__ attribute, we may use it to create the signature. The exact semantics are an implementation detail and are subject to unannounced changes.
    # https://docs.python.org/3/library/inspect.html#inspect.signature
    print("sigature:", inspect.signature(f, eval_str=True))
    # __signature__ by default doesn't exist, but can be set manually
    print("__signature__:", getattr(f, "__signature__", None))

    # got from __annotations__ instead of signature
    print("type hints:", get_type_hints(f))
    print()

    # typer.completion is using inspect.signature to confirm the required arguments and options, and using get_type_hints to confirm the type


def requireParamsFrom[T, **P](x: Callable[P, T]):
    def decorator[V](f: Callable[..., V]) -> Callable[P, V]:
        # debug_anno(x)
        # debug_anno(f)

        # set annotation
        x_annotation = (x.__init__ if inspect.isclass(x) else x).__annotations__
        f_annotations = f.__annotations__.copy()
        f_annotations.update(x_annotation)
        f_annotations["return"] = f.__annotations__.get("return", None)
        f.__annotations__ = f_annotations

        # set signature
        x_sig = inspect.signature(x)
        x_sig_params = x_sig.parameters
        if (params_iter := iter(x_sig_params.items())) and next(params_iter)[
            0
        ] == "self":
            # skip the first self
            x_sig_params = OrderedDict(params_iter)

        f_sig = inspect.signature(f)
        f_sig_params = OrderedDict(
            [
                (name, p)
                for name, p in f_sig.parameters.items()
                if p.kind
                in (
                    p.POSITIONAL_ONLY,
                    p.POSITIONAL_OR_KEYWORD,
                    p.KEYWORD_ONLY,
                )  # ignore *args & **kwargs
            ]
        )
        f_sig_params = _update_params(f_sig_params, x_sig_params)
        # The signature parameters is required to ordered from postional then keywords
        f.__signature__ = f_sig.replace(parameters=list(f_sig_params.values()))

        # debug_anno(f)
        return f

    return decorator


def _update_params(
    params: OrderedDict[str, Parameter],
    from_params: Mapping[str, Parameter],
) -> OrderedDict[str, Parameter]:
    """Update params from from_params, as the order of positional -> keywords.

    - param with the same name will be overriden by the value in from_params
    - params is placed before from_params for the same Parameter Kind.
    """
    to_be_updated = defaultdict[int, list[Parameter]](list)

    for name, param in from_params.items():
        if name in params:
            params[name] = param
            continue
        to_be_updated[param.kind].append(param)

    res: list[Parameter] = []

    for kind in (
        Parameter.POSITIONAL_ONLY,
        Parameter.POSITIONAL_OR_KEYWORD,
        Parameter.KEYWORD_ONLY,
    ):  # ignore Parameter.VAR_KEYWORD and Parameter.VAR_POSITIONAL
        res.extend(p for p in params.values() if p.kind is kind)
        res.extend(to_be_updated[kind])

    return OrderedDict((p.name, p) for p in res)



class CliHelper:
    """CliHelper facilitates create commands that require a class instance and command on its method. It uses Typer as underlying CLI library.

    Typer provides infer arguments & options from Python Typing Annotaion. However, there is no good way to get rid of the object initialization and self instance pass. It will be redundant to write all arguments twice.

    # Example Usage
    # Create a class with method
    ---
    cli = CliHelper()

    cli.require_obj(VideoDurationRepairer)  # all intialization parameters will be taken as the common arguments and options for all commands.

    # the obj instance will be passed to the method
    cli.create_command(VideoDurationRepairer.check_and_fix_video, "fix")
    cli.create_command(VideoDurationRepairer.check_and_fix_videos, "fixall")

    cli.app() # run typer app
    """

    def __init__(self, app: typer.Typer | None = None):
        self.app = app or typer.Typer()
        self._obj_initializer: Callable | None = None
        self._wrapped_callback: Callable[[], Callable] | None = None

    def require_obj(self, custom_type: Callable) -> Callable:
        @requireParamsFrom(custom_type)
        def obj_initializer(ctx: typer.Context, *args, **kwargs):
            ctx.obj = custom_type(*args, **kwargs)

        self._obj_initializer = obj_initializer

        if self._wrapped_callback:
            return self._wrapped_callback()
        else:
            return self.app.callback()(obj_initializer)

    def create_command(self, f: Callable, name: str | None = None, **kwargs):
        @requireParamsFrom(f)
        def cmd(ctx: typer.Context, *args, **kwargs):
            f(ctx.obj, *args, **kwargs)

        return self.app.command(name=name or f.__name__, **kwargs)(cmd)

    def callback(
        self, *args, **kwargs
    ):  # TODO: copy annotaion from typer.Typer.callback
        """Has the same siginature as typer.Typer.callback. However, this callback method doesn't override original obj_initializer callback, but combine them togther."""

        def wrap(f: Callable):
            self._wrapped_callback = functools.partial(
                self.callback(*args, **kwargs), f
            )

            if not self._obj_initializer:
                return self.app.callback(*args, **kwargs)(f)

            obj_initializer = self._obj_initializer

            params = inspect.signature(f).parameters.copy()
            ctx_param = self._get_context_param(params)
            if ctx_param:
                params.pop(ctx_param.name)

            # check
            self._check_no_position_only_params(params)
            self._check_sig_params_conflict(
                inspect.signature(obj_initializer).parameters, params
            )

            @self.app.callback(*args, **kwargs)
            @requireParamsFrom(f)
            @requireParamsFrom(obj_initializer)
            def wrapped(ctx: typer.Context, *args, **kwargs):
                splited_kwargs = {
                    k: kwargs.pop(k) for k in list(kwargs.keys()) if k in params
                }
                if ctx_param:
                    # even there are two parameters requiring typer.Context, only one will be set
                    ctx = kwargs.pop(ctx_param.name, None) or ctx
                    splited_kwargs[ctx_param.name] = ctx

                obj_initializer(ctx, *args, **kwargs)

                f(**splited_kwargs)

            return wrapped

        return wrap

    def _check_sig_params_conflict(
        self,
        params1: Mapping[str, Parameter],
        params2: Mapping[str, Parameter],
    ):
        if intersects := set(params1.keys()).intersection(params2.keys()):
            raise Exception(
                "Callback has parameter conflicts with obj initializer (required object)",
                intersects,
            )

    def _check_no_position_only_params(self, params: Mapping[str, Parameter]):
        for param in params.values():
            if param.kind is Parameter.POSITIONAL_ONLY:
                raise Exception(
                    "POSITIONAL_ONLY parameter is not allowed in the callback when used together with object initializer",
                    param,
                )

    def _get_context_param(self, params: Mapping[str, Parameter]) -> Parameter | None:
        for param in params.values():
            if param.annotation is typer.Context:
                return param