Coverage for collectra/cli.py: 96.91%
97 statements
« prev ^ index » next coverage.py v7.16.2, created at 2026-10-16 09:56 +0000
1"""Generic, recursive Cappa adapter for runtime Python objects."""
3from __future__ import annotations
5import inspect
6from dataclasses import field, fields, make_dataclass, replace
7from typing import (
8 Annotated,
9 Any,
10 Callable,
11 Mapping,
12 get_args,
13 get_origin,
14 get_type_hints,
15)
17import cappa
19CLI_KIND = "__collectra_cli_kind__"
20CLI_NAME = "__collectra_cli_name__"
23def _member(kind: str, name: str | None = None):
24 def decorate(func: Callable[..., Any]) -> Callable[..., Any]:
25 setattr(func, CLI_KIND, kind)
26 setattr(func, CLI_NAME, name)
27 return func
29 return decorate
32def command(func: Callable[..., Any] | None = None, *, name: str | None = None):
33 """Mark a method as a terminal CLI command."""
34 decorator = _member("command", name)
35 return decorator(func) if func is not None else decorator
38def group(
39 func: Callable[..., Mapping[str, Any]] | None = None, *, name: str | None = None
40):
41 """Mark a method as a mapping of named child CLI objects."""
42 decorator = _member("group", name)
43 return decorator(func) if func is not None else decorator
46def _cli_members(obj: Any):
47 for member_name in dir(obj):
48 # Properties can load models or optional backends. Only bind members
49 # after confirming that their static definition is a CLI command.
50 definition = inspect.getattr_static(obj, member_name)
51 if isinstance(definition, (staticmethod, classmethod)):
52 definition = definition.__func__
53 if not getattr(definition, CLI_KIND, None):
54 continue
55 member = getattr(obj, member_name)
56 kind = getattr(member, CLI_KIND, None)
57 if callable(member) and kind:
58 cli_name = getattr(member, CLI_NAME, None) or member_name
59 yield cli_name.replace("_", "-"), kind, member
62def _callable_schema(name: str, func: Callable[..., Any]) -> type:
63 signature = inspect.signature(func)
64 hints = get_type_hints(func, include_extras=True)
65 schema_fields = []
67 for parameter in signature.parameters.values():
68 if parameter.kind is inspect.Parameter.VAR_KEYWORD:
69 raise TypeError(
70 f"CLI command {func.__qualname__} cannot expose **{parameter.name}; "
71 "declare typed parameters instead"
72 )
74 annotation = hints.get(parameter.name, str)
75 if parameter.kind is inspect.Parameter.VAR_POSITIONAL:
76 annotation = list[annotation]
77 schema_fields.append(
78 (parameter.name, annotation, field(default_factory=list))
79 )
80 elif parameter.default is inspect.Parameter.empty:
81 schema_fields.append((parameter.name, annotation))
82 else:
83 base_type = annotation
84 metadata = []
85 if get_origin(annotation) is Annotated:
86 base_type, *metadata = get_args(annotation)
87 if base_type is bool:
88 cli_name = parameter.name.replace("_", "-")
89 long = f"--{cli_name}/--no-{cli_name}"
90 else:
91 long = True
92 if any(isinstance(item, cappa.Arg) for item in metadata):
93 metadata = [
94 (
95 replace(item, long=long)
96 if isinstance(item, cappa.Arg)
97 and not item.long
98 and not item.short
99 else item
100 )
101 for item in metadata
102 ]
103 else:
104 metadata.append(cappa.Arg(long=long))
105 option = Annotated[(base_type, *metadata)]
106 schema_fields.append((parameter.name, option, parameter.default))
108 return make_dataclass(f"{name.title().replace('-', '')}Arguments", schema_fields)
111def command_for_callable(name: str, func: Callable[..., Any]) -> cappa.Command:
112 """Build a terminal Cappa command from a bound callable."""
113 schema = _callable_schema(name, func)
114 signature = inspect.signature(func)
116 def invoke(parsed):
117 values = {item.name: getattr(parsed, item.name) for item in fields(parsed)}
118 positional = []
119 keyword = {}
120 for parameter in signature.parameters.values():
121 value = values[parameter.name]
122 if parameter.kind is inspect.Parameter.VAR_POSITIONAL:
123 positional.extend(value)
124 elif parameter.kind is inspect.Parameter.POSITIONAL_ONLY:
125 positional.append(value)
126 else:
127 keyword[parameter.name] = value
128 return func(*positional, **keyword)
130 invoke.__annotations__ = {"parsed": schema}
131 invoke.__name__ = f"invoke_{name.replace('-', '_')}"
133 return cappa.Command(
134 schema,
135 name=name,
136 help=inspect.getdoc(func),
137 invoke=invoke,
138 )
141def command_for_object(
142 obj: Any, *, name: str | None = None, help: str | None = None
143) -> cappa.Command:
144 """Recursively convert a decorated runtime object into a Cappa command."""
145 options: dict[str, cappa.Command] = {}
147 for cli_name, kind, member in _cli_members(obj):
148 if kind == "command":
149 options[cli_name] = command_for_callable(cli_name, member)
150 continue
152 children = member()
153 if not isinstance(children, Mapping):
154 raise TypeError(f"CLI group {member.__qualname__} must return a mapping")
155 group_options = {
156 str(child_name): command_for_object(
157 child,
158 name=str(child_name),
159 help=inspect.getdoc(child.__class__),
160 )
161 for child_name, child in children.items()
162 }
163 options[cli_name] = _group_command(
164 cli_name, group_options, inspect.getdoc(member)
165 )
167 command_name = name or obj.__class__.__name__.lower()
168 return _group_command(command_name, options, help or inspect.getdoc(obj.__class__))
171def _group_command(
172 name: str, options: Mapping[str, cappa.Command], help: str | None
173) -> cappa.Command:
174 schema = make_dataclass(
175 f"{name.title().replace('-', '')}Command",
176 [("selected", Any, field(default=None))],
177 )
178 arguments = []
179 if options:
180 arguments.append(cappa.Subcommand(field_name="selected", options=options))
181 return cappa.Command(schema, name=name, arguments=arguments, help=help)
184def invoke(obj: Any, argv: list[str] | None = None, *, name: str = "collectra"):
185 """Build and invoke the recursive command tree for ``obj``."""
186 return cappa.invoke(command_for_object(obj, name=name), argv=argv)