Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.11/site-packages/face/command.py: 39%
Shortcuts on this page
r m x toggle line displays
j k next/prev highlighted chunk
0 (zero) top of page
1 (one) first highlighted chunk
1import sys
2from collections import OrderedDict
3from typing import Callable, List, Optional, Union
5from face.utils import unwrap_text, get_rdep_map, echo
6from face.errors import ArgumentParseError, CommandLineError, UsageError
7from face.parser import Parser, Flag, PosArgSpec
8from face.helpers import HelpHandler
9from face.middleware import (inject,
10 get_arg_names,
11 is_middleware,
12 face_middleware,
13 check_middleware,
14 get_middleware_chain,
15 _BUILTIN_PROVIDES)
17from boltons.strutils import camel2under
18from boltons.iterutils import unique
21def _get_default_name(func):
22 from functools import partial
23 if isinstance(func, partial):
24 func = func.func # just one level of partial for now
26 ret = getattr(func, '__name__', None) # most functions hit this
28 if ret is None:
29 ret = camel2under(func.__class__.__name__).lower() # callable instances, etc.
31 return ret
34def _docstring_to_doc(func):
35 doc = func.__doc__
36 if not doc:
37 return ''
39 unwrapped = unwrap_text(doc)
40 try:
41 ret = [g for g in unwrapped.splitlines() if g][0]
42 except IndexError:
43 ret = ''
45 return ret
48def default_print_error(msg):
49 return echo.err(msg)
52DEFAULT_HELP_HANDLER = HelpHandler()
55# TODO: should name really go here?
56class Command(Parser):
57 """The central type in the face framework. Instantiate a Command,
58 populate it with flags and subcommands, and then call
59 command.run() to execute your CLI.
61 Args:
62 func: The function called when this command is
63 run with an argv that contains no subcommands.
64 name: The name of this command, used when this
65 command is included as a subcommand. (Defaults to name
66 of function)
67 doc: A description or message that appears in various
68 help outputs.
69 flags: A list of Flag instances to initialize the
70 Command with. Flags can always be added later with the
71 .add() method.
72 posargs: Pass True if the command takes positional
73 arguments. Defaults to False. Can also pass a PosArgSpec
74 instance.
75 post_posargs: Pass True if the command takes
76 additional positional arguments after a conventional '--'
77 specifier.
78 help: Pass False to disable the automatically added
79 --help flag. Defaults to True. Also accepts a HelpHandler
80 instance.
81 middlewares: A list of @face_middleware decorated
82 callables which participate in dispatch.
83 """
84 def __init__(self,
85 func: Optional[Callable],
86 name: Optional[str] = None,
87 doc: Optional[str] = None,
88 *,
89 flags: Optional[List[Flag]] = None,
90 posargs: Optional[Union[bool, PosArgSpec]] = None,
91 post_posargs: Optional[bool] = None,
92 flagfile: bool = True,
93 help: Union[bool, HelpHandler] = DEFAULT_HELP_HANDLER,
94 middlewares: Optional[List[Callable]] = None) -> None:
95 name = name if name is not None else _get_default_name(func)
96 if doc is None:
97 doc = _docstring_to_doc(func)
99 # TODO: default posargs if none by inspecting func
100 super().__init__(name, doc,
101 flags=flags,
102 posargs=posargs,
103 post_posargs=post_posargs,
104 flagfile=flagfile)
106 self.help_handler = help
108 # TODO: if func is callable, check that "next_" isn't taken
109 self._path_func_map = OrderedDict()
110 self._path_func_map[()] = func
112 middlewares = list(middlewares or [])
113 self._path_mw_map = OrderedDict()
114 self._path_mw_map[()] = []
115 self._path_wrapped_map = OrderedDict()
116 self._path_wrapped_map[()] = func
117 for mw in middlewares:
118 self.add_middleware(mw)
120 if help:
121 if help is True:
122 help = DEFAULT_HELP_HANDLER
123 if help.flag:
124 self.add(help.flag)
125 if help.subcmd:
126 self.add(help.func, help.subcmd) # for 'help' as a subcmd
128 if not func and not help:
129 raise ValueError('Command requires a handler function or help handler'
130 ' to be set, not: %r' % func)
132 return
134 @property
135 def func(self):
136 return self._path_func_map[()]
138 def add(self, *a, **kw):
139 """Add a flag, subcommand, or middleware to this Command.
141 If the first argument is a callable, this method contructs a
142 Command from it and the remaining arguments, all of which are
143 optional. See the Command docs for for full details on names
144 and defaults.
146 If the first argument is a string, this method constructs a
147 Flag from that flag string and the rest of the method
148 arguments, all of which are optional. See the Flag docs for
149 more options.
151 If the argument is already an instance of Flag or Command, an
152 exception is only raised on conflicting subcommands and
153 flags. See add_command for details.
155 Middleware is only added if it is already decorated with
156 @face_middleware. Use .add_middleware() for automatic wrapping
157 of callables.
159 """
160 # TODO: need to check for middleware provides names + flag names
161 # conflict
163 target = a[0]
165 if is_middleware(target):
166 return self.add_middleware(target)
168 subcmd = a[0]
169 if not isinstance(subcmd, Command) and callable(subcmd) or subcmd is None:
170 subcmd = Command(*a, **kw) # attempt to construct a new subcmd
172 if isinstance(subcmd, Command):
173 self.add_command(subcmd)
174 return subcmd
176 flag = a[0]
177 if not isinstance(flag, Flag):
178 flag = Flag(*a, **kw) # attempt to construct a Flag from arguments
179 super().add(flag)
181 return flag
183 def add_command(self, subcmd):
184 """Add a Command, and all of its subcommands, as a subcommand of this
185 Command.
187 Middleware from the current command is layered on top of the
188 subcommand's. An exception may be raised if there are
189 conflicting middlewares or subcommand names.
190 """
191 if not isinstance(subcmd, Command):
192 raise TypeError(f'expected Command instance, not: {subcmd!r}')
193 self_mw = self._path_mw_map[()]
194 super().add(subcmd)
195 # map in new functions
196 for path in self.subprs_map:
197 if path not in self._path_func_map:
198 self._path_func_map[path] = subcmd._path_func_map[path[1:]]
199 sub_mw = subcmd._path_mw_map[path[1:]]
200 self._path_mw_map[path] = self_mw + sub_mw # TODO: check for conflicts
201 return
203 def add_middleware(self, mw):
204 """Add a single middleware to this command. Outermost middleware
205 should be added first. Remember: first added, first called.
207 """
208 if not is_middleware(mw):
209 mw = face_middleware(mw)
210 check_middleware(mw)
212 for flag in mw._face_flags:
213 self.add(flag)
215 for path, mws in self._path_mw_map.items():
216 self._path_mw_map[path] = [mw] + mws # TODO: check for conflicts
218 return
220 # TODO: add_flag()
222 def get_flag_map(self, path=(), with_hidden=True):
223 """Command's get_flag_map differs from Parser's in that it filters
224 the flag map to just the flags used by the endpoint at the
225 associated subcommand *path*.
226 """
227 flag_map = super().get_flag_map(path=path, with_hidden=with_hidden)
228 dep_names = self.get_dep_names(path)
229 if 'args_' in dep_names or 'flags_' in dep_names:
230 # the argument parse result and flag dict both capture
231 # _all_ the flags, so for functions accepting these
232 # arguments we bypass filtering.
234 # Also note that by setting an argument default in the
235 # function definition, the dependency becomes "weak", and
236 # this bypassing of filtering will not trigger, unless
237 # another function in the chain has a non-default,
238 # "strong" dependency. This behavior is especially useful
239 # for middleware.
241 # TODO: add decorator for the corner case where a function
242 # accepts these arguments and doesn't use them all.
243 return OrderedDict(flag_map)
245 return OrderedDict([(k, f) for k, f in flag_map.items() if f.name in dep_names
246 or f is self.flagfile_flag or f is self.help_handler.flag])
248 def get_dep_names(self, path=()):
249 """Get a list of the names of all required arguments of a command (and
250 any associated middleware).
252 By specifying *path*, the same can be done for any subcommand.
253 """
254 func = self._path_func_map[path]
255 if not func:
256 return [] # for when no handler is specified
258 mws = self._path_mw_map[path]
260 # start out with all args of handler function, which gets stronger dependencies
261 required_args = set(get_arg_names(func, only_required=False))
262 dep_map = {func: set(required_args)}
263 for mw in mws:
264 arg_names = set(get_arg_names(mw, only_required=True))
265 for provide in mw._face_provides:
266 dep_map[provide] = arg_names
267 if not mw._face_optional:
268 # all non-optional middlewares get their args required, too.
269 required_args.update(arg_names)
271 rdep_map = get_rdep_map(dep_map)
273 recursive_required_args = rdep_map[func].union(required_args)
275 return sorted(recursive_required_args)
277 def prepare(self, paths=None):
278 """Compile and validate one or more subcommands to ensure all
279 dependencies are met. Call this once all flags, subcommands,
280 and middlewares have been added (using .add()).
282 This method is automatically called by .run() method, but it
283 only does so for the specific subcommand being invoked. More
284 conscientious users may want to call this method with no
285 arguments to validate that all subcommands are ready for
286 execution.
287 """
288 # TODO: also pre-execute help formatting to make sure all
289 # values are sane there, too
290 if paths is None:
291 paths = self._path_func_map.keys()
293 for path in paths:
294 func = self._path_func_map[path]
295 if func is None:
296 continue # handled by run()
298 prs = self.subprs_map[path] if path else self
299 provides = []
300 if prs.posargs.provides:
301 provides += [prs.posargs.provides]
302 if prs.post_posargs.provides:
303 provides += [prs.post_posargs.provides]
305 deps = self.get_dep_names(path)
306 flag_names = [f.name for f in self.get_flags(path=path)]
307 all_mws = self._path_mw_map[path]
309 # filter out unused middlewares
310 mws = [mw for mw in all_mws if not mw._face_optional
311 or [p for p in mw._face_provides if p in deps]]
312 provides += _BUILTIN_PROVIDES + flag_names
313 try:
314 wrapped = get_middleware_chain(mws, func, provides)
315 except NameError as ne:
316 ne.args = (ne.args[0] + f' (in path: {path!r})',)
317 raise
319 self._path_wrapped_map[path] = wrapped
321 return
323 def run(self, argv=None, extras=None, print_error=None):
324 """Parses arguments and dispatches to the appropriate subcommand
325 handler. If there is a parse error due to invalid user input,
326 an error is printed and a CommandLineError is raised. If not
327 caught, a CommandLineError will exit the process, typically
328 with status code 1. Also handles dispatching to the
329 appropriate HelpHandler, if configured.
331 Defaults to handling the arguments on the command line
332 (``sys.argv``), but can also be explicitly passed arguments
333 via the *argv* parameter.
335 Args:
336 argv (list): A sequence of strings representing the
337 command-line arguments. Defaults to ``sys.argv``.
338 extras (dict): A map of additional arguments to be made
339 available to the subcommand's handler function.
340 print_error (callable): The function that formats/prints
341 error messages before program exit on CLI errors.
343 .. note::
345 For efficiency, :meth:`run()` only checks the subcommand
346 invoked by *argv*. To ensure that all subcommands are
347 configured properly, call :meth:`prepare()`.
349 """
350 if print_error is None or print_error is True:
351 print_error = default_print_error
352 elif print_error and not callable(print_error):
353 raise TypeError(f'expected callable for print_error, not {print_error!r}')
355 kwargs = dict(extras) if extras else {}
356 kwargs['print_error_'] = print_error # TODO: print_error_ in builtin provides?
358 try:
359 prs_res = self.parse(argv=argv)
360 except ArgumentParseError as ape:
361 prs_res = ape.prs_res
363 # even if parsing failed, check if the caller was trying to access the help flag
364 cmd = prs_res.to_cmd_scope()['subcommand_']
365 if cmd.help_handler and prs_res.flags and prs_res.flags.get(cmd.help_handler.flag.name):
366 kwargs.update(prs_res.to_cmd_scope())
367 return inject(cmd.help_handler.func, kwargs)
369 msg = 'error: ' + (prs_res.name or self.name)
370 if prs_res.subcmds:
371 msg += ' ' + ' '.join(prs_res.subcmds or ())
373 # args attribute, nothing to do with cmdline args this is
374 # the standard-issue Exception
375 e_msg = ape.args[0]
376 if e_msg:
377 msg += ': ' + e_msg
378 cle = CommandLineError(msg)
379 if print_error:
380 print_error(msg)
381 raise cle
383 kwargs.update(prs_res.to_cmd_scope())
385 # default in case no middlewares have been installed
386 func = self._path_func_map[prs_res.subcmds]
388 cmd = kwargs['subcommand_']
389 if cmd.help_handler and (not func or (prs_res.flags and prs_res.flags.get(cmd.help_handler.flag.name))):
390 return inject(cmd.help_handler.func, kwargs)
391 elif not func: # pragma: no cover
392 raise RuntimeError('expected command handler or help handler to be set')
394 self.prepare(paths=[prs_res.subcmds])
395 wrapped = self._path_wrapped_map.get(prs_res.subcmds, func)
397 try:
398 ret = inject(wrapped, kwargs)
399 except UsageError as ue:
400 if print_error:
401 print_error(ue.format_message())
402 raise
403 return ret