111 lines
3.0 KiB
Python
111 lines
3.0 KiB
Python
# SPDX-License-Identifier: Apache-2.0
|
|
from __future__ import annotations
|
|
|
|
from copy import deepcopy
|
|
from typing import Any
|
|
from collections.abc import Mapping
|
|
|
|
import yaml
|
|
|
|
from fastvideo.api.errors import ConfigValidationError
|
|
|
|
|
|
def parse_cli_overrides(overrides: list[str]) -> dict[str, Any]:
|
|
"""Parse ``--dotted.key value`` style overrides into a flat mapping."""
|
|
parsed: dict[str, Any] = {}
|
|
index = 0
|
|
while index < len(overrides):
|
|
token = overrides[index]
|
|
if not token.startswith("--"):
|
|
raise ValueError(f"Expected --dotted.key, got {token!r}")
|
|
|
|
key = token[2:]
|
|
if not key:
|
|
raise ValueError("Override key cannot be empty")
|
|
|
|
if "=" in key:
|
|
key, raw_value = key.split("=", 1)
|
|
else:
|
|
index += 1
|
|
if index >= len(overrides):
|
|
raise ValueError(f"Missing value for override {token!r}")
|
|
raw_value = overrides[index]
|
|
|
|
parsed[_normalize_override_key(key)] = _cast_override_value(raw_value)
|
|
index += 1
|
|
|
|
return parsed
|
|
|
|
|
|
def apply_overrides(config: Mapping[str, Any], overrides: Mapping[str, Any]) -> dict[str, Any]:
|
|
"""Return a copy of ``config`` with dotted-key overrides applied."""
|
|
merged = deepcopy(dict(config))
|
|
for dotted_key, value in overrides.items():
|
|
_apply_single_override(merged, dotted_key, value)
|
|
return merged
|
|
|
|
|
|
def normalize_overrides(overrides: list[str] | Mapping[str, Any] | None, ) -> dict[str, Any] | None:
|
|
"""Normalize a CLI list or mapping of overrides into a flat dict."""
|
|
if not overrides:
|
|
return None
|
|
if isinstance(overrides, list):
|
|
return parse_cli_overrides(overrides)
|
|
return dict(overrides)
|
|
|
|
|
|
def _apply_single_override(config: dict[str, Any], dotted_key: str, value: Any) -> None:
|
|
parts = dotted_key.split(".")
|
|
if not all(parts):
|
|
raise ValueError(f"Invalid override path {dotted_key!r}")
|
|
|
|
cursor = config
|
|
for depth, part in enumerate(parts[:-1]):
|
|
existing = cursor.get(part)
|
|
if existing is None:
|
|
existing = {}
|
|
cursor[part] = existing
|
|
elif not isinstance(existing, dict):
|
|
raise ConfigValidationError(
|
|
".".join(parts[:depth + 1]),
|
|
"cannot apply nested override through a non-mapping value",
|
|
)
|
|
cursor = existing
|
|
|
|
cursor[parts[-1]] = value
|
|
|
|
|
|
def _cast_override_value(raw: str) -> Any:
|
|
lowered = raw.lower()
|
|
if lowered == "true":
|
|
return True
|
|
if lowered == "false":
|
|
return False
|
|
if lowered in {"none", "null"}:
|
|
return None
|
|
|
|
try:
|
|
return int(raw)
|
|
except ValueError:
|
|
pass
|
|
|
|
try:
|
|
return float(raw)
|
|
except ValueError:
|
|
pass
|
|
|
|
if raw.startswith("[") or raw.startswith("{"):
|
|
try:
|
|
return yaml.safe_load(raw)
|
|
except yaml.YAMLError:
|
|
pass
|
|
|
|
return raw
|
|
|
|
|
|
def _normalize_override_key(key: str) -> str:
|
|
return key.replace("-", "_")
|
|
|
|
|
|
__all__ = ["apply_overrides", "normalize_overrides", "parse_cli_overrides"]
|