From fd6126fce7484c72d5e2ce7d674d74917e5c1980 Mon Sep 17 00:00:00 2001 From: worldmozara Date: Mon, 6 Jul 2026 21:55:17 +0800 Subject: [PATCH] core/config: add configuration center --- src/plyngent/config/__init__.py | 45 ++++++++++ src/plyngent/config/models.py | 47 +++++++++++ src/plyngent/config/path.py | 19 +++++ src/plyngent/config/store.py | 140 ++++++++++++++++++++++++++++++++ 4 files changed, 251 insertions(+) create mode 100644 src/plyngent/config/__init__.py create mode 100644 src/plyngent/config/models.py create mode 100644 src/plyngent/config/path.py create mode 100644 src/plyngent/config/store.py diff --git a/src/plyngent/config/__init__.py b/src/plyngent/config/__init__.py new file mode 100644 index 0000000..9b5a1d9 --- /dev/null +++ b/src/plyngent/config/__init__.py @@ -0,0 +1,45 @@ +from __future__ import annotations + +from typing import TYPE_CHECKING + +import tomlkit +from tomlkit.exceptions import TOMLKitError + +from .models import AnthropicProvider as AnthropicProvider +from .models import DeepseekProvider as DeepseekProvider +from .models import ModelConfig as ModelConfig +from .models import OpenAICompatibleProvider as OpenAICompatibleProvider +from .models import OpenAIProvider as OpenAIProvider +from .models import ProviderConfig as ProviderConfig +from .path import get_default_path as _get_default_path +from .store import ConfigFormatError as ConfigFormatError +from .store import ConfigStore as ConfigStore + +if TYPE_CHECKING: + from pathlib import Path + +default_config_source: Path = _get_default_path() + + +def load(path: Path | None = None) -> ConfigStore: + """Load configuration from a TOML file. + + Args: + path: Path to the config file. If ``None``, uses :data:`default_config_source`. + + Returns: + A :class:`ConfigStore` with parsed providers. + + Raises: + ConfigFormatError: If the file contains invalid TOML. + """ + if path is None: + path = default_config_source + try: + with path.open() as f: + doc = tomlkit.parse(f.read()) + except FileNotFoundError: + doc = tomlkit.document() + except TOMLKitError as exc: + raise ConfigFormatError(str(exc)) from exc + return ConfigStore(path=path, document=doc) diff --git a/src/plyngent/config/models.py b/src/plyngent/config/models.py new file mode 100644 index 0000000..fa501d6 --- /dev/null +++ b/src/plyngent/config/models.py @@ -0,0 +1,47 @@ +from msgspec import Struct, field + + +class ModelConfig(Struct, omit_defaults=True): + """Capability flags for a model within a provider.""" + + text: bool = True + image_in: bool = False + image_out: bool = False + audio_in: bool = False + audio_out: bool = False + video_in: bool = False + video_out: bool = False + cost_factor: float = 1.0 + + +class ProviderConfig(Struct, tag_field="preset", omit_defaults=True): + """Base config for all LLM providers. + + The ``preset`` tag field acts as the tagged-union discriminator + and is auto-generated by msgspec — do not re-declare it in subclasses. + """ + + access_key_or_token: str + url: str = "" + models: dict[str, ModelConfig] = field(default_factory=dict) + + +class OpenAIProvider(ProviderConfig, tag="openai"): + """OpenAI API provider.""" + + +class OpenAICompatibleProvider(ProviderConfig, tag="openai-compatible"): + """Generic OpenAI-compatible provider (e.g. vLLM, LiteLLM).""" + + +class AnthropicProvider(ProviderConfig, tag="anthropic"): + """Anthropic API provider.""" + + +class DeepseekProvider(ProviderConfig, tag="deepseek"): + """Deepseek API provider with optional extras (e.g. convention).""" + + extras: dict[str, str] = field(default_factory=dict) + + +type Provider = OpenAIProvider | OpenAICompatibleProvider | AnthropicProvider | DeepseekProvider diff --git a/src/plyngent/config/path.py b/src/plyngent/config/path.py new file mode 100644 index 0000000..b1003f9 --- /dev/null +++ b/src/plyngent/config/path.py @@ -0,0 +1,19 @@ +from __future__ import annotations + +from typing import TYPE_CHECKING + +from platformdirs import user_config_path + +if TYPE_CHECKING: + from pathlib import Path + +_CONFIG_FILE_NAME: str = "plyngent.toml" + + +def get_default_path() -> Path: + """Return the default config file path using the platform-standard config directory. + + On Linux, this is ``~/.config/plyngent/plyngent.toml``. + ``ensure_exists=True`` creates the directory if it does not exist. + """ + return user_config_path("plyngent", ensure_exists=True) / _CONFIG_FILE_NAME diff --git a/src/plyngent/config/store.py b/src/plyngent/config/store.py new file mode 100644 index 0000000..1245b77 --- /dev/null +++ b/src/plyngent/config/store.py @@ -0,0 +1,140 @@ +from __future__ import annotations + +from types import MappingProxyType +from typing import TYPE_CHECKING, cast + +import msgspec +import tomlkit + +from .models import Provider + +if TYPE_CHECKING: + from collections.abc import Mapping + from pathlib import Path + + +class ConfigFormatError(ValueError): + """Raised when the config file contains invalid TOML.""" + + +def _parse_providers( + document: tomlkit.TOMLDocument, +) -> tuple[dict[str, Provider], dict[str, object]]: + """Parse provider entries from the document. + + Returns: + (providers, bad_providers) — valid and invalid entries respectively. + """ + providers: dict[str, Provider] = {} + bad_providers: dict[str, object] = {} + + raw: dict[str, object] = document.unwrap() + providers_raw: object = raw.get("providers", {}) + if not isinstance(providers_raw, dict): + return providers, bad_providers + + for name, raw_entry in cast("dict[str, object]", providers_raw).items(): + if not isinstance(raw_entry, dict): + bad_providers[name] = raw_entry + continue + + # Try tagged-union dispatch via msgspec + try: + provider: Provider = msgspec.convert(raw_entry, Provider) + except msgspec.ValidationError: + bad_providers[name] = raw_entry + continue + + # msgspec silently ignores unknown fields — reject them manually. + # The tag_field ("preset") is not a struct field, so we add it back. + cls = type(provider) + known = set(cls.__struct_fields__) + known.add("preset") # tag_field, excluded from __struct_fields__ + if set(cast("dict[str, object]", raw_entry).keys()) - known: + bad_providers[name] = raw_entry + continue + + providers[name] = provider + + return providers, bad_providers + + +class ConfigStore: + """Stateful wrapper around a TOML config file.""" + + _path: Path + _document: tomlkit.TOMLDocument + _providers: dict[str, Provider] + _bad_providers: dict[str, object] + + def __init__(self, path: Path, document: tomlkit.TOMLDocument) -> None: + self._path = path + self._document = document + self._providers, self._bad_providers = _parse_providers(document) + + # -- providers (read/write) -- + + @property + def providers(self) -> MappingProxyType[str, Provider]: + """Read-only mapping view of recognised providers. + + Supports ``|`` for augmented-assignment merge. + """ + return MappingProxyType(self._providers) + + @providers.setter + def providers(self, value: Mapping[str, Provider]) -> None: # pyright: ignore[reportPropertyTypeMismatch] + """Replace all providers.""" + self._providers = dict(value) + self._bad_providers = {} + + # -- bad_providers (read-only) -- + + @property + def bad_providers(self) -> MappingProxyType[str, object]: + """Read-only view of unrecognised / malformed provider entries.""" + return MappingProxyType(self._bad_providers) + + # -- persistence -- + + def write(self) -> None: + """Serialize current providers to the TOML file.""" + self._sync_to_document() + self._path.parent.mkdir(parents=True, exist_ok=True) + with self._path.open("w") as f: + tomlkit.dump(self._document, f) + + def reload(self) -> None: + """Re-read the TOML file and re-parse providers.""" + with self._path.open() as f: + self._document = tomlkit.parse(f.read()) + self._providers, self._bad_providers = _parse_providers(self._document) + + # -- internal -- + + def _sync_to_document(self) -> None: + """Incrementally sync ``self._providers`` into the ``[providers]`` section. + + Only adds, updates, or removes individual provider entries — the rest of + the document (comments, formatting, other top-level sections) is untouched. + """ + if "providers" not in self._document: + self._document["providers"] = tomlkit.table() + section = self._document["providers"] + + # Remove deleted providers + for name in list(section.keys()): # type: ignore[union-attr] + if name not in self._providers: + del section[name] # type: ignore[union-attr] + + # Add / update providers + for name, provider in self._providers.items(): + raw: dict[str, object] = msgspec.to_builtins(provider) + if name in section: + entry = section[name] # type: ignore[union-attr] + # Clear and rebuild the existing entry to match current state + entry.clear() # type: ignore[union-attr] + for k, v in raw.items(): + entry[k] = v # type: ignore[union-attr] + else: + section[name] = raw # type: ignore[union-attr]