# This file is part of craft-application.
#
# Copyright 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/>.
"""Configuration service."""
from __future__ import annotations
import abc
import contextlib
import enum
import os
from typing import TYPE_CHECKING, Any, TypeVar, cast, final
import pydantic
import pydantic_core
import snaphelpers
from craft_cli import emit
from typing_extensions import override
from craft_application import _config, application, util
from craft_application.services import base
if TYPE_CHECKING:
from collections.abc import Iterable
from craft_application.services.service_factory import ServiceFactory
T = TypeVar("T")
[docs]
class ConfigHandler(abc.ABC):
"""The abstract class that is the parent of all configuration handlers."""
[docs]
def __init__(self, app: application.AppMetadata) -> None:
"""Create the configuration handler with the relevant application metadata."""
self._app = app
[docs]
@abc.abstractmethod
def get_raw(self, item: str) -> Any: # noqa: ANN401
"""Get the string value for a configuration item.
:param item: the name of the configuration item.
:returns: The raw value of the item.
:raises: KeyError if the item cannot be found.
"""
[docs]
@final
class AppEnvironmentHandler(ConfigHandler):
"""Configuration handler to get values from app-specific environment variables.
Environment variables used for this are prefixed with a fully upper-case form
of the application name. For example, ``TESTCRAFT_DEBUG``.
"""
[docs]
def __init__(self, app: application.AppMetadata) -> None:
super().__init__(app)
self._environ_prefix = f"{app.name.upper()}"
[docs]
@override
def get_raw(self, item: str) -> str:
return os.environ[f"{self._environ_prefix}_{item.upper()}"]
[docs]
@final
class CraftEnvironmentHandler(ConfigHandler):
"""Configuration handler to get values from CRAFT environment variables."""
[docs]
def __init__(self, app: application.AppMetadata) -> None:
super().__init__(app)
self._fields = _config.ConfigModel.model_fields
[docs]
@override
def get_raw(self, item: str) -> str:
# Ensure that CRAFT_* env vars can only be used for configuration items
# known to craft-application.
if item not in self._fields:
raise KeyError(f"{item!r} not a general craft-application config item.")
return os.environ[f"CRAFT_{item.upper()}"]
[docs]
class SnapConfigHandler(ConfigHandler):
"""Configuration handler that gets values from snapd.
Snap configuration values are set with kebab case, so the ``verbosity_level``
configuration value can be set to ``verbose`` using the command
``snap set <snap-name> verbosity-level=verbose``
"""
[docs]
def __init__(self, app: application.AppMetadata) -> None:
super().__init__(app)
if not snaphelpers.is_snap():
raise OSError("Not running as a snap.")
try:
self._snap = snaphelpers.SnapConfig()
except (KeyError, AttributeError):
raise OSError("Not running as a snap.")
except snaphelpers.SnapCtlError:
# Most likely to happen in a container that has the snap environment set.
# See: https://github.com/canonical/snapcraft/issues/5079
emit.progress(
"Snap environment is set, but cannot connect to snapd. "
"Snap configuration is unavailable.",
permanent=True,
)
raise OSError("Not running as a snap or with snapd disabled.")
[docs]
@override
def get_raw(self, item: str) -> Any:
snap_item = item.replace("_", "-")
try:
return self._snap.get(snap_item)
except snaphelpers.UnknownConfigKey as exc:
raise KeyError(f"unknown snap config item: {item!r}") from exc
[docs]
@final
class DefaultConfigHandler(ConfigHandler):
"""Configuration handler for getting default values."""
[docs]
def __init__(self, app: application.AppMetadata) -> None:
super().__init__(app)
self._config_model = app.ConfigModel
self._cache: dict[str, str] = {}
[docs]
@override
def get_raw(self, item: str) -> Any:
if item in self._cache:
return self._cache[item]
field = self._config_model.model_fields[item]
if field.default is not pydantic_core.PydanticUndefined:
self._cache[item] = field.default
return field.default
if field.default_factory is not None:
# Remove the type ignore after pydantic/pydantic#10945 is fixed
default = field.default_factory() # type: ignore[call-arg]
self._cache[item] = default
return default
raise KeyError(f"config item {item!r} has no default value.")
[docs]
class ConfigService(base.AppService):
"""Application-wide configuration access."""
_handlers: list[ConfigHandler]
[docs]
def __init__(
self,
app: application.AppMetadata,
services: ServiceFactory,
*,
extra_handlers: Iterable[type[ConfigHandler]] = (),
) -> None:
super().__init__(app, services)
self._extra_handlers = extra_handlers
self._default_handler = DefaultConfigHandler(self._app)
[docs]
@override
def setup(self) -> None:
super().setup()
self._handlers = [
AppEnvironmentHandler(self._app),
CraftEnvironmentHandler(self._app),
*(handler(self._app) for handler in self._extra_handlers),
]
try:
snap_handler = SnapConfigHandler(self._app)
except OSError:
emit.debug(
"App is not running as a snap - snap config handler not created."
)
else:
self._handlers.append(snap_handler)
[docs]
def get(self, item: str) -> Any: # noqa: ANN401
"""Get the given configuration item."""
if item not in self._app.ConfigModel.model_fields:
raise KeyError(f"unknown config item: {item!r}")
field_info = self._app.ConfigModel.model_fields[item]
for handler in self._handlers:
try:
value = handler.get_raw(item)
except KeyError: # noqa: PERF203
continue
else:
break
else:
return self._default_handler.get_raw(item)
return self._convert_type(value, field_info.annotation) # type: ignore[arg-type,return-value]
[docs]
def _convert_type(self, value: str, field_type: type[T]) -> T:
"""Convert the value to the appropriate type."""
if isinstance(field_type, type): # pyright: ignore[reportUnnecessaryIsInstance]
if issubclass(field_type, str):
return field_type(value)
if issubclass(field_type, bool):
return cast(T, util.strtobool(value))
if issubclass(field_type, enum.Enum):
with contextlib.suppress(KeyError):
return field_type[value]
with contextlib.suppress(KeyError):
return field_type[value.upper()]
field_adapter = pydantic.TypeAdapter(field_type)
return field_adapter.validate_strings(value)
[docs]
def get_all(self) -> dict[str, Any]:
"""Get a dictionary of the complete configuration per the ConfigModel.
Configuration items that are unset but have no default value are not included
in the resulting mapping.
"""
config: dict[str, Any] = {}
for field in self._app.ConfigModel.model_fields:
try:
config_value = self.get(field)
except KeyError:
continue
with contextlib.suppress(AttributeError):
default_value = getattr(self._app.ConfigModel, field).default
if config_value == default_value:
continue
config[field] = config_value
return config
