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
|
from typing import TypeVar
|
||||||
T = TypeVar('T')
|
T = TypeVar('T')
|
||||||
|
|
||||||
|
class _SubCommand(NamedTuple):
|
||||||
|
|
||||||
|
cmd: AbstractCmd
|
||||||
|
parser: ArgumentParser
|
||||||
|
|
||||||
def _get_current_event_loop() -> asyncio.AbstractEventLoop | None:
|
def _get_current_event_loop() -> asyncio.AbstractEventLoop | None:
|
||||||
"""Return the current event loop of this thread, or None if there is
|
"""Return the current event loop of this thread, or None if there is
|
||||||
none, without creating one implicitly or emitting a deprecation
|
none, without creating one implicitly or emitting a deprecation
|
||||||
|
|
@ -47,10 +52,63 @@ def _get_current_event_loop() -> asyncio.AbstractEventLoop | None:
|
||||||
except (RuntimeError, DeprecationWarning):
|
except (RuntimeError, DeprecationWarning):
|
||||||
return None
|
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
|
def _find_invoked_subcommand(
|
||||||
parser: ArgumentParser
|
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
|
class App: # export
|
||||||
|
|
||||||
|
|
@ -185,6 +243,7 @@ class App: # export
|
||||||
cmds: Collection[AbstractCmd],
|
cmds: Collection[AbstractCmd],
|
||||||
all: bool,
|
all: bool,
|
||||||
top_level: bool,
|
top_level: bool,
|
||||||
|
pos: int,
|
||||||
) -> None:
|
) -> None:
|
||||||
if not cmds:
|
if not cmds:
|
||||||
return
|
return
|
||||||
|
|
@ -223,15 +282,24 @@ class App: # export
|
||||||
sc.cmd.children,
|
sc.cmd.children,
|
||||||
all = all,
|
all = all,
|
||||||
top_level = False,
|
top_level = False,
|
||||||
|
pos = pos,
|
||||||
)
|
)
|
||||||
return
|
return
|
||||||
# -- Re-parse the command line to find the invoked subcommand.
|
# -- Find the invoked subcommand by walking the argv. The
|
||||||
# This works because every level below uses dest = 'command',
|
# subcommand names at this level are known (the commands are
|
||||||
# so each pass descends one level further into the command
|
# materialized), so the invoked path is found by matching tokens
|
||||||
# tree.
|
# against those names and skipping the options (and their
|
||||||
args, _ = self.__parser.parse_known_args(argv)
|
# values) that precede them. Walking the tokens never parses the
|
||||||
cmd_name = getattr(args, 'command', None)
|
# tail against a half-built parser, so a deeper option cannot
|
||||||
if cmd_name in scs:
|
# 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]
|
sc = scs[cmd_name]
|
||||||
add_cmds_to_parser(
|
add_cmds_to_parser(
|
||||||
sc.cmd,
|
sc.cmd,
|
||||||
|
|
@ -239,11 +307,16 @@ class App: # export
|
||||||
sc.cmd.children,
|
sc.cmd.children,
|
||||||
all = all,
|
all = all,
|
||||||
top_level = False,
|
top_level = False,
|
||||||
|
pos = new_pos,
|
||||||
)
|
)
|
||||||
|
|
||||||
cmdline = sys.argv if argv is None else argv
|
cmdline = sys.argv if argv is None else argv
|
||||||
if argv is None:
|
if argv is None:
|
||||||
argv = sys.argv[1:]
|
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 = (
|
add_all_parsers = (
|
||||||
'-h' in argv or '--help' in argv or '_ARGCOMPLETE' in os.environ
|
'-h' in argv or '--help' in argv or '_ARGCOMPLETE' in os.environ
|
||||||
)
|
)
|
||||||
|
|
@ -279,6 +352,7 @@ class App: # export
|
||||||
self.__cmds,
|
self.__cmds,
|
||||||
all = add_all_parsers,
|
all = add_all_parsers,
|
||||||
top_level = False,
|
top_level = False,
|
||||||
|
pos = 0,
|
||||||
)
|
)
|
||||||
else:
|
else:
|
||||||
add_cmds_to_parser(
|
add_cmds_to_parser(
|
||||||
|
|
@ -287,6 +361,7 @@ class App: # export
|
||||||
self.__root.children,
|
self.__root.children,
|
||||||
all = add_all_parsers,
|
all = add_all_parsers,
|
||||||
top_level = True,
|
top_level = True,
|
||||||
|
pos = 0,
|
||||||
)
|
)
|
||||||
|
|
||||||
# -- Add help only now, wouldn't want to have parse_known_args() exit
|
# -- 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.__children: list[Cmd] = []
|
||||||
self.__child_classes: list[type[Cmd]] = []
|
self.__child_classes: list[type[Cmd]] = []
|
||||||
self.__parser: ArgumentParser | None = None
|
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:
|
def set_parent(self, parent: Any | Cmd) -> None:
|
||||||
self.__parent = parent
|
self.__parent = parent
|
||||||
|
|
@ -61,12 +67,25 @@ class AbstractCmd(abc.ABC):
|
||||||
parent = parent.__parent
|
parent = parent.__parent
|
||||||
return self.__app
|
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
|
@property
|
||||||
def children(self) -> tuple[Cmd, ...]:
|
def children(self) -> tuple[Cmd, ...]:
|
||||||
|
self.__materialize_pending_subcommands()
|
||||||
return tuple(self.__children)
|
return tuple(self.__children)
|
||||||
|
|
||||||
@property
|
@property
|
||||||
def child_classes(self) -> tuple[type[Cmd], ...]:
|
def child_classes(self) -> tuple[type[Cmd], ...]:
|
||||||
|
self.__materialize_pending_subcommands()
|
||||||
return tuple(self.__child_classes)
|
return tuple(self.__child_classes)
|
||||||
|
|
||||||
@property
|
@property
|
||||||
|
|
@ -127,7 +146,11 @@ class AbstractCmd(abc.ABC):
|
||||||
modules = [type(self).__module__.replace('Cmd', '').lower()]
|
modules = [type(self).__module__.replace('Cmd', '').lower()]
|
||||||
elif isinstance(modules, str):
|
elif isinstance(modules, str):
|
||||||
modules = [modules]
|
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
|
# -- Interface to derived classes
|
||||||
|
|
||||||
|
|
|
||||||
Loading…
Reference in a new issue