AC-164: type hints added

binstar_client/scripts/cli.py

@@ -1,166 +1,134 @@
-# pylint: disable=redefined-outer-name,unspecified-encoding,missing-class-docstring,missing-function-docstring
+# -*- coding: utf8 -*-
 
-"""
-Anaconda repository command line manager
-"""
+"""Anaconda repository command line manager."""
 
-from __future__ import print_function, unicode_literals
+from __future__ import annotations
 
+__all__ = ('main',)
+
+import argparse
+from importlib import metadata
 import logging
+import os
 import sys
-from argparse import ArgumentParser, RawDescriptionHelpFormatter
-from importlib.metadata import entry_points
-from logging.handlers import RotatingFileHandler
-from os import makedirs
-from os.path import join, exists, isfile
+import types
+import typing
 
-import urllib3
 from clyent import add_subparser_modules
 
-from binstar_client import __version__ as version
-from binstar_client import commands as command_module
+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 USER_LOGDIR
+from binstar_client.utils import logging_utils
+
 
 logger = logging.getLogger('binstar')
 
 
-def file_or_token(value):
+def file_or_token(value: str) -> str:
     """
-    If value is a file path and the file exists its contents are stripped and returned, otherwise value is returned.
-    """
-    if isfile(value):
-        with open(value) as file:
-            return file.read().strip()
+    Retrieve a token from input.
 
-    if any(char in value for char in '/\\.'):
+    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 _custom_excepthook(logger, show_traceback=False):
-    def excepthook(exc_type, exc_value, exc_traceback):
-        if issubclass(exc_type, KeyboardInterrupt):
-            return
-
-        if show_traceback:
-            logger.error('', exc_info=(exc_type, exc_value, exc_traceback))
-        else:
-            logger.error('%s', exc_value)
-
-    return excepthook
-
-
-class ConsoleFormatter(logging.Formatter):
-    def format(self, record):
-        fmt = '%(message)s' if record.levelno == logging.INFO \
-            else '[%(levelname)s] %(message)s'
-        self._style._fmt = fmt  # pylint: disable=protected-access
-        return super().format(record)
-
-
-def _setup_logging(logger, log_level=logging.INFO, show_traceback=False, disable_ssl_warnings=False):
-    logger.setLevel(logging.DEBUG)
-
-    if not exists(USER_LOGDIR):
-        makedirs(USER_LOGDIR)
-
-    log_file = join(USER_LOGDIR, 'cli.log')
-
-    file_handler = RotatingFileHandler(log_file, maxBytes=10 * (1024 ** 2), backupCount=5)
-    file_handler.setLevel(logging.DEBUG)
-
-    console_handler = logging.StreamHandler()
-    console_handler.setLevel(log_level)
-
-    console_handler.setFormatter(ConsoleFormatter())
-    file_handler.setFormatter(logging.Formatter('%(asctime)s %(levelname)-8s %(name)-15s %(message)s'))
-
-    logger.addHandler(console_handler)
-    logger.addHandler(file_handler)
-
-    sys.excepthook = _custom_excepthook(logger, show_traceback=show_traceback)
-
-    if disable_ssl_warnings:
-        urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)
-
-
-def add_default_arguments(parser, version=None):
-    output_group = parser.add_argument_group('output')
-    output_group.add_argument('--disable-ssl-warnings', action='store_true', default=False,
-                              help='Disable SSL warnings (default: %(default)s)')
-    output_group.add_argument('--show-traceback', action='store_true',
-                              help='Show the full traceback for chalmers user errors (default: %(default)s)')
-    output_group.add_argument('-v', '--verbose',
-                              action='store_const', help='print debug information on the console',
-                              dest='log_level',
-                              default=logging.INFO, const=logging.DEBUG)
-    output_group.add_argument('-q', '--quiet',
-                              action='store_const', help='Only show warnings or errors on the console',
-                              dest='log_level', const=logging.WARNING)
-
-    if version:
-        parser.add_argument('-V', '--version', action='version',
-                            version='%%(prog)s Command line client (version %s)' % (version,))
-
-
-def binstar_main(sub_command_module, args=None, exit=True,  # pylint: disable=redefined-builtin,too-many-arguments
-                 description=None, version=None, epilog=None):
-    parser = ArgumentParser(description=description, epilog=epilog,
-                            formatter_class=RawDescriptionHelpFormatter)
-
-    add_default_arguments(parser, version)
-    bgroup = parser.add_argument_group('anaconda-client options')
-    bgroup.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')
-    bgroup.add_argument('-s', '--site',
-                        help='select the anaconda-client site to use', default=None)
+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')
 
-    args = parser.parse_args(args)
+    arguments: argparse.Namespace = parser.parse_args(args)
 
-    _setup_logging(logger, log_level=args.log_level, show_traceback=args.show_traceback,
-                   disable_ssl_warnings=args.disable_ssl_warnings)
+    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 not hasattr(args, 'main'):
-                parser.error(
-                    'A sub command must be given. To show all available sub commands, run:\n\n\t anaconda -h\n',
-                )
-            return args.main(args)
+            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 not sys.stdin.isatty() or args.token:
-                # Don't try the interactive login
-                # Just exit
-                raise
-
+            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(args)
-            return args.main(args)
+            interactive_login(arguments)
+            return arguments.main(arguments)
     except errors.ShowHelp as error:
-        args.sub_parser.print_help()
-        if exit:
+        arguments.sub_parser.print_help()
+        if exit_:
             raise SystemExit(1) from error
         return 1
+    return 0  # type: ignore
 
 
-def _load_main_plugin():
+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 = 'anaconda_cli.main'
+    plugin_group_name: typing.Final[str] = 'anaconda_cli.main'
 
     # 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 <= 9:
-        plugin_mains = entry_points().get(plugin_group_name, [])
+    plugin_mains: typing.List[metadata.EntryPoint]
+    if sys.version_info.major == 3 and sys.version_info.minor < 10:
+        plugin_mains = metadata.entry_points().get(plugin_group_name, [])
     else:
-        plugin_mains = entry_points().select(group=plugin_group_name)
+        plugin_mains = metadata.entry_points().select(group=plugin_group_name)  # type: ignore
 
     if len(plugin_mains) > 1:
         raise EnvironmentError(
@@ -178,13 +146,20 @@ def _load_main_plugin():
     return None
 
 
-def main(args=None, _exit=True, allow_plugin_main=True):
-    plugged_in_main = _load_main_plugin()
-    if allow_plugin_main and plugged_in_main is not None:
-        plugged_in_main()
-    else:
-        binstar_main(command_module, args, _exit,
-                     description=__doc__, version=version)
+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:
+        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__':

binstar_client/utils/logging_utils.py

@@ -0,0 +1,82 @@
+# -*- coding: utf8 -*-
+
+"""Utilities to configure logging for the application."""
+
+from __future__ import annotations
+
+__all__ = ['setup_logging']
+
+import logging.handlers
+import os
+import sys
+import types
+import typing
+
+import urllib3.exceptions
+
+from . import config
+
+
+def _custom_excepthook(
+        logger: logging.Logger,
+        show_traceback: bool = False,
+) -> typing.Callable[[typing.Type[BaseException], BaseException, typing.Optional[types.TracebackType]], None]:
+    """Generate custom exception hook to log captured exceptions."""
+    def excepthook(
+            exc_type: typing.Type[BaseException],
+            exc_value: BaseException,
+            exc_traceback: typing.Optional[types.TracebackType],
+    ) -> None:
+        if issubclass(exc_type, KeyboardInterrupt):
+            return
+        if show_traceback:
+            logger.error('', exc_info=(exc_type, exc_value, exc_traceback))
+        else:
+            logger.error('%s', exc_value)
+
+    return excepthook
+
+
+class ConsoleFormatter(logging.Formatter):
+    """Custom logging formatter."""
+
+    FORMAT_DEFAULT: typing.Final[str] = '[%(levelname)s] %(message)s'
+    FORMAT_CUSTOM: typing.Final[typing.Mapping[int, str]] = {logging.INFO: '%(message)s'}
+
+    def format(self, record: logging.LogRecord) -> str:
+        """Format log record before printing it."""
+        # pylint: disable=protected-access
+        self._style._fmt = self.FORMAT_CUSTOM.get(record.levelno, self.FORMAT_DEFAULT)
+        return super().format(record)
+
+
+def setup_logging(
+        logger: logging.Logger,
+        log_level: int = logging.INFO,
+        show_traceback: bool = False,
+        disable_ssl_warnings: bool = False
+) -> None:
+    """Configure logging for the application."""
+    logger.setLevel(logging.DEBUG)
+
+    os.makedirs(config.USER_LOGDIR, exist_ok=True)
+    log_file: str = os.path.join(config.USER_LOGDIR, 'cli.log')
+
+    file_handler: logging.Handler = logging.handlers.RotatingFileHandler(
+        log_file,
+        maxBytes=10 * (1024 ** 2),
+        backupCount=5,
+    )
+    file_handler.setLevel(logging.DEBUG)
+    file_handler.setFormatter(logging.Formatter('%(asctime)s %(levelname)-8s %(name)-15s %(message)s'))
+    logger.addHandler(file_handler)
+
+    console_handler: logging.Handler = logging.StreamHandler()
+    console_handler.setLevel(log_level)
+    console_handler.setFormatter(ConsoleFormatter())
+    logger.addHandler(console_handler)
+
+    sys.excepthook = _custom_excepthook(logger, show_traceback=show_traceback)
+
+    if disable_ssl_warnings:
+        urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)

setup.cfg

@@ -18,7 +18,7 @@ ignore_errors = False
 ignore_missing_imports = True
 namespace_packages = True
 no_implicit_optional = True
-python_version = 3.7
+python_version = 3.8
 strict_optional = True
 warn_no_return = True
 warn_redundant_casts = True

tests/fixture.py

@@ -30,7 +30,7 @@ def setUp(self):
         self.store_token_patch = mock.patch('binstar_client.utils.config.store_token')
         self.store_token = self.store_token_patch.start()
 
-        self.setup_logging_patch = mock.patch('binstar_client.scripts.cli._setup_logging')
+        self.setup_logging_patch = mock.patch('binstar_client.utils.logging_utils.setup_logging')
         self.setup_logging_patch.start()
 
         self.logger = logger = logging.getLogger('binstar')