"""ConfigService - Manages CLI configuration file storage.
This service manages the CLI configuration settings by persisting to a
local YAML file in the user's config directory.
"""
from __future__ import annotations
from pathlib import Path
from typing import Optional
import yaml
from platformdirs import user_config_dir
from pydantic import BaseModel, Field
from kelvin.sdk.exceptions import CLIError
[docs]
class ConfigServiceError(CLIError):
"""Base exception for ConfigService errors."""
exit_code: int = 78 # EX_CONFIG - configuration error
[docs]
class InvalidConfigKeyError(ConfigServiceError):
"""Raised when an invalid configuration key is provided."""
def __init__(self, key: str, valid_keys: list[str]) -> None:
self.key: str = key
self.valid_keys: list[str] = valid_keys
super().__init__(f"Invalid configuration key: '{key}'. Valid keys are: {', '.join(valid_keys)}")
[docs]
class InvalidConfigValueError(ConfigServiceError):
"""Raised when an invalid configuration value is provided."""
def __init__(self, key: str, value: str, expected_type: str) -> None:
self.key: str = key
self.value: str = value
self.expected_type: str = expected_type
super().__init__(f"Invalid value '{value}' for '{key}'. Expected {expected_type}.")
# ============ Configuration Model ============
[docs]
class CLIConfig(BaseModel):
"""CLI configuration settings.
This is the single source of truth for configuration defaults.
Field defaults here are used everywhere in the CLI.
"""
version_warning: bool = Field(
default=True,
description="Show version update warnings",
)
colored_logs: bool = Field(
default=True,
description="Enable colored log output",
)
verbose: int = Field(
default=0,
description="Verbosity level (0=normal, 1=info, 2=debug)",
)
json_output: bool = Field(
default=False,
description="Default to JSON output mode",
)
no_prompt: bool = Field(
default=False,
description="Skip all confirmation prompts",
)
analytics: bool = Field(
default=True,
description="Enable usage analytics tracking",
)
[docs]
def get_config_keys() -> dict[str, dict[str, object]]:
"""Get configuration keys with metadata from the Pydantic model.
Returns:
Dictionary of key names to metadata (type, description, default).
"""
result: dict[str, dict[str, object]] = {}
defaults = CLIConfig()
for field_name, field_info in CLIConfig.model_fields.items():
# Map Python types to simple type names
annotation = field_info.annotation
if annotation is bool:
type_name = "bool"
elif annotation is int:
type_name = "int"
elif annotation is str:
type_name = "str"
else:
type_name = "str"
result[field_name] = {
"type": type_name,
"description": field_info.description or "",
"default": getattr(defaults, field_name),
}
return result
# Backward compatibility alias
CONFIG_KEYS = get_config_keys()
# ============ Service ============
[docs]
class ConfigService:
"""Manages CLI configuration file storage.
This is a leaf service with no dependencies. It persists configuration
settings to a YAML file in the user's config directory.
"""
CONFIG_DIR: str = "kelvin"
CONFIG_FILE: str = "config.yaml"
[docs]
def __init__(self) -> None:
"""Initialize ConfigService."""
self._config_dir: Path = Path(user_config_dir(self.CONFIG_DIR))
self._cached_config: Optional[CLIConfig] = None
def _get_config_file_path(self) -> Path:
"""Get the path to the config file."""
return self._config_dir / self.CONFIG_FILE
def _ensure_config_dir(self) -> None:
"""Ensure the config directory exists."""
self._config_dir.mkdir(parents=True, exist_ok=True)
[docs]
def get_config(self) -> CLIConfig:
"""Get the current configuration.
Returns:
CLIConfig with current settings.
"""
if self._cached_config is not None:
return self._cached_config
config_file = self._get_config_file_path()
if not config_file.exists():
self._cached_config = CLIConfig()
return self._cached_config
try:
with open(config_file) as f:
data: dict[str, object] = yaml.safe_load(f) or {} # pyright: ignore[reportUnknownMemberType]
# Pydantic handles defaults for missing fields and type coercion
self._cached_config = CLIConfig.model_validate(data)
return self._cached_config
except (OSError, yaml.YAMLError, ValueError):
# OSError: file permission issues
# YAMLError: corrupt YAML syntax
# ValueError: Pydantic validation failure
self._cached_config = CLIConfig()
return self._cached_config
[docs]
def save_config(self, config: CLIConfig) -> None:
"""Save the configuration to file.
Args:
config: The configuration to save.
"""
self._ensure_config_dir()
config_file = self._get_config_file_path()
with open(config_file, "w") as f:
yaml.safe_dump(config.model_dump(), f, default_flow_style=False) # pyright: ignore[reportUnknownMemberType]
self._cached_config = config
[docs]
def set_value(self, key: str, value: str) -> CLIConfig:
"""Set a configuration value.
Args:
key: The configuration key.
value: The value to set (as string, will be converted).
Returns:
Updated CLIConfig.
Raises:
InvalidConfigKeyError: If key is not valid.
InvalidConfigValueError: If value cannot be parsed.
"""
config_keys = get_config_keys()
if key not in config_keys:
raise InvalidConfigKeyError(key, list(config_keys.keys()))
key_meta = config_keys[key]
key_type = key_meta["type"]
# Parse value based on type
parsed_value: object
if key_type == "bool":
if value.lower() in ("true", "1", "yes", "on"):
parsed_value = True
elif value.lower() in ("false", "0", "no", "off"):
parsed_value = False
else:
raise InvalidConfigValueError(key, value, "boolean (true/false)")
elif key_type == "int":
try:
parsed_value = int(value)
except ValueError:
raise InvalidConfigValueError(key, value, "integer") from None
else:
parsed_value = value
# Create new config with updated value
config = self.get_config()
new_config = config.model_copy(update={key: parsed_value})
self.save_config(new_config)
return new_config
[docs]
def unset_value(self, key: str) -> CLIConfig:
"""Reset a configuration value to its default.
Args:
key: The configuration key.
Returns:
Updated CLIConfig.
Raises:
InvalidConfigKeyError: If key is not valid.
"""
config_keys = get_config_keys()
if key not in config_keys:
raise InvalidConfigKeyError(key, list(config_keys.keys()))
# Get default from a fresh CLIConfig instance
default_config = CLIConfig()
default_value: object = getattr(default_config, key)
# Create new config with default value
config = self.get_config()
new_config = config.model_copy(update={key: default_value})
self.save_config(new_config)
return new_config
[docs]
def reset(self) -> None:
"""Reset all configuration to defaults and delete the config file."""
config_file = self._get_config_file_path()
if config_file.exists():
config_file.unlink()
self._cached_config = None
[docs]
def get_config_keys(self) -> dict[str, dict[str, object]]:
"""Get all configuration keys with their metadata.
Returns:
Dictionary of key names to metadata.
"""
return get_config_keys()
[docs]
def get_config_file_path(self) -> Path:
"""Get the path to the config file (for display purposes).
Returns:
Path to the config file.
"""
return self._get_config_file_path()