lib.App / lib.Cmd: Build command tree lazily #68
2 changed files with 110 additions and 12 deletions
|
|
@ -36,6 +36,11 @@ if TYPE_CHECKING:
|
|||
from typing import TypeVar
|
||||
T = TypeVar('T')
|
||||
|
||||
class _SubCommand(NamedTuple):
|
||||
|
||||
cmd: AbstractCmd
|
||||
parser: ArgumentParser
|
||||
|
||||
def _get_current_event_loop() -> asyncio.AbstractEventLoop | None:
|
||||
"""Return the current event loop of this thread, or None if there is
|
||||
none, without creating one implicitly or emitting a deprecation
|
||||
|
|
@ -47,10 +52,63 @@ def _get_current_event_loop() -> asyncio.AbstractEventLoop | None:
|
|||
except (RuntimeError, DeprecationWarning):
|
||||
return None
|
||||
|
||||
class _SubCommand(NamedTuple):
|
||||
def _build_option_map(parser: ArgumentParser) -> dict[str, int | str | None]:
|
||||
"""Map each option string to its nargs, so the invoked-path walk can tell
|
||||
which options consume the following token (a value) and which are bare
|
||||
flags."""
|
||||
opt_map: dict[str, int | str | None] = {}
|
||||
for action in parser._actions: # noqa: SLF001
|
||||
for opt in getattr(action, 'option_strings', ()):
|
||||
opt_map[opt] = action.nargs
|
||||
return opt_map
|
||||
|
||||
cmd: AbstractCmd
|
||||
parser: ArgumentParser
|
||||
def _find_invoked_subcommand(
|
||||
opt_map: dict[str, int | str | None],
|
||||
names: set[str],
|
||||
argv: list[str],
|
||||
pos: int,
|
||||
) -> tuple[str | None, int]:
|
||||
"""Find the invoked subcommand name in argv starting at pos.
|
||||
|
||||
Walk the argv, skipping options and the tokens they consume (using
|
||||
opt_map), and return the first positional token that is a subcommand
|
||||
name, along with the position just past it. Return (None, pos) if no
|
||||
subcommand name is present at or after pos.
|
||||
"""
|
||||
n = len(argv)
|
||||
i = pos
|
||||
while i < n:
|
||||
tok = argv[i]
|
||||
if tok.startswith('-') and len(tok) > 1:
|
||||
if '=' in tok:
|
||||
# --opt=value: the value is inline, consume the token only.
|
||||
i += 1
|
||||
elif tok in opt_map and opt_map[tok] != 0:
|
||||
nargs = opt_map[tok]
|
||||
if nargs in ('+', '*') or (isinstance(nargs, int) and nargs > 1):
|
||||
# -- Multi-value option: consume the following non-option,
|
||||
# non-subcommand tokens (its values).
|
||||
j = i + 1
|
||||
while (j < n and not argv[j].startswith('-')
|
||||
and argv[j] not in names):
|
||||
j += 1
|
||||
i = j
|
||||
elif i + 1 < n and not argv[i + 1].startswith('-'):
|
||||
# -- One value: consume the following token.
|
||||
i += 2
|
||||
else:
|
||||
i += 1
|
||||
else:
|
||||
# -- Flag or unknown option: consume the token only.
|
||||
i += 1
|
||||
continue
|
||||
if tok in names:
|
||||
return tok, i + 1
|
||||
# -- A positional token that is not a subcommand name: the invoked
|
||||
# path ends here.
|
||||
return None, i
|
||||
# -- Ran off the end of the argv consuming option values: no subcommand.
|
||||
return None, i
|
||||
|
||||
class App: # export
|
||||
|
||||
|
|
@ -185,6 +243,7 @@ class App: # export
|
|||
cmds: Collection[AbstractCmd],
|
||||
all: bool,
|
||||
top_level: bool,
|
||||
pos: int,
|
||||
) -> None:
|
||||
if not cmds:
|
||||
return
|
||||
|
|
@ -223,15 +282,24 @@ class App: # export
|
|||
sc.cmd.children,
|
||||
all = all,
|
||||
top_level = False,
|
||||
pos = pos,
|
||||
)
|
||||
return
|
||||
# -- Re-parse the command line to find the invoked subcommand.
|
||||
# This works because every level below uses dest = 'command',
|
||||
# so each pass descends one level further into the command
|
||||
# tree.
|
||||
args, _ = self.__parser.parse_known_args(argv)
|
||||
cmd_name = getattr(args, 'command', None)
|
||||
if cmd_name in scs:
|
||||
# -- Find the invoked subcommand by walking the argv. The
|
||||
# subcommand names at this level are known (the commands are
|
||||
# materialized), so the invoked path is found by matching tokens
|
||||
# against those names and skipping the options (and their
|
||||
# values) that precede them. Walking the tokens never parses the
|
||||
# tail against a half-built parser, so a deeper option cannot
|
||||
# collide with a same-named option of a shallower level the way
|
||||
# a discovery re-parse would.
|
||||
cmd_name, new_pos = _find_invoked_subcommand(
|
||||
_build_option_map(parser),
|
||||
set(scs.keys()),
|
||||
argv_list,
|
||||
pos,
|
||||
)
|
||||
if cmd_name is not None and cmd_name in scs:
|
||||
sc = scs[cmd_name]
|
||||
add_cmds_to_parser(
|
||||
sc.cmd,
|
||||
|
|
@ -239,11 +307,16 @@ class App: # export
|
|||
sc.cmd.children,
|
||||
all = all,
|
||||
top_level = False,
|
||||
pos = new_pos,
|
||||
)
|
||||
|
||||
cmdline = sys.argv if argv is None else argv
|
||||
if argv is None:
|
||||
argv = sys.argv[1:]
|
||||
# -- The argv the invoked-path walk (below) operates on. A separate
|
||||
# binding so the nested closure sees a plain list[str], not the
|
||||
# optional parameter type.
|
||||
argv_list: list[str] = argv
|
||||
add_all_parsers = (
|
||||
'-h' in argv or '--help' in argv or '_ARGCOMPLETE' in os.environ
|
||||
)
|
||||
|
|
@ -279,6 +352,7 @@ class App: # export
|
|||
self.__cmds,
|
||||
all = add_all_parsers,
|
||||
top_level = False,
|
||||
pos = 0,
|
||||
)
|
||||
else:
|
||||
add_cmds_to_parser(
|
||||
|
|
@ -287,6 +361,7 @@ class App: # export
|
|||
self.__root.children,
|
||||
all = add_all_parsers,
|
||||
top_level = True,
|
||||
pos = 0,
|
||||
)
|
||||
|
||||
# -- Add help only now, wouldn't want to have parse_known_args() exit
|
||||
|
|
|
|||
|
|
@ -25,6 +25,12 @@ class AbstractCmd(abc.ABC):
|
|||
self.__children: list[Cmd] = []
|
||||
self.__child_classes: list[type[Cmd]] = []
|
||||
self.__parser: ArgumentParser | None = None
|
||||
# -- Subcommands registered via load_subcommands() are not built
|
||||
# immediately; the module search path and name filter are stored here
|
||||
# and the subcommands are materialized on first access to `children`
|
||||
# (see __materialize_pending_subcommands()). This keeps a simple run
|
||||
# from instantiating the whole command tree.
|
||||
self.__pending_subcommands: tuple[list[str], str] | None = None
|
||||
|
||||
def set_parent(self, parent: Any | Cmd) -> None:
|
||||
self.__parent = parent
|
||||
|
|
@ -61,12 +67,25 @@ class AbstractCmd(abc.ABC):
|
|||
parent = parent.__parent
|
||||
return self.__app
|
||||
|
||||
def __materialize_pending_subcommands(self) -> None:
|
||||
# -- Build the subcommands that load_subcommands() registered
|
||||
# lazily, if any. Called on first access to `children` (and
|
||||
# `child_classes`) so that only the parts of the tree a run actually
|
||||
# descends into are ever instantiated.
|
||||
if self.__pending_subcommands is None:
|
||||
return
|
||||
modules, name_filter = self.__pending_subcommands
|
||||
self.__pending_subcommands = None
|
||||
self.add_subcommands(LoadTypes(modules, type_name_filter = name_filter))
|
||||
|
||||
@property
|
||||
def children(self) -> tuple[Cmd, ...]:
|
||||
self.__materialize_pending_subcommands()
|
||||
return tuple(self.__children)
|
||||
|
||||
@property
|
||||
def child_classes(self) -> tuple[type[Cmd], ...]:
|
||||
self.__materialize_pending_subcommands()
|
||||
return tuple(self.__child_classes)
|
||||
|
||||
@property
|
||||
|
|
@ -127,7 +146,11 @@ class AbstractCmd(abc.ABC):
|
|||
modules = [type(self).__module__.replace('Cmd', '').lower()]
|
||||
elif isinstance(modules, str):
|
||||
modules = [modules]
|
||||
self.add_subcommands(LoadTypes(modules, type_name_filter = name_filter))
|
||||
# -- Defer the actual subcommand construction: store the search path
|
||||
# and filter, and materialize on first access to `children`. Building
|
||||
# them here (in __init__) would instantiate the entire tree up front,
|
||||
# even for a run that only descends into a single branch.
|
||||
self.__pending_subcommands = (modules, name_filter)
|
||||
|
||||
# -- Interface to derived classes
|
||||
|
||||
|
|
|
|||
Loading…
Reference in a new issue