48 lines
1.5 KiB
Python
48 lines
1.5 KiB
Python
from __future__ import annotations
|
|
|
|
import json
|
|
import os
|
|
from pathlib import Path
|
|
from uuid import uuid4
|
|
|
|
from cloud.internal_api.models import HostGovernancePolicyModel
|
|
|
|
|
|
class HostPolicyCacheError(RuntimeError):
|
|
"""Raised when the locally cached non-secret Host policy is invalid."""
|
|
|
|
|
|
class HostPolicyCacheStore:
|
|
"""Atomically persists only the Cloud-supplied, non-secret policy cache."""
|
|
|
|
def __init__(self, path: str | Path) -> None:
|
|
self.path = Path(path)
|
|
|
|
def load(self) -> HostGovernancePolicyModel | None:
|
|
if not self.path.exists():
|
|
return None
|
|
try:
|
|
payload = json.loads(self.path.read_text(encoding="utf-8"))
|
|
return HostGovernancePolicyModel.model_validate(payload)
|
|
except (OSError, ValueError, json.JSONDecodeError) as exc:
|
|
raise HostPolicyCacheError("Host policy cache is invalid") from exc
|
|
|
|
def save(self, policy: HostGovernancePolicyModel) -> None:
|
|
self.path.parent.mkdir(parents=True, exist_ok=True)
|
|
temporary = self.path.with_name(f".{self.path.name}.{uuid4().hex}.tmp")
|
|
try:
|
|
temporary.write_text(
|
|
policy.model_dump_json(indent=2) + "\n",
|
|
encoding="utf-8",
|
|
)
|
|
os.replace(temporary, self.path)
|
|
finally:
|
|
if temporary.exists():
|
|
temporary.unlink()
|
|
|
|
def clear(self) -> None:
|
|
try:
|
|
self.path.unlink()
|
|
except FileNotFoundError:
|
|
return
|