112 lines
3.2 KiB
Python
112 lines
3.2 KiB
Python
"""Centralized environment variable access for PAL MCP Server."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import os
|
|
from collections.abc import Mapping
|
|
from contextlib import contextmanager
|
|
from pathlib import Path
|
|
|
|
try:
|
|
from dotenv import dotenv_values, load_dotenv
|
|
except ImportError: # pragma: no cover - optional dependency
|
|
dotenv_values = None # type: ignore[assignment]
|
|
load_dotenv = None # type: ignore[assignment]
|
|
|
|
_PROJECT_ROOT = Path(__file__).resolve().parent.parent
|
|
_ENV_PATH = _PROJECT_ROOT / ".env"
|
|
|
|
_DOTENV_VALUES: dict[str, str | None] = {}
|
|
_FORCE_ENV_OVERRIDE = False
|
|
|
|
|
|
def _read_dotenv_values() -> dict[str, str | None]:
|
|
if dotenv_values is not None and _ENV_PATH.exists():
|
|
loaded = dotenv_values(_ENV_PATH)
|
|
return dict(loaded)
|
|
return {}
|
|
|
|
|
|
def _compute_force_override(values: Mapping[str, str | None]) -> bool:
|
|
raw = (values.get("PAL_MCP_FORCE_ENV_OVERRIDE") or "false").strip().lower()
|
|
return raw == "true"
|
|
|
|
|
|
def reload_env(dotenv_mapping: Mapping[str, str | None] | None = None) -> None:
|
|
"""Reload .env values and recompute override semantics.
|
|
|
|
Args:
|
|
dotenv_mapping: Optional mapping used instead of reading the .env file.
|
|
Intended for tests; when provided, load_dotenv is not invoked.
|
|
"""
|
|
|
|
global _DOTENV_VALUES, _FORCE_ENV_OVERRIDE
|
|
|
|
if dotenv_mapping is not None:
|
|
_DOTENV_VALUES = dict(dotenv_mapping)
|
|
_FORCE_ENV_OVERRIDE = _compute_force_override(_DOTENV_VALUES)
|
|
return
|
|
|
|
_DOTENV_VALUES = _read_dotenv_values()
|
|
_FORCE_ENV_OVERRIDE = _compute_force_override(_DOTENV_VALUES)
|
|
|
|
if load_dotenv is not None and _ENV_PATH.exists():
|
|
load_dotenv(dotenv_path=_ENV_PATH, override=_FORCE_ENV_OVERRIDE)
|
|
|
|
|
|
reload_env()
|
|
|
|
|
|
def env_override_enabled() -> bool:
|
|
"""Return True when PAL_MCP_FORCE_ENV_OVERRIDE is enabled via the .env file."""
|
|
|
|
return _FORCE_ENV_OVERRIDE
|
|
|
|
|
|
def get_env(key: str, default: str | None = None) -> str | None:
|
|
"""Retrieve environment variables respecting PAL_MCP_FORCE_ENV_OVERRIDE."""
|
|
|
|
if env_override_enabled():
|
|
if key in _DOTENV_VALUES:
|
|
value = _DOTENV_VALUES[key]
|
|
return value if value is not None else default
|
|
return default
|
|
|
|
return os.getenv(key, default)
|
|
|
|
|
|
def get_env_bool(key: str, default: bool = False) -> bool:
|
|
"""Boolean helper that respects override semantics."""
|
|
|
|
raw_default = "true" if default else "false"
|
|
raw_value = get_env(key, raw_default)
|
|
return (raw_value or raw_default).strip().lower() == "true"
|
|
|
|
|
|
def get_all_env() -> dict[str, str | None]:
|
|
"""Expose the loaded .env mapping for diagnostics/logging."""
|
|
|
|
return dict(_DOTENV_VALUES)
|
|
|
|
|
|
@contextmanager
|
|
def suppress_env_vars(*names: str):
|
|
"""Temporarily remove environment variables during the context.
|
|
|
|
Args:
|
|
names: Environment variable names to remove. Empty or falsy names are ignored.
|
|
"""
|
|
|
|
removed: dict[str, str] = {}
|
|
try:
|
|
for name in names:
|
|
if not name:
|
|
continue
|
|
if name in os.environ:
|
|
removed[name] = os.environ[name]
|
|
del os.environ[name]
|
|
yield
|
|
finally:
|
|
for name, value in removed.items():
|
|
os.environ[name] = value
|