"""domain_policy.yaml loader with lru_cache.""" from __future__ import annotations import os from functools import lru_cache from pathlib import Path import yaml from policy.schema import DomainPolicy DEFAULT_POLICY_FILENAME = "domain_policy.yaml" POLICY_PATH_ENV = "POLICY_PATH" def _resolve_path(path: str | None) -> Path: if path is not None: return Path(path) env_path = os.environ.get(POLICY_PATH_ENV) if env_path: return Path(env_path) # 검색 순서 (multi-env 호환): # 1. cwd / domain_policy.yaml 로컬 pytest (repo-root 실행) # 2. /app / domain_policy.yaml container bind-mount 경로 # 3. /app/../domain_policy.yaml container: /app 의 parent # 4. .parent.parent.parent / yaml policy 패키지 기준 repo-root candidates = [ Path.cwd() / DEFAULT_POLICY_FILENAME, Path("/app") / DEFAULT_POLICY_FILENAME, Path("/app").parent / DEFAULT_POLICY_FILENAME, Path(__file__).resolve().parent.parent.parent / DEFAULT_POLICY_FILENAME, ] for c in candidates: if c.is_file(): return c # 찾지 못한 경우 첫 후보 반환 → 나중에 FileNotFoundError 로 명확히 실패 return candidates[0] @lru_cache(maxsize=8) def _load_cached(resolved: str) -> DomainPolicy: text = Path(resolved).read_text(encoding="utf-8") raw = yaml.safe_load(text) return DomainPolicy.model_validate(raw) def load_policy(path: str | None = None) -> DomainPolicy: """Load policy yaml and validate via pydantic. Cache key = resolved absolute path (문자열). 테스트에서 다른 path 주면 별도 캐시. """ resolved = str(_resolve_path(path).resolve()) return _load_cached(resolved) def clear_cache() -> None: """테스트용 — 연속 호출 시 서로 다른 yaml 을 반영해야 할 때.""" _load_cached.cache_clear() def read_policy_bytes(path: str | None = None) -> bytes: """policy_version hash 계산용 — yaml 원본 바이트.""" resolved = _resolve_path(path).resolve() return resolved.read_bytes()