eomsangdonandClaude Sonnet 5
9bf3258509
Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_0148XFUpPfpiuTjxF1EK9c95
95 lines
3.4 KiB
Python
95 lines
3.4 KiB
Python
"""M02 마스터 템플릿 — 시스템 층 저장소 (`resources/master_template/{table,drawing}/<이름>.json`).
|
|
|
|
M01 `Store` 방식 — 판(파일 sha256 앞 16자) · 판이 다르면 409 · 원자 쓰기.
|
|
권한은 등록하는 쪽(`main.py`)이 붙임.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import hashlib
|
|
import json
|
|
import re
|
|
import threading
|
|
from datetime import datetime
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
from common_util.common_util_json import atomic_write_json
|
|
|
|
FOLDER: Path = Path(__file__).resolve().parent.parent / "resources" / "master_template"
|
|
KINDS = ("table", "drawing") # 시험은 FOLDER 를 사본으로 바꿈
|
|
_BAD_NAME = re.compile(r'[\/:*?"<>|\x00-\x1f]')
|
|
# ponytail: 저장은 한 번에 하나(프로세스 안 잠금) · 서버를 여럿 띄우면 파일 잠금으로
|
|
_LOCK = threading.Lock()
|
|
|
|
|
|
class StoreError(Exception):
|
|
def __init__(self, status: int, detail):
|
|
super().__init__(detail)
|
|
self.status, self.detail = status, detail
|
|
|
|
|
|
def version_of(raw: bytes) -> str:
|
|
return hashlib.sha256(raw).hexdigest()[:16]
|
|
|
|
|
|
def _path(kind: str, name: str) -> Path:
|
|
if kind not in KINDS:
|
|
raise StoreError(404, f"없는 종류 「{kind}」")
|
|
if not name or name != name.strip() or name.startswith(".") or _BAD_NAME.search(name):
|
|
raise StoreError(422, f"쓸 수 없는 이름 「{name}」")
|
|
return FOLDER / kind / f"{name}.json"
|
|
|
|
|
|
def _info(kind: str, path: Path) -> dict[str, Any]:
|
|
raw = path.read_bytes()
|
|
stamp = datetime.fromtimestamp(path.stat().st_mtime).isoformat(timespec="seconds")
|
|
return {"종류": kind, "이름": path.stem, "판": version_of(raw), "수정일": stamp}
|
|
|
|
|
|
def list_all() -> list[dict[str, Any]]:
|
|
rows = [
|
|
_info(kind, p)
|
|
for kind in KINDS
|
|
for p in sorted((FOLDER / kind).glob("*.json"))
|
|
if not p.name.startswith(".")
|
|
]
|
|
return rows
|
|
|
|
|
|
def read(kind: str, name: str) -> dict[str, Any]:
|
|
path = _path(kind, name)
|
|
if not path.is_file():
|
|
raise StoreError(404, f"없는 양식 「{name}」")
|
|
raw = path.read_bytes()
|
|
return {"종류": kind, "이름": name, "판": version_of(raw), "문서": json.loads(raw)}
|
|
|
|
|
|
def _check_skeleton(kind: str, doc: Any) -> None:
|
|
"""빈 문서·뼈대 없는 문서는 거절 — 표는 열 목록 · 도면은 entities 목록."""
|
|
key = "열" if kind == "table" else "entities"
|
|
if not isinstance(doc, dict) or not isinstance(doc.get(key), list):
|
|
raise StoreError(400, f"양식 문서에 「{key}」 목록이 없음 — 저장하지 않음")
|
|
|
|
|
|
def write(kind: str, name: str, version: str, doc: Any) -> dict[str, Any]:
|
|
"""`version` 이 빈 글이면 새로 만듦(이미 있으면 409) · 아니면 그 판일 때만 덮어씀."""
|
|
path = _path(kind, name)
|
|
_check_skeleton(kind, doc)
|
|
with _LOCK:
|
|
have = version_of(path.read_bytes()) if path.is_file() else ""
|
|
if have != (version or ""):
|
|
raise StoreError(409, {"stale": [name], "판": have})
|
|
atomic_write_json(path, doc)
|
|
return _info(kind, path)
|
|
|
|
|
|
def delete(kind: str, name: str, version: str | None = None) -> None:
|
|
path = _path(kind, name)
|
|
with _LOCK:
|
|
if not path.is_file():
|
|
raise StoreError(404, f"없는 양식 「{name}」")
|
|
if version is not None and version_of(path.read_bytes()) != version:
|
|
raise StoreError(409, {"stale": [name], "판": version_of(path.read_bytes())})
|
|
path.unlink()
|