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__ = ['Command', 'ModuleCommand', 'CmdBuilder', 'register_command_provider', 'unregister_command_provider', 'import_from_path', '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 in a private duho._cmdbuilder. namespace (so a loose file NEVER clobbers a real installed module of the same dotted name, even one not imported yet -- e.g. CmdBuilder("json", ...) never becomes sys.modules["json"]). A directory with __init__.py is a package and imported by qualname, but ONLY after verifying that qualname actually resolves (via sys.path) to this same directory -- otherwise a same-named package elsewhere on sys.path would be silently wrapped instead, or a real ImportError with no message raised. A directory without __init__.py 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/_builder.py

def __init__(
    self,
    qualname: _ty.Union[str, _PythonName],
    source: _ty.Optional[
        _ty.Union[_Path, str, _os.PathLike, _ModuleType, Command]
    ] = 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 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/_command.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 (a module-level _parsername_ override, else the file stem with _ -> -);
• description / help -- from module.__doc__;
• the entrypoint -- module.main (primary), falling back to module.run / module.call. Only a callable genuinely DEFINED in the module counts (see :func:_own_callable) -- an imported function of the same name (from subprocess import run) is never mistaken for the entrypoint;
• 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.
• _mcp_ -- a module-level _mcp_ = False opts this command (and, when it is a namespace, its whole subtree) out of the MCP tool surface -- duho.mcp's per-command exclusion. Default True (read via getattr(module, "_mcp_", True), so a module that never mentions it is unaffected).

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.

Only a callable genuinely DEFINED in the module is bound to a hook name (see :func:_own_callable) -- from colorama import init never becomes the init hook.

Hook signatures / logger source. The lifecycle hooks (init/success/ finally_) and the entrypoint receive the parsed args instance and read their logger from that instance's _logger_ (present on LoggingArgs-based commands) rather than a separately threaded logger argument. The driver (runtime.run_command) ensures args._logger_ is present before calling them: when the parsed instance has none, it sets one via :meth:_logger_for (the args instance's own _logger_ if present, else this module's "duho" logger) so a hook written against the documented convention never hits AttributeError on a plain root. 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/_command.py

def __init__(
    self,
    module: object,
    *,
    name: _ty.Optional[str] = None,
    entrypoint: _ty.Optional[_ty.Callable[..., object]] = None,
) -> None:
    self.module = module
    self._parsername_ = name or _resolved_module_name(module)
    # A module-level `_mcp_ = False` excludes this command from the MCP
    # tool surface; `duho.mcp` reads this attribute, never the module.
    self._mcp_ = getattr(module, "_mcp_", True)
    self._mcp_output_ = getattr(module, "_mcp_output_", None)

    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 plain class is mixed with the root class at registration. The
    # identity check rejects `Args`/`Cmd` themselves, which a module that
    # only imported them resolves here; subclasses pass.
    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 hooks, defaulting to no-ops that swallow extra args so the
    # driver's call shape need not match a module's arity. `_own_callable`
    # keeps an imported function of the same name from becoming a hook.
    self.register = _own_callable(module, "register") or _noop
    self.init = _own_callable(module, "init") or _init_noop
    self.success = _own_callable(module, "success") or _noop
    self.finally_ = _own_callable(module, "finally_") 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.

entrypoint property

The resolved callable (main/run/call or the one passed in).

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

help property

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

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

module = module instance-attribute

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

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

__call__(args=None)

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

Source code in src/duho/discovery/_command.py

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

__repr__()

Source code in src/duho/discovery/_command.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/_command.py

def main(self, args: _ty.Optional[object] = 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, *, on_error=None, providers=False, sort=True)

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

source may also be a list or tuple of sources: each is discovered on its own, a command of the same name in a later source replaces the earlier one, and the merged result is sorted by name as below. An empty sequence gives []. Each member goes through the same checks as a single source.

sort=False keeps discovery order instead of sorting: the sources as given; within a directory or package, files by name; within a file, definition order. A later command of the same name takes the place of the earlier one.

on_error(source, exc), when given, replaces the resilience rule below: it is called for any exception (SystemExit included) raised while importing one file or one package module, building its commands, or building a provider command, with source the file Path, the dotted module name, or the provider directory. Returning skips that one; raising aborts discovery. None keeps the rule below.

providers=True (filesystem sources only) first offers the source directory itself, then each of its child directories not starting with _ or ., to the registered command providers (:func:register_command_provider); a matching one builds a command named from the directory (_ as -). A source directory a provider claims yields that one command and its files are not scanned. Off by default.

A single source is dispatched by shape:

• a Path/os.PathLike, or a str containing / or \ -> 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 -> FIRST taken as a 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. Only when the name does not resolve to an importable module/package does it fall back to a same-named directory relative to the current working directory (filesystem mode, as above). This ordering is deliberate: a bare name is usually an intended package, and trying it first prevents an unrelated same-named directory in the CWD from silently shadowing it and having its .py files imported/executed instead (see :func:_looks_like_path).

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.

An empty source is rejected outright (ValueError): a bare "" or an os.PathLike whose __fspath__() is "" would otherwise silently discover the current working directory as a side effect of a blank/uninitialised value (a security-relevant fix -- this is also the path :func:duho.runtime.app's source= argument goes through, so app(source="") is covered the same way). Pass "." explicitly (a string, or any Path/os.PathLike -- Path("") included, since it already normalises to Path(".") before reaching here) to deliberately scan the current directory.

A bare drive letter is also rejected outright (ValueError): a "C:"-shaped source (see :func:_is_bare_drive_source) is Windows' own spelling for "the current directory on that drive" -- another, platform-specific way to smuggle the CWD in past the empty-source check above (a security-relevant fix, likewise covering app(source="C:")). Spell an actual path ("C:\cmds", or "." for the CWD) instead.

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

Source code in src/duho/discovery/_scan.py

def discover_commands(
    source: _ty.Union[
        str, _os.PathLike, _Path, _ty.Sequence[_ty.Union[str, _os.PathLike, _Path]]
    ],
    *,
    on_error: _ty.Optional[_ty.Callable[[object, BaseException], object]] = None,
    providers: bool = False,
    sort: bool = True,
) -> list[Command]:
    """Discover commands from a package name or a directory, resiliently.

    ``source`` may also be a list or tuple of sources: each is discovered on its
    own, a command of the same name in a later source replaces the earlier one,
    and the merged result is sorted by name as below. An empty sequence gives
    ``[]``. Each member goes through the same checks as a single source.

    ``sort=False`` keeps discovery order instead of sorting: the sources as
    given; within a directory or package, files by name; within a file,
    definition order. A later command of the same name takes the place of the
    earlier one.

    ``on_error(source, exc)``, when given, replaces the resilience rule below:
    it is called for any exception (``SystemExit`` included) raised while
    importing one file or one package module, building its commands, or building
    a provider command, with ``source`` the file ``Path``, the dotted module
    name, or the provider directory. Returning skips that one; raising aborts
    discovery. ``None`` keeps the rule below.

    ``providers=True`` (filesystem sources only) first offers the source
    directory itself, then each of its child directories not starting with ``_``
    or ``.``, to the registered command providers
    (:func:`register_command_provider`); a matching one builds a command named
    from the directory (``_`` as ``-``). A source directory a provider claims
    yields that one command and its files are not scanned. Off by default.

    A single ``source`` is dispatched by shape:

    * a ``Path``/``os.PathLike``, or a ``str`` containing ``/`` or ``\\`` ->
      **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`` -> FIRST taken as a **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. Only when the name does not resolve to an importable
      module/package does it fall back to a same-named directory relative to
      the current working directory (filesystem mode, as above). This
      ordering is deliberate: a bare name is usually an intended package, and
      trying it first prevents an unrelated same-named directory in the CWD
      from silently shadowing it and having its ``.py`` files imported/executed
      instead (see :func:`_looks_like_path`).

    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.

    **An empty source is rejected outright** (``ValueError``): a bare ``""``
    or an ``os.PathLike`` whose ``__fspath__()`` is ``""`` would otherwise
    silently discover the current working directory as a side effect of a
    blank/uninitialised value (a security-relevant fix -- this is also the
    path :func:`duho.runtime.app`'s ``source=`` argument goes through, so
    ``app(source="")`` is covered the same way). Pass ``"."`` explicitly (a
    string, or any ``Path``/``os.PathLike`` -- ``Path("")`` included, since it
    already normalises to ``Path(".")`` before reaching here) to deliberately
    scan the current directory.

    **A bare drive letter is also rejected outright** (``ValueError``): a
    ``"C:"``-shaped source (see :func:`_is_bare_drive_source`) is Windows'
    own spelling for "the current directory on that drive" -- another,
    platform-specific way to smuggle the CWD in past the empty-source check
    above (a security-relevant fix, likewise covering ``app(source="C:")``).
    Spell an actual path (``"C:\\cmds"``, or ``"."`` for the CWD) instead.

    The result is sorted by resolved subcommand name for deterministic
    ``--help`` output (filesystem iteration order is OS-dependent).
    """
    if isinstance(source, (list, tuple)):
        merged: dict[str, Command] = {}
        for member in source:
            for command in discover_commands(
                member, on_error=on_error, providers=providers, sort=sort
            ):
                merged[_command_name(command)] = command
        return (
            sorted(merged.values(), key=_command_name)
            if sort
            else list(merged.values())
        )
    if _is_empty_source(source):
        raise ValueError(
            "discover_commands(): an empty source is never valid -- it would "
            "silently discover the current working directory as a side "
            "effect of a blank/uninitialised value; pass '.' explicitly if "
            "that is intended"
        )
    if _is_bare_drive_source(source):
        raise ValueError(
            "discover_commands(): %r is a bare drive segment -- Windows "
            "resolves it to the current directory on that drive; use an "
            "actual path, or '.' for the current directory" % (source,)
        )
    namespace_dirs = (
        _namespace_locations(source)
        if isinstance(source, str) and not _looks_like_path(source)
        else []
    )
    if namespace_dirs:
        commands = [
            c
            for d in namespace_dirs
            for c in _discover_from_path(d, on_error=on_error, providers=providers)
        ]
    elif _looks_like_path(source):
        commands = _discover_from_path(
            _Path(source), on_error=on_error, providers=providers
        )
    else:
        commands = _discover_from_package(str(source), on_error=on_error)
    return sorted(commands, key=_command_name) if sort else commands

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. The result is sorted by resolved subcommand name for deterministic --help output.

Source code in src/duho/discovery/_entrypoints.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. 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)

import_from_path(base_name, path)

Import a .py file at path under a sys.modules key derived from base_name, guaranteed not to clobber an existing module of that name (:func:_unique_module_name) and reused across repeat imports of the SAME file (see :func:_import_from_path).

Public counterpart of the _unique_module_name + _import_from_path pair this module already used internally for its own filesystem-based discovery (:class:CmdBuilder, :func:discover_commands) -- exposed so an external command-provider package (duho.runpath is the first, and so far only, consumer) can import a .py file the exact same way without reaching into either private helper directly.

Source code in src/duho/discovery/_importing.py

def import_from_path(base_name: str, path: _Path) -> _ModuleType:
    """Import a ``.py`` file at ``path`` under a ``sys.modules`` key derived
    from ``base_name``, guaranteed not to clobber an existing module of that
    name (:func:`_unique_module_name`) and reused across repeat imports of
    the SAME file (see :func:`_import_from_path`).

    Public counterpart of the ``_unique_module_name`` + ``_import_from_path``
    pair this module already used internally for its own filesystem-based
    discovery (:class:`CmdBuilder`, :func:`discover_commands`) -- exposed
    so an external command-provider package (``duho.runpath`` is the
    first, and so far only, consumer) can import a ``.py`` file the exact
    same way without reaching into either private helper directly.
    """
    with _IMPORT_LOCK:
        return _import_from_path(_unique_module_name(base_name), path)

is_class_command(obj)

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

Source code in src/duho/discovery/_command.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/_command.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/_providers.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))

unregister_command_provider(predicate, builder)

Remove a provider registered with :func:register_command_provider -- the exact (predicate, builder) pair (matched the same way list.remove would).

A no-op if that exact pair is not currently registered, so a caller does not need to track whether it already unregistered. Use it for test isolation or a plugin reloading itself.

Source code in src/duho/discovery/_providers.py

def unregister_command_provider(
    predicate: _ty.Callable[[_Path], bool],
    builder: _ty.Callable[[_Path, str], object],
) -> None:
    """Remove a provider registered with
    :func:`register_command_provider` -- the exact ``(predicate, builder)``
    pair (matched the same way ``list.remove`` would).

    A no-op if that exact pair is not currently registered, so a caller does
    not need to track whether it already unregistered. Use it for test
    isolation or a plugin reloading itself.
    """
    try:
        _PROVIDERS.remove((predicate, builder))
    except ValueError:
        pass