Hyungi Ahn
99672292d3
프로덕션 컨테이너는 /app 을 cwd 로 실행하고 import 는 `from api...`, `from core...`, `from workers...` 처럼 무접두 스타일을 사용한다. PR-A 내부 import 가 `from app.policy...`, `from app.ai.envelope` 로 되어 있어서 컨테이너에서 ModuleNotFoundError 발생. 변경: - app/policy/*.py: `from app.policy.X` → `from policy.X` - app/services/prompt_versions.py: lazy import 도 `from policy.prompt_render` - app/ai/envelope.py: 영향 없음 (내부 import 없음) - tests/policy/*.py: 모두 `from policy.X` / `from ai.envelope` 로 통일 - tests/policy/conftest.py: 로컬 pytest 용 sys.path.insert(app/) 추가 (MacBook 에서 repo-root 기준 실행 시 app/ 를 package root 로 취급) CI: pytest tests/policy/ -q → 98 passed (로컬, 동일 결과) 프로덕션: docker exec fastapi python -c "from policy.loader import load_policy" → OK Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
53 lines
1.4 KiB
Python
53 lines
1.4 KiB
Python
"""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_PATH = "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)
|
|
# repo root 기준 상대 경로 — working dir 에 따라 결정
|
|
return Path(DEFAULT_POLICY_PATH)
|
|
|
|
|
|
@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()
|