73 lines
2.2 KiB
Python
73 lines
2.2 KiB
Python
"""Safe, small input helpers for harness contracts."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
from pathlib import Path
|
|
from typing import Any, TypeVar
|
|
|
|
import yaml
|
|
from pydantic import BaseModel
|
|
|
|
|
|
MAX_INPUT_BYTES = 2 * 1024 * 1024
|
|
ModelT = TypeVar("ModelT", bound=BaseModel)
|
|
|
|
|
|
class InputError(ValueError):
|
|
"""Raised when an input file cannot be safely decoded as a model payload."""
|
|
|
|
|
|
def load_mapping(path: str | Path, *, max_bytes: int = MAX_INPUT_BYTES) -> dict[str, Any]:
|
|
"""Load one JSON/YAML mapping without constructing arbitrary Python objects."""
|
|
|
|
input_path = Path(path)
|
|
try:
|
|
size = input_path.stat().st_size
|
|
except OSError as exc:
|
|
raise InputError(f"입력 파일을 읽을 수 없습니다: {input_path}") from exc
|
|
|
|
if size > max_bytes:
|
|
raise InputError(f"입력 파일이 {max_bytes}바이트 제한을 초과했습니다: {input_path}")
|
|
|
|
try:
|
|
text = input_path.read_text(encoding="utf-8")
|
|
except (OSError, UnicodeDecodeError) as exc:
|
|
raise InputError(f"입력 파일은 UTF-8 텍스트여야 합니다: {input_path}") from exc
|
|
|
|
suffix = input_path.suffix.casefold()
|
|
try:
|
|
if suffix == ".json":
|
|
value = json.loads(text)
|
|
elif suffix in {".yaml", ".yml"}:
|
|
value = yaml.safe_load(text)
|
|
else:
|
|
raise InputError("지원 형식은 .json, .yaml, .yml입니다.")
|
|
except (json.JSONDecodeError, yaml.YAMLError) as exc:
|
|
raise InputError(f"JSON/YAML 구문이 올바르지 않습니다: {input_path}") from exc
|
|
|
|
if not isinstance(value, dict):
|
|
raise InputError(f"입력 최상위 값은 객체(mapping)여야 합니다: {input_path}")
|
|
return value
|
|
|
|
|
|
def load_model(path: str | Path, model_type: type[ModelT]) -> ModelT:
|
|
"""Load and validate a Pydantic contract from JSON/YAML."""
|
|
|
|
return model_type.model_validate(load_mapping(path))
|
|
|
|
|
|
def dump_json(model: BaseModel) -> str:
|
|
"""Serialize a contract deterministically for audit-friendly output."""
|
|
|
|
return json.dumps(
|
|
model.model_dump(mode="json"),
|
|
ensure_ascii=False,
|
|
indent=2,
|
|
sort_keys=True,
|
|
)
|
|
|
|
|
|
__all__ = ["InputError", "MAX_INPUT_BYTES", "dump_json", "load_mapping", "load_model"]
|
|
|