Source code for craft_application.application

# This file is part of craft_application.
#
# Copyright 2023-2024 Canonical Ltd.
#
# This program is free software: you can redistribute it and/or modify it
# under the terms of the GNU Lesser General Public License version 3, as
# published by the Free Software Foundation.
#
# This program is distributed in the hope that it will be useful, but WITHOUT
# ANY WARRANTY; without even the implied warranties of MERCHANTABILITY,
# SATISFACTORY QUALITY, or FITNESS FOR A PARTICULAR PURPOSE.
# See the GNU Lesser General Public License for more details.
#
# You should have received a copy of the GNU Lesser General Public License along
# with this program.  If not, see <http://www.gnu.org/licenses/>.
"""Main application classes for a craft-application."""

from __future__ import annotations

import importlib
import os
import pathlib
import signal
import sys
import traceback
from dataclasses import dataclass, field
from functools import cached_property
from importlib import metadata
from typing import TYPE_CHECKING, Annotated, Any, cast, final

import annotated_types
import craft_cli
import craft_platforms
from platformdirs import user_cache_path

from craft_application import _config, commands, errors, models, util
from craft_application.errors import PathInvalidError
from craft_application.util.logging import handle_runtime_error

if TYPE_CHECKING:
    import argparse
    from collections.abc import Iterable, Sequence

    from craft_parts.plugins.plugins import PluginType

    from craft_application.services import service_factory
    from craft_application.util import ProServices

GLOBAL_VERSION = craft_cli.GlobalArgument(
    "version", "flag", "-V", "--version", "Show the application version and exit"
)

DEFAULT_CLI_LOGGERS = frozenset(
    {
        "craft_archives",
        "craft_parts",
        "craft_providers",
        "craft_store",
        "craft_application",
        "httpx",  # Used by craft-store
    }
)




@final
@dataclass(frozen=True)
class AppMetadata:
    """Metadata about a craft application."""

    name: str
    """The name of the application."""
    summary: str | None = None
    """A short summary of the application."""
    version: str = field(init=False)
    docs_url: str | None = None
    """The root URL for the app's documentation."""
    artifact_type: Annotated[
        str,
        annotated_types.IsAscii,
        annotated_types.LowerCase,
    ] = "artifact"
    """The name to refer to the output artifact for this app.

    This gets used in messages and should be an all lower-case single-word value, like
    ``snap`` or ``rock``. Defaults to ``artifact``.
    """
    source_ignore_patterns: list[str] = field(default_factory=list[str])
    managed_instance_project_path = pathlib.PurePosixPath("/root/project")
    project_variables: list[str] = field(
        default_factory=lambda: ["version", "summary", "description"]
    )
    """Fields that are adoptable using craftctl set."""
    mandatory_adoptable_fields: list[str] = field(
        default_factory=lambda: ["version", "summary", "description"]
    )
    """Fields that must either be in the YAML file or adopted with craftctl set."""
    ConfigModel: type[_config.ConfigModel] = _config.ConfigModel

    ProjectClass: type[models.Project] = models.Project
    """The project model to use for this app.

    Most applications will need to override this, but a very basic application could use
    the default model without modification.
    """
    supports_multi_base: bool = False
    always_repack: bool = (
        True  # Gating for https://github.com/canonical/craft-application/pull/810
    )
    check_supported_base: bool = False
    """Whether this application allows building on unsupported bases.

    When True, the app can build on a base even if it is end-of-life. Relevant apt
    repositories will be migrated to ``old-releases.ubuntu.com``. Currently only
    supports EOL Ubuntu releases.

    When False, the repositories are not migrated and base support is not checked.
    """

    enable_for_grammar: bool = False
    """Whether this application supports the 'for' variant of advanced grammar."""

    enable_pro_support: bool = False
    """Whether this application supports Ubuntu Pro services."""

    allow_git_build_root: bool = False
    """Whether to allow mounting the git working tree root as the build root.

    When True and the user enables the ``experimental_monorepo`` config option,
    the git root is mounted as /root/project in the managed instance rather than
    the project directory. The working directory inside the instance is adjusted
    to the project's subdirectory within the git root, giving the build access to
    code outside the project directory (e.g. shared libraries in a monorepo).

    Has no effect when the project directory is not inside a git repository,
    or when the project directory is the git root itself.
    """

    def __post_init__(self) -> None:
        setter = super().__setattr__

        # Try to determine the app version.
        try:
            # First, via the __version__ attribute on the app's main package.
            version = importlib.import_module(self.name).__version__
        except (AttributeError, ModuleNotFoundError):
            try:
                # If that fails, try via the installed metadata.
                version = metadata.version(self.name)
            except metadata.PackageNotFoundError:
                # If that fails too, default to "dev".
                version = "dev"

        setter("version", version)
        if self.summary is None:
            md = metadata.metadata(self.name)
            setter("summary", md["summary"])

    @property
    def versioned_docs_url(self) -> str | None:
        """The ``docs_url`` with the proper app version."""
        if self.docs_url is None:
            return None

        return util.render_doc_url(self.docs_url, self.version)





class Application:
    """Craft Application Builder.

    :ivar app: Metadata about this application
    :ivar services: A ServiceFactory for this application
    :param extra_loggers: Logger names to integrate with craft-cli beyond the defaults.

    """



    def __init__(
        self,
        app: AppMetadata,
        services: service_factory.ServiceFactory,
        *,
        extra_loggers: Iterable[str] = (),
    ) -> None:
        self.app = app
        self.services = services
        self._command_groups: list[craft_cli.CommandGroup] = []
        self._global_arguments: list[craft_cli.GlobalArgument] = [GLOBAL_VERSION]
        self._cli_loggers = DEFAULT_CLI_LOGGERS | set(extra_loggers)
        self._partitions: list[str] | None = None
        # Cached project object, allows only the first time we load the project
        # to specify things like the project directory.
        # This is set as a private attribute in order to discourage real application
        # implementations from accessing it directly. They should always use
        # ``get_project`` to access the project.
        self.__project: models.Project | None = None
        # Set a globally usable project directory for the application.
        # This may be overridden by specific application implementations.
        self.project_dir = pathlib.Path.cwd()
        # ProServices instance containing relevant Pro services specified by the user.
        # Storage of this instance may change in the future as we migrate Pro operations towards
        # an application service.
        self._pro_services: ProServices | None = None

        if self.is_managed():
            self._work_dir = pathlib.Path("/root")
        else:
            self._work_dir = pathlib.Path.cwd()


    @final
    def _load_plugins(self) -> None:
        """Load application plugins."""
        # https://packaging.python.org/en/latest/specifications/entry-points/#data-model
        for plugin_entry_point in metadata.entry_points(
            group="craft_application_plugins.application"
        ):
            craft_cli.emit.debug(f"Loading app plugin {plugin_entry_point.name}")
            try:
                app_plugin_module = plugin_entry_point.load()
                app_plugin_module.configure(self)
            except Exception:  # noqa: BLE001
                craft_cli.emit.progress(
                    f"Failed to load plugin {plugin_entry_point.name}",
                    permanent=True,
                )
                craft_cli.emit.debug(traceback.format_exc())

    @property
    def app_config(self) -> dict[str, Any]:
        """Get the configuration passed to dispatcher.load_command().

        This can generally be left as-is. It's strongly recommended that if you are
        overriding it, you begin with ``config = super().app_config`` and update the
        dictionary from there.
        """
        return {
            "app": self.app,
            "services": self.services,
        }

    @property
    def command_groups(self) -> list[craft_cli.CommandGroup]:
        """Return command groups.

        Merges command groups provided by the application with craft-application's
        default commands.

        If the application and craft-application provide a command with the same name
        in the same group, the application's command is used.

        Note that a command with the same name cannot exist in multiple groups.
        """
        lifeycle_default_commands = commands.get_lifecycle_command_group()
        other_default_commands = commands.get_other_command_group()

        merged = {group.name: group for group in self._command_groups}

        merged[lifeycle_default_commands.name] = self._merge_defaults(
            app_commands=merged.get(lifeycle_default_commands.name),
            default_commands=lifeycle_default_commands,
        )
        merged[other_default_commands.name] = self._merge_defaults(
            app_commands=merged.get(other_default_commands.name),
            default_commands=other_default_commands,
        )

        return list(merged.values())

    def _merge_defaults(
        self,
        *,
        app_commands: craft_cli.CommandGroup | None,
        default_commands: craft_cli.CommandGroup,
    ) -> craft_cli.CommandGroup:
        """Merge default commands with application commands for a particular group.

        Default commands are only used if the application does not have a command
        with the same name.

        The order of the merged commands follow the order of the default commands.
        Extra application commands are appended to the end of the command list.

        :param app_commands: The application's commands.
        :param default_commands: Craft Application's default commands.

        :returns: A list of app commands and default commands.
        """
        if not app_commands:
            return default_commands

        craft_cli.emit.debug(f"Merging commands for group {default_commands.name!r}:")

        # for lookup of commands by name
        app_commands_dict = {command.name: command for command in app_commands.commands}

        merged_commands: list[type[craft_cli.BaseCommand]] = []
        processed_command_names: set[str] = set()

        for default_command in default_commands.commands:
            # prefer the application command if it exists
            command_name = default_command.name
            if command_name in app_commands_dict:
                craft_cli.emit.debug(
                    f"  - using application command for {command_name!r}."
                )
                merged_commands.append(app_commands_dict[command_name])
                processed_command_names.add(command_name)
            # otherwise use the default
            else:
                merged_commands.append(default_command)

        # append remaining commands from the application
        merged_commands.extend(
            app_command
            for app_command in app_commands.commands
            if app_command.name not in processed_command_names
        )

        return craft_cli.CommandGroup(
            name=default_commands.name,
            commands=merged_commands,
            ordered=default_commands.ordered,
        )

    @property
    def log_path(self) -> pathlib.Path | None:
        """Get the path to this process's log file, if any."""
        if self.is_managed():
            return util.get_managed_logpath(self.app)
        return None



    def add_global_argument(self, argument: craft_cli.GlobalArgument) -> None:
        """Add a global argument to the Application."""
        self._global_arguments.append(argument)




    def add_command_group(
        self,
        name: str,
        commands: Sequence[type[craft_cli.BaseCommand]],
        *,
        ordered: bool = False,
    ) -> None:
        """Add a CommandGroup to the Application."""
        self._command_groups.append(craft_cli.CommandGroup(name, commands, ordered))


    @cached_property
    def cache_dir(self) -> pathlib.Path:
        """Get the directory for caching any data."""
        try:
            return user_cache_path(self.app.name, ensure_exists=True)
        except FileExistsError as err:
            raise PathInvalidError(
                f"The cache path is not a directory: {err.strerror}"
            ) from err
        except OSError as err:
            raise PathInvalidError(
                f"Unable to create/access cache directory: {err.strerror}"
            ) from err

    def _configure_early_services(self) -> None:
        """Configure early-starting services.

        This should only contain configuration for services that are needed during
        application startup. All other configuration belongs in ``_configure_services``
        """
        self.services.update_kwargs(
            "project",
            project_dir=self.project_dir,
        )

    def _configure_services(self, provider_name: str | None) -> None:
        """Configure additional keyword arguments for any service classes.

        Any child classes that override this must either call this directly or must
        provide a valid ``project`` to ``self.services``.
        """
        self.services.update_kwargs(
            "lifecycle",
            cache_dir=self.cache_dir,
            work_dir=self._work_dir,
            use_host_sources=bool(self._pro_services),
        )
        self.services.update_kwargs(
            "provider",
            work_dir=self._work_dir,
            provider_name=provider_name,
            pro_services=self._pro_services,
        )



    def is_managed(self) -> bool:
        """Shortcut to tell whether we're running in managed mode."""
        return self.services.get_class("provider").is_managed()




    def configure(self, global_args: dict[str, Any]) -> None:
        """Configure the application using any global arguments."""


    def _get_dispatcher(self) -> craft_cli.Dispatcher:
        """Configure this application.

        Should be called by the _run_inner method.
        Side-effect: This method may exit the process.

        :returns: A ready-to-run Dispatcher object
        """
        dispatcher = self._create_dispatcher()

        try:
            craft_cli.emit.trace("pre-parsing arguments...")
            app_config = self.app_config
            # Workaround for the fact that craft_cli requires a command.
            # https://github.com/canonical/craft-cli/issues/141
            if "--version" in sys.argv or "-V" in sys.argv:
                try:
                    global_args = dispatcher.pre_parse_args(
                        ["pull", *sys.argv[1:]], app_config
                    )
                except craft_cli.ArgumentParsingError:
                    global_args = dispatcher.pre_parse_args(sys.argv[1:], app_config)
            else:
                global_args = dispatcher.pre_parse_args(sys.argv[1:], app_config)

            if global_args.get("version"):
                craft_cli.emit.message(f"{self.app.name} {self.app.version}")
                craft_cli.emit.ended_ok()
                sys.exit(0)
        except craft_cli.ProvideHelpException as err:
            print(err, file=sys.stderr)  # to stderr, as argparse normally does
            craft_cli.emit.ended_ok()
            sys.exit(0)
        except craft_cli.ArgumentParsingError as err:
            print(err, file=sys.stderr)  # to stderr, as argparse normally does
            craft_cli.emit.ended_ok()
            sys.exit(os.EX_USAGE)
        except KeyboardInterrupt as err:
            self._emit_error(craft_cli.CraftError("Interrupted."), cause=err)
            sys.exit(128 + signal.SIGINT)
        except Exception as err:
            self._emit_error(
                craft_cli.CraftError(
                    f"Internal error while loading {self.app.name}: {err!r}"
                )
            )
            if self.services.config.get("debug"):
                raise
            sys.exit(os.EX_SOFTWARE)

        craft_cli.emit.debug("Configuring application...")
        self.configure(global_args)

        return dispatcher

    def _create_dispatcher(self) -> craft_cli.Dispatcher:
        """Create the Dispatcher that will run the application's command.

        Subclasses can override this if they need to create a Dispatcher with
        different parameters.
        """
        return craft_cli.Dispatcher(
            self.app.name,
            self.command_groups,
            summary=str(self.app.summary),
            extra_global_args=self._global_arguments,
            docs_base_url=self.app.versioned_docs_url,
        )

    def _get_app_plugins(self) -> dict[str, PluginType]:
        """Get the plugins for this application.

        Should be overridden by applications that need to register plugins at startup.
        """
        return {}

    def _register_default_plugins(self) -> None:
        """Register per application plugins when initializing."""
        if plugins := self._get_app_plugins():
            from craft_parts.plugins import register  # noqa: PLC0415

            craft_cli.emit.trace("Registering plugins...")
            craft_cli.emit.trace(f"Plugins: {', '.join(plugins.keys())}")
            register(plugins)

    def _pre_run(self, dispatcher: craft_cli.Dispatcher) -> None:
        """Do any final setup before running the command.

        At the time this is run, the command is loaded in the dispatcher, but
        the project has not yet been loaded.
        """
        args = dispatcher.parsed_args()

        # Some commands might have a project_dir parameter. Those commands and
        # only those commands should get a project directory, but only when
        # not managed.
        if self.is_managed():
            self.project_dir = pathlib.Path("/root/project")
        elif project_dir := getattr(args, "project_dir", None):
            self.project_dir = pathlib.Path(project_dir).expanduser().resolve()
            if self.project_dir.exists() and not self.project_dir.is_dir():
                raise errors.ProjectFileMissingError(
                    "Provided project directory is not a directory.",
                    details=f"Not a directory: {project_dir}",
                    resolution="Ensure the path entered is correct.",
                )



    def get_arg_or_config(self, parsed_args: argparse.Namespace, item: str) -> Any:  # noqa: ANN401
        """Get a configuration option that could be overridden by a command argument.

        :param parsed_args: The argparse Namespace to check.
        :param item: the name of the namespace or config item.
        :returns: the requested value.
        """
        arg_value = getattr(parsed_args, item, None)
        if arg_value is not None:
            return arg_value
        return self.services.get("config").get(item)


    def _run_inner(self) -> int:
        """Actual run implementation."""
        dispatcher = self._get_dispatcher()
        command = cast(
            commands.AppCommand,
            dispatcher.load_command(self.app_config),
        )
        parsed_args = dispatcher.parsed_args()
        platform = self.get_arg_or_config(parsed_args, "platform")
        build_for = self.get_arg_or_config(parsed_args, "build_for")

        # Some commands (e.g. remote build) can allow multiple platforms
        # or build-fors, comma-separated. In these cases, we create the
        # project using the first defined platform.
        if platform and "," in platform:
            platform = platform.split(",", maxsplit=1)[0]
        if build_for and "," in build_for:
            build_for = build_for.split(",", maxsplit=1)[0]

        if self.app.enable_pro_support:
            self._pro_services = getattr(parsed_args, "pro", None)

        craft_cli.emit.debug(f"Build plan: platform={platform}, build_for={build_for}")
        self._pre_run(dispatcher)

        if command.needs_project(parsed_args):
            self.services.update_kwargs("project", pro_services=self._pro_services)
            project_service = self.services.get("project")
            # This branch always runs, except during testing.
            if not project_service.is_configured:
                project_service.configure(platform=platform, build_for=build_for)

        provider_name = command.provider_name(parsed_args)
        self._configure_services(provider_name)

        craft_cli.emit.debug(f"Running '{self.app.name} {command.name}'.")
        dispatcher_return = dispatcher.run()
        return dispatcher_return if dispatcher_return is not None else os.EX_OK



    def run(self) -> int:
        """Bootstrap and run the application."""
        self._setup_logging()
        self._configure_early_services()
        self._initialize_craft_parts()
        self._load_plugins()

        craft_cli.emit.debug("Preparing application...")

        debug_mode = self.services.get("config").get("debug")

        try:
            return_code = self._run_inner()
        # Other BaseException classes should be passed through, not caught.
        except (Exception, KeyboardInterrupt) as error:  # noqa: BLE001, this is not blind due to the handler code
            return_code = handle_runtime_error(
                self.app, error, print_error=self._emit_error, debug_mode=debug_mode
            )
        else:
            craft_cli.emit.ended_ok()

        return return_code


    def _emit_error(
        self, error: craft_cli.CraftError, *, cause: BaseException | None = None
    ) -> None:
        """Emit the error in a centralized way so we can alter it consistently."""
        # set the cause, if any
        if cause is not None:
            error.__cause__ = cause

        # Do not report the internal logpath if running inside an instance
        if self.is_managed():
            error.logpath_report = False

        craft_cli.emit.error(error)

    def _setup_logging(self) -> None:
        """Initialize the logging system."""
        # Set the logging level to DEBUG for all craft-libraries. This is OK even if
        # the specific application doesn't use a specific library, the call does not
        # import the package.
        emitter_mode: craft_cli.EmitterMode = craft_cli.EmitterMode.BRIEF
        invalid_emitter_level = False
        util.setup_loggers(*self._cli_loggers)

        # environment variable takes precedence over the default
        emitter_verbosity_level_env = os.environ.get("CRAFT_VERBOSITY_LEVEL", None)

        if emitter_verbosity_level_env:
            try:
                emitter_mode = craft_cli.EmitterMode[
                    emitter_verbosity_level_env.strip().upper()
                ]
            except KeyError:
                invalid_emitter_level = True

        craft_cli.emit.init(
            mode=emitter_mode,
            appname=self.app.name,
            greeting=f"Starting {self.app.name}, version {self.app.version}",
            log_filepath=self.log_path,
            streaming_brief=True,
            docs_base_url=self.app.versioned_docs_url,
        )

        craft_cli.emit.debug(f"Log verbosity level set to {emitter_mode.name}")

        if invalid_emitter_level:
            craft_cli.emit.progress(
                f"Invalid verbosity level '{emitter_verbosity_level_env}', using default 'BRIEF'.\n"
                f"Valid levels are: {', '.join(emitter.name for emitter in craft_cli.EmitterMode)}",
                permanent=True,
            )

    def _enable_craft_parts_features(self) -> None:
        """Enable any specific craft-parts Feature that the application will need."""

    @final
    def _set_plugin_group(self) -> None:
        """Set the plugin group from the lifecycle service.

        If no plugin group is provided or an error occurs while determining
        the build info, no plugin group is registered.
        """
        try:
            build_plan = self.services.get("build_plan").plan()
        except (craft_cli.CraftError, craft_platforms.CraftPlatformsError):
            # We can do this here because when we start the lifecycle
            # we actually exit the app if there's an error.
            craft_cli.emit.debug("No plugin group registered due to error.")
            return
        group = self.services.get_class("lifecycle").get_plugin_group(build_plan[0])

        # We don't need to import this unless we have a group to set.
        from craft_parts.plugins import set_plugin_group  # noqa: PLC0415

        if group:
            set_plugin_group(group)

    def _initialize_craft_parts(self) -> None:
        """Perform craft-parts-specific initialization, like features and plugins."""
        self._enable_craft_parts_features()
        self._register_default_plugins()
        self._set_plugin_group()