# Licensed under the Apache License, Version 2.0 (the "License"); you may # not use this file except in compliance with the License. You may obtain # a copy of the License at # # http://www.apache.org/licenses/LICENSE-2.0 # # Unless required by applicable law or agreed to in writing, software # distributed under the License is distributed on an "AS IS" BASIS, WITHOUT # WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the # License for the specific language governing permissions and limitations # under the License. """Discover and lookup command plugins.""" import collections.abc import importlib.metadata import logging from typing import TypeAlias import stevedore from cliff import command LOG = logging.getLogger(__name__) def _get_commands_by_partial_name( args: list[str], commands: list[str] ) -> list[str]: n = len(args) candidates = [] for command_name in commands: command_parts = command_name.split() if len(command_parts) != n: continue if all(command_parts[i].startswith(args[i]) for i in range(n)): candidates.append(command_name) return candidates class EntryPointWrapper: """An entrypoint-like object. Wrap up a command class already imported to make it look like a plugin. """ def __init__( self, name: str, command_class: type[command.Command] ) -> None: self.name = name self.command_class = command_class def load(self) -> type[command.Command]: return self.command_class @property def value(self) -> str: # fake entry point target for logging purposes return ':'.join( [self.command_class.__module__, self.command_class.__name__] ) EntryPointT: TypeAlias = EntryPointWrapper | importlib.metadata.EntryPoint class CommandManager: """Discovers commands and handles lookup based on argv data. :param namespace: String containing the entrypoint namespace for the plugins to be loaded. For example, ``'cliff.formatter.list'``. If provided, commands will be loaded from this namespace initially, though it remains possible to load from additional namespaces with :meth:`CommandManager.load_commands`. :param convert_underscores: Whether cliff should convert underscores to spaces in entry_point commands. :param ignored_modules: A list of module names to ignore when loading commands. This will be matched partial, so be as specific as needed. """ def __init__( self, namespace: str | None = None, convert_underscores: bool = True, *, ignored_modules: collections.abc.Iterable[str] | None = None, ) -> None: self.namespace = namespace self.convert_underscores = convert_underscores self.ignored_modules = ignored_modules or () self.commands: dict[str, EntryPointT] = {} self._legacy: dict[str, str] = {} self.group_list: list[str] = [] self._load_commands() def _load_commands(self) -> None: # NOTE(jamielennox): kept for compatibility. if self.namespace: self.load_commands(self.namespace) @staticmethod def _is_module_ignored( module_name: str, ignored_modules: collections.abc.Iterable[str] ) -> bool: # given module_name = 'foo.bar.baz:wow', we expect to match any of # the following ignores: foo.bar.baz:wow, foo.bar.baz, foo.bar, foo while True: if module_name in ignored_modules: return True split_index = max(module_name.rfind(':'), module_name.rfind('.')) if split_index == -1: break module_name = module_name[:split_index] return False def load_commands(self, namespace: str) -> None: """Load all the commands from an entrypoint :param namespace: The namespace to load commands from. :returns: None """ self.group_list.append(namespace) em: stevedore.ExtensionManager[command.Command] # note that we don't invoke stevedore's conflict resolver functionality # because that is namespace specific and we care about conflicts # regardless of the namespace em = stevedore.ExtensionManager(namespace) for ext in em: LOG.debug('found command %r', ext.name) if self._is_module_ignored(ext.module_name, self.ignored_modules): LOG.debug( 'extension found in ignored module %r: skipping', ext.module_name, ) continue cmd_name = ( ext.name.replace('_', ' ') if self.convert_underscores else ext.name ) if cmd_name in self.commands: # Attention, programmers: If you arrived here attempting to # resolve a warning in your application then you have a command # with the same name either defined more than once in the same # application (a typo?) or in multiple packages (for example, # a package that adds plugins to your applications). The latter # can often happen if you e.g. move a command from a plugin to # the core application. In this situation, you should add the # old location to 'ignored_modules' and the plugin will now be # ignored. LOG.warning( 'found duplicate implementations of the %(name)r command ' 'in the following modules: %(modules)s: this is likely ' 'programmer error and should be reported as a bug to the ' 'relevant project(s)', { 'name': cmd_name, 'modules': ', '.join( [ self.commands[cmd_name].value, ext.entry_point.value, ] ), }, ) self.commands[cmd_name] = ext.entry_point def __iter__( self, ) -> collections.abc.Iterator[tuple[str, EntryPointT]]: return iter(self.commands.items()) def add_command( self, name: str, command_class: type[command.Command] ) -> None: self.commands[name] = EntryPointWrapper(name, command_class) def add_legacy_command(self, old_name: str, new_name: str) -> None: """Map an old command name to the new name. :param old_name: The old command name. :type old_name: str :param new_name: The new command name. :type new_name: str """ self._legacy[old_name] = new_name def find_command( self, argv: list[str] ) -> tuple[type[command.Command], str, list[str]]: """Given an argument list, find a command and return the processor and any remaining arguments. """ start = self._get_last_possible_command_index(argv) for i in range(start, 0, -1): name = ' '.join(argv[:i]) search_args = argv[i:] # The legacy command handling may modify name, so remember # the value we actually found in argv so we can return it. return_name = name # Convert the legacy command name to its new name. if name in self._legacy: name = self._legacy[name] found = None if name in self.commands: found = name else: candidates = _get_commands_by_partial_name( argv[:i], list(self.commands.keys()) ) if len(candidates) == 1: found = candidates[0] if found: cmd_ep = self.commands[found] cmd_factory = cmd_ep.load() return (cmd_factory, return_name, search_args) else: raise ValueError(f'Unknown command {argv!r}') def _get_last_possible_command_index(self, argv: list[str]) -> int: """Returns the index after the last argument in argv that can be a command word """ for i, arg in enumerate(argv): if arg.startswith('-'): return i return len(argv) def add_command_group(self, group: str | None = None) -> None: """Adds another group of command entrypoints""" if group: self.load_commands(group) def get_command_groups(self) -> list[str]: """Returns a list of the loaded command groups""" return self.group_list def get_command_names(self, group: str | None = None) -> list[str]: """Returns a list of commands loaded for the specified group""" group_list: list[str] = [] if group is not None: em: stevedore.ExtensionManager[command.Command] em = stevedore.ExtensionManager(group) for ep in em: cmd_name = ( ep.name.replace('_', ' ') if self.convert_underscores else ep.name ) group_list.append(cmd_name) return group_list return list(self.commands.keys())