ef806f8fbb
Phase 1-3 of decoupling plan: - Shared DB, models, infrastructure, browser, config, utils - Event infrastructure (domain_events outbox, bus, processor) - Structured logging - Generic container concept (container_type/container_id) - Alembic migrations for all schema changes Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
85 lines
2.7 KiB
Python
85 lines
2.7 KiB
Python
# suma_browser/config.py
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import os
|
|
from types import MappingProxyType
|
|
from typing import Any, Optional
|
|
import copy
|
|
import yaml
|
|
|
|
# Default config path (override with APP_CONFIG_FILE)
|
|
_DEFAULT_CONFIG_PATH = os.environ.get(
|
|
"APP_CONFIG_FILE",
|
|
os.path.join(os.getcwd(), "config/app-config.yaml"),
|
|
)
|
|
|
|
# Module state
|
|
_init_lock = asyncio.Lock()
|
|
_data_frozen: Any = None # read-only view (mappingproxy / tuples / frozensets)
|
|
_data_plain: Any = None # plain builtins for pretty-print / logging
|
|
|
|
# ---------------- utils ----------------
|
|
def _freeze(obj: Any) -> Any:
|
|
"""Deep-freeze containers to read-only equivalents."""
|
|
if isinstance(obj, dict):
|
|
# freeze children first, then wrap dict in mappingproxy
|
|
return MappingProxyType({k: _freeze(v) for k, v in obj.items()})
|
|
if isinstance(obj, list):
|
|
return tuple(_freeze(v) for v in obj)
|
|
if isinstance(obj, set):
|
|
return frozenset(_freeze(v) for v in obj)
|
|
if isinstance(obj, tuple):
|
|
return tuple(_freeze(v) for v in obj)
|
|
return obj
|
|
|
|
# ---------------- API ----------------
|
|
async def init_config(path: Optional[str] = None, *, force: bool = False) -> None:
|
|
"""
|
|
Load YAML exactly as-is and cache both a frozen (read-only) and a plain copy.
|
|
Idempotent; pass force=True to reload.
|
|
"""
|
|
global _data_frozen, _data_plain
|
|
|
|
if _data_frozen is not None and not force:
|
|
return
|
|
|
|
async with _init_lock:
|
|
if _data_frozen is not None and not force:
|
|
return
|
|
|
|
cfg_path = path or _DEFAULT_CONFIG_PATH
|
|
if not os.path.exists(cfg_path):
|
|
raise FileNotFoundError(f"Config file not found: {cfg_path}")
|
|
|
|
with open(cfg_path, "r", encoding="utf-8") as f:
|
|
raw = yaml.safe_load(f) # whatever the YAML root is
|
|
|
|
# store plain as loaded; store frozen for normal use
|
|
_data_plain = raw
|
|
_data_frozen = _freeze(raw)
|
|
|
|
def config() -> Any:
|
|
"""
|
|
Return the read-only (frozen) config. Call init_config() first.
|
|
"""
|
|
if _data_frozen is None:
|
|
raise RuntimeError("init_config() has not been awaited yet.")
|
|
return _data_frozen
|
|
|
|
def as_plain() -> Any:
|
|
"""
|
|
Return a deep copy of the plain config for safe external use/pretty printing.
|
|
"""
|
|
if _data_plain is None:
|
|
raise RuntimeError("init_config() has not been awaited yet.")
|
|
return copy.deepcopy(_data_plain)
|
|
|
|
def pretty() -> str:
|
|
"""
|
|
YAML pretty string without mappingproxy noise.
|
|
"""
|
|
if _data_plain is None:
|
|
raise RuntimeError("init_config() has not been awaited yet.")
|
|
return yaml.safe_dump(_data_plain, sort_keys=False, allow_unicode=True)
|