Source code for paper_firehose.commands.config_cmd

"""Config management CLI subcommands."""

from __future__ import annotations

import logging
from pathlib import Path
from typing import Any, Dict, Optional

import yaml

from ..core.config import ConfigManager, _KNOWN_MAIN_KEYS

logger = logging.getLogger(__name__)


def _coerce_value(raw: str) -> Any:
    """Auto-detect type from a CLI string value."""
    if raw.lower() in ("true", "yes"):
        return True
    if raw.lower() in ("false", "no"):
        return False
    try:
        return int(raw)
    except ValueError:
        pass
    try:
        return float(raw)
    except ValueError:
        pass
    return raw


def _traverse(data: dict, keys: list[str]) -> Any:
    """Traverse a nested dict by dot-notation key parts."""
    current = data
    for k in keys:
        if not isinstance(current, dict) or k not in current:
            raise KeyError(f"Key not found: {'.'.join(keys)}")
        current = current[k]
    return current


def _set_nested(data: dict, keys: list[str], value: Any) -> None:
    """Set a value in a nested dict by dot-notation key parts."""
    current = data
    for k in keys[:-1]:
        if k not in current or not isinstance(current[k], dict):
            current[k] = {}
        current = current[k]
    current[keys[-1]] = value


[docs] def show(config_path: str) -> str: """Return the main config as formatted YAML.""" cfg = ConfigManager(config_path) config = cfg.load_config() return yaml.safe_dump(config, default_flow_style=False, sort_keys=False)
[docs] def get_value(config_path: str, key: str) -> Any: """Get a value from the main config using dot-notation.""" cfg = ConfigManager(config_path) config = cfg.load_config() parts = key.split(".") return _traverse(config, parts)
[docs] def set_value(config_path: str, key: str, raw_value: str) -> None: """Set a value in the main config using dot-notation. Writes the updated config back to disk. """ cfg = ConfigManager(config_path) config = cfg.load_config() parts = key.split(".") value = _coerce_value(raw_value) # Warn on unknown top-level key if parts[0] not in _KNOWN_MAIN_KEYS: logger.warning("Unknown config key: %s", parts[0]) _set_nested(config, parts, value) config_file = Path(cfg.config_path) config_file.write_text( yaml.safe_dump(config, default_flow_style=False, sort_keys=False), encoding="utf-8", )
[docs] def validate(config_path: str) -> tuple[bool, list[str]]: """Run full config validation. Returns: ``(is_valid, unknown_keys)`` """ cfg = ConfigManager(config_path) valid = cfg.validate_config() unknown = cfg.check_unknown_keys() return valid, unknown