# -*- coding: utf8 -*- """Anaconda repository command line manager.""" from __future__ import annotations __all__ = ('main',) import argparse from importlib import metadata import json import logging import os import pkgutil import sys import types import typing from binstar_client import __version__ from binstar_client import commands from binstar_client import errors from binstar_client.commands.login import interactive_login from binstar_client.utils import logging_utils logger = logging.getLogger('binstar') def _get_entry_points(group: str) -> typing.List[metadata.EntryPoint]: # The API was changed in Python 3.10, see https://docs.python.org/3/library/importlib.metadata.html#entry-points if sys.version_info.major == 3 and sys.version_info.minor < 10: return metadata.entry_points().get(group, []) return metadata.entry_points().select(group=group) # type: ignore def file_or_token(value: str) -> str: """ Retrieve a token from input. If :code:`value` is a path to a valid file - content of this file will be returned. Otherwise - value itself is returned. """ if os.path.isfile(value): stream: typing.TextIO with open(value, 'rt', encoding='utf8') as stream: result: str = stream.read(8193) if len(result) > 8192: raise ValueError('file is too large for a token') return result.strip() if not set('/\\.').isdisjoint(value): # This chars will never be in a token value, but may be in a path # The error message will be handled by the parser raise ValueError() return value def _json_action(action): # pylint: disable=protected-access # intentional access of argparse object members a_data = dict(action._get_kwargs()) if a_data.get('help'): a_data['help'] = a_data['help'] % a_data if isinstance(action, argparse._SubParsersAction): a_data.pop('choices', None) choices = {} for choice in action._get_subactions(): choices[choice.dest] = choice.help a_data['choices'] = choices reg = {value: key for key, value in action.container._registries['action'].items()} a_data['action'] = reg.get(type(action), type(action).__name__) if a_data['action'] == 'store' and not a_data.get('metavar'): a_data['metavar'] = action.dest.upper() a_data.pop('type', None) a_data.pop('default', None) return a_data def _json_group(group): # pylint: disable=protected-access # intentional access of argparse object members grp_data = { 'description': group.description, 'title': group.title, 'actions': [_json_action(action) for action in group._group_actions if action.help != argparse.SUPPRESS], } if group._action_groups: grp_data['groups'] = [_json_group(group) for group in group._action_groups] return grp_data class _JSONHelp(argparse.Action): # pylint: disable-next=redefined-builtin def __init__(self, option_strings, dest, nargs=0, help=argparse.SUPPRESS, **kwargs): argparse.Action.__init__(self, option_strings, dest, nargs=nargs, help=help, **kwargs) def __call__(self, parser, namespace, values, option_string=None): # pylint: disable=protected-access # intentional access of argparse object members self.nargs = 0 docs = { 'prog': parser.prog, 'usage': parser.format_usage()[7:], 'description': parser.description, 'epilog': parser.epilog, } docs['groups'] = [] for group in parser._action_groups: if group._group_actions: docs['groups'].append(_json_group(group)) json.dump(docs, sys.stdout, indent=2) raise SystemExit(0) def _get_sub_command_names(module): return [name for _, name, _ in pkgutil.iter_modules([os.path.dirname(module.__file__)]) if not name.startswith('_')] def _get_sub_commands(module): names = _get_sub_command_names(module) this_module = __import__(module.__package__ or module.__name__, fromlist=names) for name in names: yield getattr(this_module, name) def _add_subparser_modules(parser, module=None, entry_point_name=None): subparsers = parser.add_subparsers(title='Commands', metavar='') if module: # LOAD sub parsers from module for command_module in _get_sub_commands(module): command_module.add_parser(subparsers) if entry_point_name: # LOAD sub parsers from setup.py entry_point for entry_point in _get_entry_points(entry_point_name): add_parser = entry_point.load() add_parser(subparsers) for key, sub_parser in subparsers.choices.items(): sub_parser.set_defaults(sub_command_name=key) sub_parser.add_argument('--json-help', action=_JSONHelp) def binstar_main( sub_command_module: types.ModuleType, args: typing.Optional[typing.Sequence[str]] = None, exit_: bool = True, ) -> int: """Run `anaconda-client` cli utility.""" parser: argparse.ArgumentParser = argparse.ArgumentParser( description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter, ) group = parser.add_argument_group('output') group.add_argument( '--disable-ssl-warnings', action='store_true', default=False, help='Disable SSL warnings (default: %(default)s)', ) group.add_argument( '--show-traceback', action='store_true', help='Show the full traceback for chalmers user errors (default: %(default)s)', ) group.add_argument( '-v', '--verbose', action='store_const', dest='log_level', default=logging.INFO, const=logging.DEBUG, help='print debug information on the console', ) group.add_argument( '-q', '--quiet', action='store_const', dest='log_level', const=logging.WARNING, help='Only show warnings or errors on the console', ) group = parser.add_argument_group('anaconda-client options') group.add_argument( '-t', '--token', type=file_or_token, help='Authentication token to use. May be a token or a path to a file containing a token', ) group.add_argument('-s', '--site', default=None, help='select the anaconda-client site to use') if __version__: parser.add_argument( '-V', '--version', action='version', version=f'%(prog)s Command line client (version {__version__})', ) _add_subparser_modules(parser, sub_command_module, 'conda_server.subcommand') arguments: argparse.Namespace = parser.parse_args(args) logging_utils.setup_logging( logger, log_level=arguments.log_level, show_traceback=arguments.show_traceback, disable_ssl_warnings=arguments.disable_ssl_warnings, ) try: try: if hasattr(arguments, 'main'): return arguments.main(arguments) parser.error('A sub command must be given. To show all available sub commands, run:\n\n\t anaconda -h\n') except errors.Unauthorized: if arguments.token or (not sys.stdin.isatty()): raise # Don't try the interactive login, just exit logger.info('The action you are performing requires authentication, please sign in:') interactive_login(arguments) return arguments.main(arguments) except errors.ShowHelp as error: arguments.sub_parser.print_help() if exit_: raise SystemExit(1) from error return 1 return 0 # type: ignore def _load_main_plugin() -> typing.Optional[typing.Callable[[], typing.Any]]: """Allow loading a new CLI main entrypoint via plugin mechanisms. There can only be one.""" plugin_group_name: typing.Final[str] = 'anaconda_cli.main' plugin_mains: typing.List[metadata.EntryPoint] = _get_entry_points(plugin_group_name) if len(plugin_mains) > 1: raise EnvironmentError( 'More than one `anaconda_cli.main` plugin is installed. Please ensure only one ' 'of the following packages are installed:\n\n' + '\n'.join(f' * {ep.value}' for ep in plugin_mains) ) if plugin_mains: # The `.load()` function returns a callable, which is defined inside the package implementing the plugin # e.g. in pyproject.toml, where my_plugin_library.cli.main is the callable entrypoint function # [project.entry-points."anaconda_cli.main"] # anaconda = "my_plugin_library.cli:main" return tuple(plugin_mains)[0].load() return None def main( args: typing.Optional[typing.Sequence[str]] = None, *, exit_: bool = True, allow_plugin_main: bool = True, ) -> None: """Entrypoint for CLI interface of `anaconda`.""" if allow_plugin_main and (not os.environ.get('ANACONDA_CLIENT_FORCE_STANDALONE', '')): plugged_in_main: typing.Optional[typing.Callable[[], typing.Any]] = _load_main_plugin() if plugged_in_main is not None: plugged_in_main() return binstar_main(commands, args, exit_) if __name__ == '__main__': main()