imbajin commented on code in PR #350:
URL: https://github.com/apache/hugegraph-ai/pull/350#discussion_r3329366584
##########
hugegraph-llm/src/hugegraph_llm/config/models/base_config.py:
##########
@@ -16,130 +16,342 @@
# under the License.
+import collections.abc
import os
+import threading
+import time
+from typing import ClassVar, Optional
-from dotenv import dotenv_values, set_key
-from pydantic_settings import BaseSettings
+from dotenv import dotenv_values
+from omegaconf import DictConfig, OmegaConf
+from pydantic import BaseModel, ConfigDict, TypeAdapter
from hugegraph_llm.utils.log import log
dir_name = os.path.dirname
-env_path = os.path.join(os.getcwd(), ".env") # Load .env from the current
working directory
+YAML_PATH = os.path.join(os.getcwd(), "config.yaml")
+ENV_PATH = os.path.join(os.getcwd(), ".env")
-class BaseConfig(BaseSettings):
- class Config:
- env_file = env_path
- case_sensitive = False
- extra = "ignore" # ignore extra fields to avoid ValidationError
- env_ignore_empty = True
+def _flat_to_nested(flat_dict: dict, mapping: dict) -> dict:
+ """Convert flat field names to nested dict using dot-notation mapping.
- def generate_env(self):
- if os.path.exists(env_path):
+ Mapping: {"flat_name": "nested.path.key", ...}
+ Fields not in the mapping are kept at the top level.
+ """
+ if not mapping:
+ return flat_dict
+ result: dict = {}
+ for field_name, value in flat_dict.items():
+ if field_name in mapping:
+ path = mapping[field_name]
+ parts = path.split(".")
+ d = result
+ for part in parts[:-1]:
+ if part not in d:
+ d[part] = {}
+ d = d[part]
+ d[parts[-1]] = value
+ else:
+ result[field_name] = value
+ return result
+
+
+def _nested_to_flat(nested_dict: dict, mapping: dict) -> dict:
+ """Convert nested dict from YAML to flat field names using dot-notation
mapping.
+
+ Reverse of _flat_to_nested. Walks the nested dict, matching dot-joined
+ paths against the mapping keys.
+ """
+ if not mapping or not nested_dict:
+ return nested_dict
+ reverse_map = {v: k for k, v in mapping.items()}
+ result = {}
+
+ def _walk(prefix: str, d: dict) -> None:
+ for key, value in d.items():
+ full_key = f"{prefix}.{key}" if prefix else key
+ if full_key in reverse_map:
+ result[reverse_map[full_key]] = value
+ elif isinstance(value, collections.abc.Mapping):
+ _walk(full_key, value)
+ else:
+ result[full_key.replace(".", "_")] = value
+
+ _walk("", nested_dict)
+ return result
+
+
+class ConfigManager:
+ """Singleton manager for OmegaConf-based YAML configuration.
+
+ Lifecycle:
+ 1. __init__: load config.yaml or migrate from .env, start file watcher
+ 2. Config classes read via get_section_with_env_override()
+ 3. Config classes write via update_section() + save()
+ 4. Background watcher polls for external changes → reload()
+ """
+
+ _instance: ClassVar[Optional["ConfigManager"]] = None
+
+ def __new__(cls, sections=None):
+ if cls._instance is None:
+ cls._instance = super().__new__(cls)
+ cls._instance._initialized = False
+ return cls._instance
+
+ def __init__(self, sections=None):
+ if self._initialized:
+ return
+ self._initialized = True
+ self._yaml_path = YAML_PATH
+ self._env_path = ENV_PATH
+ self._sections: dict = sections or {}
+ self._cfg: DictConfig = OmegaConf.create({})
+ self._reload_lock = threading.Lock()
+ self._watching = False
+ self._watcher_thread: Optional[threading.Thread] = None
+ self._last_mtime: float = 0.0
+ self._reload_targets: list = [] # (section_name, config_object) tuples
+
+ # Load .env into os.environ for backward compatibility and priority
override
+ if os.path.exists(self._env_path):
+ for k, v in dotenv_values(self._env_path).items():
+ os.environ[k] = v
Review Comment:
‼️ **Preserve real environment-variable precedence**
This loop copies every key from `.env` into `os.environ` before YAML/env
override resolution. That reverses the intended deployment precedence when the
process already has real environment variables: I reproduced this with `.env`
containing `OPENAI_API_KEY=from_dotenv` while launching with
`OPENAI_API_KEY=from_real_env`, and `llm_settings.openai_chat_api_key` became
`from_dotenv`. Existing pydantic settings gave the real environment precedence
over the dotenv file, so a stale local `.env` can override Docker/Kubernetes
secrets after this migration. Please only fill missing keys from `.env` (for
example `os.environ.setdefault(...)`, skipping empty values) and add a
regression test for env > .env/config.yaml precedence.
```suggestion
for k, v in dotenv_values(self._env_path).items():
if v:
os.environ.setdefault(k, v)
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
