pyapp-org/pyapp

View on GitHub
src/pyapp/app/__init__.py

Summary

Maintainability
C
1 day
Test Coverage
"""
Application
~~~~~~~~~~~

*Application with bindings for commands*

The application object handles all the initial configuration to set up the
run-time environment.

Quick demo::

    >>> from pyapp.app import CliApplication

    >>> app = CliApplication()

    >>> @app.command()
    >>> def hello(*, verbose: bool):
    ...     if verbose:
    ...         print("Being verbose!")
    ...     print("Hello")

    >>> if __name__ == '__main__':
    ...     app.dispatch()


This example provides an application with a command `hello` that takes an
optional `verbose` flag. The framework also provides help, configures and loads
settings (using :py:mod:`pyapp.conf`), an interface to the checks framework
and configures the Python logging framework.

There are however a few more things that are required to get this going. The
:py:class:`CliApplication` class expects a certain structure of your
application to allow for it's (customisable) defaults to be applied.

Your application can have one of two structures

An application::

    my_app/__init__.py          # Include a __version__ variable
           __main__.py          # This is where the quick demo is located
           default_settings.py  # The default settings file
           checks.py            # Optional checks file


A single script::

    my_app.py                   # A script that contains the `CliApplication`


Generation of CLI from command Signature
~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~

.. versionadded:: 4.4

As of pyApp 4.4 command functions can supply all required arguments in the function
signature.

As an example consider the command function:

.. code-block:: python

    @app.command
    def my_command(
        arg1: str,
        *,
        arg2: bool= Arg(help="Enable the argilizer"),
        arg3: int = 42,
        arg4: str = Arg("-a", choices=("foo", "bar"), default="foo")
    ):
        ...

This translates into the following on the CLI:

.. code-block:: shell

    > python -m my_app my_command --help
    usage: my_app my_command [-h] ARG1 [--arg2] [--arg3 ARG3]
                             [--arg4 {foo,bar}]

    positional arguments:
      ARG1

    optional arguments:
      -h, --help  show this help message and exit
      --arg2    Enable the argilizer
      --arg3
      --arg4 {foo,bar}


The following types are supported as arguments:

    - Basic types eg int, str, float, this covers any type that can be provided
      to argparse in the type field.

    - bool, this is made into an argparse `store_true` action.

    - Enum types using the pyApp EnumAction.

    - Generic types
        - Mapping/Dict as well as a basic dict for Key/Value pairs

        - Sequence/List for typed sequences, ``nargs="+"`` for positional arguments
          of ``action="append"`` for optional.

        - Tuple for typed sequences of a fixed size eg ``nargs=len(tuple)``. Only
          the first type is used, the others are ignored.

    - FileType from ``argparse``.

.. tip:: Too get access to the parse results from `argparse` provide a vairable
    with the type ``pyapp.app.CommandOptions``.


CliApplication
--------------

.. autoclass:: CliApplication
    :members: command, create_command_group, default, dispatch

Events
~~~~~~

CliApplication generates the following events, all methods are provided with the
``argparse`` namespace.

+--------------------------------------------------------------+----------------------------------------------------+
| ``pre_dispatch[[argparse.Namespace], None]``                 | Generated before command dispatch is called        |
+--------------------------------------------------------------+----------------------------------------------------+
| ``post_dispatch[[Optional[int], argparse.Namespace], None]`` | Generated after command dispatch returns without   |
|                                                              | error includes the return code if one is provided. |
+--------------------------------------------------------------+----------------------------------------------------+
| ``dispatch_error[[Exception, argparse.Namespace], None]``    | Generated when an exception is raised in a command |
|                                                              | function before the standard exception reporting.  |
+--------------------------------------------------------------+----------------------------------------------------+


Arguments
---------

.. automodule:: pyapp.app.arguments


Argument Types
--------------

.. automodule:: pyapp.app.argument_types


Argument Actions
----------------

.. automodule:: pyapp.app.argument_actions

"""

import argparse
import io
import logging.config
import os
import sys
import warnings
from argparse import ArgumentParser
from argparse import Namespace as CommandOptions
from typing import Callable, Optional, Sequence

import argcomplete
import colorama

from .. import conf, extensions, feature_flags
from ..app import builtin_handlers
from ..events import Event
from ..exceptions import ApplicationExit
from ..injection import register_factory
from ..utils.inspect import import_root_module
from . import init_logger
from .argument_actions import *  # noqa
from .arguments import *  # noqa
from .logging_formatter import ColourFormatter

logger = logging.getLogger(__name__)


def _key_help(key: str) -> str:
    """Formats a key value from environment vars."""
    if key in os.environ:
        return f"{key} [{os.environ[key]}]"
    return key


class CliApplication(CommandGroup):  # noqa: F405
    """Application interface that provides a CLI interface.

    :param root_module: The root module for this application (used for discovery of
        other modules)
    :param prog: Name of your application; defaults to `sys.argv[0]`
    :param description: A description of your application for `--help`.
    :param version: Specify a specific version; defaults to
        `getattr(root_module, '__version__')`
    :param ext_allow_list: Sequence of extension names or globs that are allowed;
        default is `None` or all extensions.
    :param ext_block_list: Sequence of extension names or globs that are blocked;
        default is `None` or no blocking.
    :param application_settings: The default settings for this application;
        defaults to `root_module.default_settings`
    :param application_checks: Location of application checks file; defaults to
        `root_module.checks` if it exists.
    :param env_settings_key: Key used to define settings file in environment.
    :param env_loglevel_key: Key used to define log level in environment

    """

    default_log_handler = logging.StreamHandler(sys.stderr)
    """Log handler applied by default to root logger."""

    default_log_formatter = logging.Formatter(
        "%(asctime)s | %(levelname)s | %(name)s | %(message)s"
    )
    """Log formatter applied by default to root logger handler."""

    default_color_log_formatter = ColourFormatter(
        f"{colorama.Fore.YELLOW}%(asctime)s{colorama.Fore.RESET} "
        f"%(clevelname)s "
        f"{colorama.Fore.LIGHTBLUE_EX}%(name)s{colorama.Fore.RESET} "
        f"%(message)s"
    )
    """Log formatter applied by default to root logger handler."""

    env_settings_key = conf.DEFAULT_ENV_KEY
    """Key used to define settings file in environment."""

    env_loglevel_key = "PYAPP_LOGLEVEL"
    """Key used to define log level in environment."""

    additional_handlers = (
        builtin_handlers.checks,
        builtin_handlers.extensions,
        builtin_handlers.settings,
    )
    """Handlers to be added when builtin handlers are registered."""

    # Events
    pre_dispatch = Event[Callable[[argparse.Namespace], None]]()
    post_dispatch = Event[Callable[[Optional[int], argparse.Namespace], None]]()
    dispatch_error = Event[Callable[[Exception, argparse.Namespace], None]]()

    def __init__(  # noqa: PLR0913
        self,
        root_module=None,
        *,
        prog: str = None,
        description: str = None,
        epilog: str = None,
        version: str = None,
        ext_white_list: Sequence[str] = None,
        ext_allow_list: Sequence[str] = None,
        ext_block_list: Sequence[str] = None,
        application_settings: str = None,
        application_checks: str = None,
        env_settings_key: str = None,
        env_loglevel_key: str = None,
    ):
        root_module = root_module or import_root_module()
        self.root_module = root_module
        super().__init__(ArgumentParser(prog, description=description, epilog=epilog))
        self.application_version = version or getattr(
            root_module, "__version__", "Unknown"
        )
        self.ext_allow_list = ext_allow_list
        if ext_white_list:
            warnings.warn(
                "ext_white_list is deprecated, use ext_allow_list",
                DeprecationWarning,
                stacklevel=2,
            )
            self.ext_allow_list = ext_white_list
        self.ext_block_list = ext_block_list

        # Determine application settings (disable for standalone scripts)
        if application_settings is None and root_module.__name__ != "__main__":
            application_settings = f"{root_module.__name__}.default_settings"
        self.application_settings = application_settings

        # Determine application checks
        if application_checks is None:
            application_checks = f"{root_module.__name__}.checks"
        self.application_checks = application_checks

        # Override default value
        if env_settings_key is not None:
            self.env_settings_key = env_settings_key
        if env_loglevel_key is not None:
            self.env_loglevel_key = env_loglevel_key

        # Configure Logging as early as possible
        self._init_logger = init_logger.InitHandler(self.default_log_handler)
        self.pre_configure_logging()

        self._init_parser()
        self.register_builtin_handlers()

    def __repr__(self) -> str:
        return f"{type(self).__name__}(<module {self.root_module.__name__}>)"

    def __str__(self) -> str:
        return self.application_summary

    @property
    def application_name(self) -> str:
        """Name of the application."""
        return self.parser.prog

    @property
    def application_summary(self) -> str:
        """Summary of the application, name version and description."""
        description = self.parser.description
        if description:
            return f"{self.application_name} version {self.application_version} - {description}"
        return f"{self.application_name} version {self.application_version}"

    def _init_parser(self):
        # Create argument parser
        self.argument(
            "--settings",
            help="Settings to load; either a Python module or settings URL. "
            f"Defaults to the env variable: {_key_help(self.env_settings_key)}",
        )
        self.argument(
            "--version",
            action="version",
            version=f"%(prog)s version: {self.application_version}",
        )
        self.argument(
            "--nocolor",
            "--nocolour",
            dest="no_color",
            action="store_true",
            help="Disable colour output.",
        )

        # Log configuration
        arg_group = self.argument_group(
            title="logging arguments", description="Customise log output"
        )
        arg_group.add_argument(
            "--log-level",
            default=os.environ.get(self.env_loglevel_key, "DEFAULT"),
            choices=("DEBUG", "INFO", "WARN", "ERROR", "CRITICAL"),
            help="Specify the log level to be used. "
            f"Defaults to env variable: {_key_help(self.env_loglevel_key)}",
        )
        # arg_group.add_argument(
        #     "--log-file",
        #     type=FileType(mode="w", encoding="UTF-8"),
        #     help="Optionally override log file output.",
        # )
        arg_group.add_argument(
            "--log-color",
            "--log-colour",
            dest="log_color",
            default=None,
            action="store_true",
            help="Force coloured output from logger (on console).",
        )
        arg_group.add_argument(
            "--log-nocolor",
            "--log-nocolour",
            dest="log_color",
            action="store_false",
            help="Disable coloured output from logger (on console).",
        )

        # Global check values
        arg_group = self.argument_group(
            title="check arguments", description="Enable and configure run-time checks"
        )
        arg_group.add_argument(
            "--checks",
            dest="checks_on_startup",
            action="store_true",
            help="Run checks on startup, any serious error will result "
            "in the application terminating.",
        )
        arg_group.add_argument(
            "--checks-level",
            dest="checks_message_level",
            default="INFO",
            choices=("DEBUG", "INFO", "WARN", "ERROR", "CRITICAL"),
            help="Minimum level of check message to display",
        )

        # Feature flags
        arg_group = self.argument_group(
            title="feature flags", description="Enable/Disable feature flags"
        )
        arg_group.add_argument(
            "--enable-flag",
            dest="enable_feature_flags",
            action="append",
            help="Enable a named feature flag; this argument can be used multiple times",
        )
        arg_group.add_argument(
            "--disable-flag",
            dest="disable_feature_flags",
            action="append",
            help="Disable a named feature flag; this argument can be used multiple times",
        )

    def register_builtin_handlers(self):
        """Register any built in handlers."""
        # Register any additional handlers
        for additional_handler in self.additional_handlers:
            additional_handler(self)

    def pre_configure_logging(self):
        """Set some default logging so settings are logged.

        The main logging configuration is in settings leaving us with a chicken
        and egg situation.
        """
        self.default_log_handler.formatter = self.default_log_formatter

        # Apply handler to root logger
        logging.root.setLevel(logging.DEBUG)
        logging.root.handlers = [self._init_logger]

    @staticmethod
    def register_factories():
        """Register any abstract interface factories."""
        # pylint: disable=import-outside-toplevel
        from asyncio import AbstractEventLoop, get_event_loop

        register_factory(AbstractEventLoop, get_event_loop)

    def load_extensions(self):
        """Load/Configure extensions."""
        entry_points = extensions.ExtensionEntryPoints(
            self.ext_allow_list, self.ext_block_list
        )
        extensions.registry.load_from(entry_points.extensions())
        extensions.registry.register_commands(self)

    def configure_settings(self, opts: CommandOptions):
        """Configure settings container."""
        application_settings = list(extensions.registry.default_settings)
        if self.application_settings:
            application_settings.append(self.application_settings)

        conf.settings.configure(
            application_settings, opts.settings, env_settings_key=self.env_settings_key
        )

    @staticmethod
    def configure_feature_flags(opts: CommandOptions):
        """Configure feature flags cache."""
        if opts.enable_feature_flags:
            for flag in opts.enable_feature_flags:
                feature_flags.DEFAULT.set(flag, True)

        if opts.disable_feature_flags:
            for flag in opts.disable_feature_flags:
                feature_flags.DEFAULT.set(flag, False)

    def get_log_formatter(self, log_color) -> logging.Formatter:
        """Get log formatter."""
        log_handler = self.default_log_handler

        # Auto-detect colour mode
        if (
            log_color is None
            and isinstance(log_handler, logging.StreamHandler)
            and hasattr(log_handler.stream, "isatty")
        ):
            log_color = log_handler.stream.isatty()

        # Enable colour if specified.
        if log_color:
            return self.default_color_log_formatter

        return self.default_log_formatter

    def configure_logging(self, opts: CommandOptions):
        """Configure the logging framework."""
        # Prevent duplicate runs
        if hasattr(self, "_init_logger"):
            self.default_log_handler.formatter = self.get_log_formatter(opts.log_color)

            if conf.settings.LOGGING:
                logger.info("Applying logging configuration.")

            # Replace root handler with the default handler
            logging.root.handlers.pop(0)
            logging.root.handlers.append(self.default_log_handler)

            if conf.settings.LOGGING:
                # Set a default version if not supplied by settings
                dict_config = conf.settings.LOGGING.copy()
                dict_config.setdefault("version", 1)
                logging.config.dictConfig(dict_config)

            # Configure root log level
            loglevel = opts.log_level
            if loglevel == "DEFAULT":
                handler = self.resolve_handler(opts)
                loglevel = getattr(handler, "loglevel", logging.INFO)
            logging.root.setLevel(loglevel)

            # Replay initial entries and remove
            self._init_logger.replay()
            del self._init_logger

    def checks_on_startup(self, opts: CommandOptions):
        """Run checks on startup."""
        # pylint: disable=import-outside-toplevel
        from pyapp.checks.report import execute_report

        if opts.checks_on_startup:
            out = io.StringIO()

            serious_error = execute_report(
                out,
                self.application_checks,
                opts.checks_message_level,
                verbose=True,
                header=f"Check report for {self.application_summary}",
            )
            if serious_error:
                logger.error("Check results:\n%s", out.getvalue())
                sys.exit(4)
            else:
                logger.info("Check results:\n%s", out.getvalue())

    def exception_report(self, exception: BaseException, opts: CommandOptions):
        """Generate a report for any unhandled exceptions caught by the framework."""
        logger.exception(
            "Un-handled exception %s caught executing handler: %s",
            exception,
            getattr(opts, self.handler_dest),
        )
        return False

    @staticmethod
    def logging_shutdown():
        """Call at shutdown to ensure logging is cleaned up."""
        logging.shutdown()

    def dispatch(self, args: Sequence[str] = None) -> None:
        """Dispatch command to registered handler."""
        logger.info("Starting %s", self.application_summary)

        # Initialisation phase
        _set_running_application(self)
        self.register_factories()
        self.load_extensions()

        # Parse arguments phase
        argcomplete.autocomplete(self.parser)
        opts = self.parser.parse_args(args)

        # Load settings and configure logger
        self.configure_settings(opts)
        self.configure_feature_flags(opts)
        self.configure_logging(opts)

        handler_name = getattr(opts, ":handler", None)
        if handler_name != "checks":
            self.checks_on_startup(opts)
        else:
            self.configure_settings(opts)

        extensions.registry.ready()

        # Dispatch to handler.
        self.pre_dispatch(opts)
        try:
            exit_code = self.dispatch_handler(opts)

        except Exception as ex:  # pylint: disable=broad-except
            self.dispatch_error(ex, opts)
            if not self.exception_report(ex, opts):
                raise

        except ApplicationExit as ex:
            if ex.message:
                print(f"\n\n{ex.message}", file=sys.stderr)
            raise

        except KeyboardInterrupt:
            print("\n\nInterrupted.", file=sys.stderr)
            sys.exit(2)

        else:
            # Provide exit code.
            self.post_dispatch(exit_code, opts)
            if exit_code:
                sys.exit(exit_code)

        finally:
            self.logging_shutdown()


CURRENT_APP: Optional[CliApplication] = None


def _set_running_application(app: CliApplication):
    global CURRENT_APP  # noqa: PLW0603
    CURRENT_APP = app


def get_running_application() -> CliApplication:
    """Get the current running application instance."""
    return CURRENT_APP