Discovery

Command discovery: turn a class, module, import path, directory, or installed distribution's entry points into runnable commands.

duho.discovery

Command discovery: turn a class, module, import path, or directory into Cmds.

This module answers "give me the runnable commands living over there" for four shapes of there:

• a :class:~duho.Cmd subclass -- already a command, used as-is;
• a command module -- a .py file whose top-level main/run/call is the entrypoint, adapted to the command contract by :class:ModuleCommand (a plain wrapper -- it does NOT subclass types.ModuleType);
• an import path (dotted qualname) or a filesystem path -- resolved to a module by :class:CmdBuilder;
• a package or directory -- walked by :func:discover_commands, which collects both class commands and module commands from every submodule/file.

Two design points worth calling out:

• Resilience. :func:discover_commands treats a single unimportable or unsupported command as skippable, not fatal: ImportError and NotImplementedError (and subclasses) on one command are logged and skipped so the rest still load. A genuinely broken command file (e.g. a SyntaxError) is NOT swallowed -- it is a real bug the author wants surfaced. See :func:discover_commands for the exact caught set and rationale.
• Injection hook. :func:register_command_provider lets an external package teach :class:CmdBuilder how to build a command from a directory shape core duho does not itself understand (e.g. a directory of numbered step files), WITHOUT core duho importing that package. If no provider matches, a directory or module is imported normally.

All union annotations are quoted so the module imports cleanly on Python 3.9.

__all__ = ['Command', 'ModuleCommand', 'CmdBuilder', 'register_command_provider', 'discover_commands', 'discover_entry_points', 'is_class_command', 'is_module_command'] module-attribute

CmdBuilder(qualname, source=None)

Build a :class:Command from an import path or a filesystem path.

CmdBuilder(qualname, source=None) resolves source to a command:

• source a Path (or path-like) -- a filesystem source. If a registered provider (:func:register_command_provider) matches it, that provider builds the command; otherwise it is imported. A .py file is imported via spec_from_file_location under a synthesized unique sys.modules key (so a loose file never clobbers a real installed module of the same dotted name). A directory with __init__.py is a package and imported by qualname; a directory without one is offered to providers first (the seam where a run-path-style runtime plugs in) and, if unclaimed, raises ImportError (core duho has no meaning for a bare dir of files -- that meaning is exactly what a provider supplies).
• source omitted/None -- qualname is treated as a dotted import path and imported via importlib.import_module (after checking providers for a namespace-package directory, mirroring the path branch).
• source already a module or Command -- used directly.

The resolved command is exposed as :attr:command. For a module source it is a :class:ModuleCommand; for a provider it is whatever the provider returns; for an already-Command source it is that object.

Source code in src/duho/discovery.py

def __init__(
    self,
    qualname: "str | _PythonName",
    source: "_Path | str | _os.PathLike | _ModuleType | Command | None" = None,
) -> None:
    self.qualname = str(qualname)

    if isinstance(source, (_Path, _os.PathLike)) and not isinstance(source, str):
        self.command = self._from_path(_Path(source))
    elif source is None:
        self.command = self._from_import(self.qualname)
    elif isinstance(source, _ModuleType):
        self.command = self._wrap_module(source)
    elif is_class_command(source) or is_module_command(source):
        self.command = _ty.cast(Command, source)
    else:
        # A path given as a plain string.
        self.command = self._from_path(_Path(_ty.cast(str, source)))

command = self._from_path(_Path(source)) instance-attribute

qualname = str(qualname) instance-attribute

Command

Bases: Protocol

The shape discover_commands/dispatch needs from a command.

A command is anything that can name itself as a subcommand and be run. Two concrete kinds fulfil it:

• a :class:~duho.Cmd subclass -- a class command; its _parsername_/class name names the subcommand, _parser_ builds its parser, and an instance is run via __call__;
• a :class:ModuleCommand -- a module command wrapping a .py module, exposing the same surface.

This is a structural Protocol (not an ABC): the predicates :func:is_class_command / :func:is_module_command classify concrete objects, and callers branch on those rather than instantiating an ABC.

__call__()

Source code in src/duho/discovery.py

def __call__(self) -> object:  # pragma: no cover - protocol stub
    ...

ModuleCommand(module, *, name=None, entrypoint=None)

Adapt a command module to the :class:Command contract.

Wraps a module (a types.ModuleType, or anything exposing the same attributes) whose top-level function is the command body. It is a plain wrapper -- it does NOT subclass ModuleType -- so it stays a normal object with no import-system entanglement.

It carries:

• module -- the wrapped module;
• _parsername_ -- the resolved subcommand name (module _parsername_/ _cli_name override, else the file stem with _ -> -);
• description / help -- from module.__doc__;
• the entrypoint -- module.main (primary), falling back to module.run / module.call;
• optional lifecycle hooks -- register (default no-op), init (default returns None -- no context), success / finally_ (default no-ops);
• args_cls -- an optional module-level Args declaring this module's own CLI fields DECLARATIVELY (added to the subparser before register runs -- see runtime._register_module_command), as an alternative to adding everything imperatively in register. Accepts either an already-Args-subclassing Args (used directly) or a plain class with annotated fields (mixed in with Args on the fly so its annotations still work as CLI fields, with no explicit import/subclass of duho.Args required). None if the module declares no usable Args.

A ModuleCommand with no entrypoint raises NotImplementedError at construction (a module offering no main/run/call is not a command) -- discover_commands treats that as a skippable "not a command" signal, so a helpers-only module simply contributes nothing.

Hook signatures / logger source. The lifecycle hooks (init/success/ finally_) and the entrypoint receive the parsed args instance and take their logger from that instance's _logger_ (present on LoggingArgs-based commands) rather than a separately threaded logger argument. The resolved logger is available to hook authors as args._logger_ where the args class provides it, else this module's "duho" logger. The concrete hook calls made by the driver are: ctx = init(args), main(args) / entrypoint(args), success(ctx, args), finally_(ctx, args); the defaults installed here accept *args, **kwargs so a module may omit any hook (or define a narrower signature and simply not receive extras it does not declare).

The register hook is the one exception, and is arity-tolerant: it may be written either register(parser, args) (2-arg) or register(parser, args, logger) (3-arg). The driver (runtime._register_module_command) inspects the hook's signature and, for a 3-arg hook, passes logger = getattr(args, "_logger_", logging.getLogger("duho")); a 2-arg hook is called unchanged. A *args hook is treated as 3-arg-capable, and a non-introspectable hook falls back to the 2-arg call. Either way the hook adds its own arguments directly on the subcommand's argparse parser.

Source code in src/duho/discovery.py

def __init__(
    self,
    module: object,
    *,
    name: "str | None" = None,
    entrypoint: "_ty.Callable[..., object] | None" = None,
) -> None:
    self.module = module
    self._parsername_ = name or _resolved_module_name(module)

    entry = entrypoint if entrypoint is not None else _module_entrypoint(module)
    if entry is None:
        raise NotImplementedError(
            "module %r is not a command: it defines none of %s"
            % (getattr(module, "__name__", module), ", ".join(_ENTRYPOINT_NAMES))
        )
    self._entrypoint = entry

    # A module-level `Args` declares this module's own CLI fields,
    # combined with the app's shared root class (see
    # `runtime._register_module_command`, which does the actual mixing
    # since the root class is only known at registration time) and added
    # to the subparser before `register` runs. Stored as-is here: either
    # a real `Args` subclass (used directly by the caller) or a plain
    # class (mixed with the root at registration time so its own
    # annotated attrs still work as CLI fields -- `_introspect.get_clsargs`
    # walks the MRO -- without the author needing to import/subclass
    # `duho.Args` or the app's root explicitly).
    #
    # A STRICT-subclass check distinguishes "a real declared class" from
    # "the module did `from duho import Args` for its own use but never
    # subclassed it" -- `getattr(module, "Args", None)` would resolve to
    # `duho.args.Args` (or `Cmd`/`Cli`) itself there, which this correctly
    # treats as "nothing declared", not a usable class.
    args_cls = getattr(module, "Args", None)
    self.args_cls: "type | None" = (
        args_cls
        if _inspect.isclass(args_cls) and args_cls not in (_Args, _Cmd)
        else None
    )

    # Bind lifecycle hooks with contract defaults. ``init`` defaults to a
    # context-less builder (returns None); the others to no-ops. All
    # defaults swallow extra args so the driver's call shape need not match
    # a module's chosen arity exactly.
    self.register = getattr(module, "register", None) or _noop
    self.init = getattr(module, "init", None) or _init_noop
    self.success = getattr(module, "success", None) or _noop
    self.finally_ = getattr(module, "finally_", None) or _noop

args_cls = args_cls if _inspect.isclass(args_cls) and args_cls not in (_Args, _Cmd) else None instance-attribute

description property

Full command help -- the wrapped module's docstring, stripped.

finally_ = getattr(module, 'finally_', None) or _noop instance-attribute

help property

One-line help -- the first line of :attr:description.

init = getattr(module, 'init', None) or _init_noop instance-attribute

module = module instance-attribute

register = getattr(module, 'register', None) or _noop instance-attribute

success = getattr(module, 'success', None) or _noop instance-attribute

__call__(args=None)

A ModuleCommand is directly callable; delegates to :meth:main.

Source code in src/duho/discovery.py

def __call__(self, args: "object | None" = None) -> object:
    """A ``ModuleCommand`` is directly callable; delegates to :meth:`main`."""
    return self.main(args)

__repr__()

Source code in src/duho/discovery.py

def __repr__(self) -> str:
    return "ModuleCommand(name=%r, module=%r)" % (
        self._parsername_,
        getattr(self.module, "__name__", self.module),
    )

main(args=None)

Run the command by invoking the wrapped module's entrypoint.

Called with the parsed args instance during dispatch. Kept arg-optional so a ModuleCommand is trivially callable in tests / direct use; the driver always passes the parsed args.

Source code in src/duho/discovery.py

def main(self, args: "object | None" = None) -> object:
    """Run the command by invoking the wrapped module's entrypoint.

    Called with the parsed args instance during dispatch. Kept
    arg-optional so a ``ModuleCommand`` is trivially callable in tests /
    direct use; the driver always passes the parsed args.
    """
    if args is None:
        return self._entrypoint()
    return self._entrypoint(args)

discover_commands(source)

Discover commands from a package name or a directory, resiliently.

source is dispatched by shape:

• a Path/os.PathLike, or a str containing / or \ or naming an existing directory -> filesystem: iterate sorted(dir.glob("*.py")), skip _-prefixed files, import each under a synthesized unique sys.modules name, and collect its commands;
• any other str -> dotted package: import_module it, require a __path__ (it must be a package, not a plain module), walk its submodules with pkgutil.iter_modules, import each, and collect.

From each module it collects BOTH class commands (Cmd subclasses defined in that module) and, if the module has a top-level main/run/call, one :class:ModuleCommand. A module with neither contributes nothing.

Resilience. Per-command import/build is wrapped to catch only ImportError and NotImplementedError (and their subclasses): these mean "unsupported/optional-dep-missing" or "not actually a command", which are skippable -- logged and skipped so the other commands still load. Any other exception (notably SyntaxError -- a typo in a command file -- and unexpected runtime errors) propagates: a genuinely broken command file is a real bug the author wants surfaced, not silently swallowed.

The result is sorted by resolved subcommand name for deterministic --help output (filesystem iteration order is OS-dependent).

Source code in src/duho/discovery.py

def discover_commands(source: "str | _os.PathLike | _Path") -> "list[Command]":
    """Discover commands from a package name or a directory, resiliently.

    ``source`` is dispatched by shape:

    * a ``Path``/``os.PathLike``, or a ``str`` containing ``/`` or ``\\`` or
      naming an existing directory -> **filesystem**: iterate
      ``sorted(dir.glob("*.py"))``, skip ``_``-prefixed files, import each under
      a synthesized unique ``sys.modules`` name, and collect its commands;
    * any other ``str`` -> **dotted package**: ``import_module`` it, require a
      ``__path__`` (it must be a package, not a plain module), walk its
      submodules with ``pkgutil.iter_modules``, import each, and collect.

    From each module it collects BOTH class commands (``Cmd`` subclasses defined
    in that module) and, if the module has a top-level ``main``/``run``/``call``,
    one :class:`ModuleCommand`. A module with neither contributes nothing.

    **Resilience.** Per-command import/build is wrapped to catch **only**
    ``ImportError`` and ``NotImplementedError`` (and their subclasses): these
    mean "unsupported/optional-dep-missing" or "not actually a command", which
    are skippable -- logged and skipped so the *other* commands still load. Any
    other exception (notably ``SyntaxError`` -- a typo in a command file -- and
    unexpected runtime errors) propagates: a genuinely broken command file is a
    real bug the author wants surfaced, not silently swallowed.

    The result is sorted by resolved subcommand name for deterministic
    ``--help`` output (filesystem iteration order is OS-dependent).
    """
    if _looks_like_path(source):
        commands = _discover_from_path(_Path(source))
    else:
        commands = _discover_from_package(str(source))
    return sorted(commands, key=_command_name)

discover_entry_points(group)

Discover commands from installed-distribution entry points in group.

This is the plugin-discovery source behind duho.app(root, entry_points="myapp.commands"): every entry point advertised in group by an installed distribution is loaded (EntryPoint.load()) and coerced to a :class:Command via :func:_coerce_entry_point_command -- a Cmd subclass becomes a class command, a module becomes a module command.

Resilience mirrors :func:discover_commands: an entry point that fails to load (a broken/renamed target, a missing optional dependency) or that does not resolve to a command is logged at WARNING and skipped, so one bad plugin never takes the app down -- the rest still load.

importlib.metadata is imported lazily (inside :func:_compat.iter_entry_points) so a plain import duho never pays its cost -- only calling this triggers the load (plan 02 P1). The result is sorted by resolved subcommand name for deterministic --help output.

Source code in src/duho/discovery.py

def discover_entry_points(group: str) -> "list[Command]":
    """Discover commands from installed-distribution entry points in ``group``.

    This is the plugin-discovery source behind ``duho.app(root,
    entry_points="myapp.commands")``: every entry point advertised in ``group``
    by an installed distribution is loaded (``EntryPoint.load()``) and coerced to
    a :class:`Command` via :func:`_coerce_entry_point_command` -- a ``Cmd``
    subclass becomes a class command, a module becomes a module command.

    **Resilience** mirrors :func:`discover_commands`: an entry point that fails to
    load (a broken/renamed target, a missing optional dependency) or that does
    not resolve to a command is logged at ``WARNING`` and skipped, so one bad
    plugin never takes the app down -- the rest still load.

    ``importlib.metadata`` is imported lazily (inside :func:`_compat.iter_entry_points`)
    so a plain ``import duho`` never pays its cost -- only calling this triggers
    the load (plan 02 P1). The result is sorted by resolved subcommand name for
    deterministic ``--help`` output.
    """
    commands: "list[Command]" = []
    for entry_point in _compat.iter_entry_points(group):
        ep_name = getattr(entry_point, "name", None)
        try:
            loaded = entry_point.load()
            command = _coerce_entry_point_command(loaded, ep_name)
        except Exception as exc:  # noqa: BLE001 - a bad plugin must not abort the app
            _log_exception(
                _LOGGER,
                "skipping entry point %r in group %r: failed to load (%s)",
                ep_name if ep_name is not None else entry_point,
                group,
                exc,
                level=_logging.WARNING,
            )
            continue
        if command is None:
            _LOGGER.warning(
                "skipping entry point %r in group %r: %r is not a command "
                "(expected a Cmd subclass, a command module, or a Command)",
                ep_name if ep_name is not None else entry_point,
                group,
                loaded,
            )
            continue
        commands.append(command)
    return sorted(commands, key=_command_name)

is_class_command(obj)

True if obj is a class command: a Cmd subclass (not Cmd itself).

Source code in src/duho/discovery.py

def is_class_command(obj: object) -> bool:
    """True if ``obj`` is a class command: a ``Cmd`` subclass (not ``Cmd`` itself)."""
    return _inspect.isclass(obj) and issubclass(obj, _Cmd) and obj is not _Cmd

is_module_command(obj)

True if obj is a module command (a :class:ModuleCommand wrapper).

Source code in src/duho/discovery.py

def is_module_command(obj: object) -> bool:
    """True if ``obj`` is a module command (a :class:`ModuleCommand` wrapper)."""
    return isinstance(obj, ModuleCommand)

register_command_provider(predicate, builder)

Register an external provider that builds a Command from a directory.

This is the extension seam that keeps directory-shaped command runtimes (e.g. an ordered "run-path" of numbered step files) OUT of core duho: an external package registers (predicate, builder); when CmdBuilder resolves a filesystem source, it consults registered providers before importing the path normally, and the first matching provider's builder produces the command. If no provider matches, the path is imported as a plain module/package.

• predicate(path: Path) -> bool -- True if this provider handles path.
• builder(path: Path, qualname: str) -> Command -- build the command.

Providers are consulted most-recently-registered first, so a later registration can take precedence over an earlier one for the same shape.

Source code in src/duho/discovery.py

def register_command_provider(
    predicate: "_ty.Callable[[_Path], bool]",
    builder: "_ty.Callable[[_Path, str], object]",
) -> None:
    """Register an external provider that builds a ``Command`` from a directory.

    This is the extension seam that keeps directory-shaped command runtimes
    (e.g. an ordered "run-path" of numbered step files) OUT of core duho: an
    external package registers ``(predicate, builder)``; when ``CmdBuilder``
    resolves a filesystem source, it consults registered providers *before*
    importing the path normally, and the first matching provider's ``builder``
    produces the command. If no provider matches, the path is imported as a
    plain module/package.

    * ``predicate(path: Path) -> bool`` -- True if this provider handles ``path``.
    * ``builder(path: Path, qualname: str) -> Command`` -- build the command.

    Providers are consulted most-recently-registered first, so a later
    registration can take precedence over an earlier one for the same shape.
    """
    _PROVIDERS.insert(0, (predicate, builder))