104 lines
2.9 KiB
Python
104 lines
2.9 KiB
Python
"""Shared utility functions for hermes-agent."""
|
|
|
|
import json
|
|
import os
|
|
import tempfile
|
|
from pathlib import Path
|
|
from typing import Any, Union
|
|
|
|
import yaml
|
|
|
|
|
|
def atomic_json_write(
|
|
path: Union[str, Path],
|
|
data: Any,
|
|
*,
|
|
indent: int = 2,
|
|
**dump_kwargs: Any,
|
|
) -> None:
|
|
"""Write JSON data to a file atomically.
|
|
|
|
Uses temp file + fsync + os.replace to ensure the target file is never
|
|
left in a partially-written state. If the process crashes mid-write,
|
|
the previous version of the file remains intact.
|
|
|
|
Args:
|
|
path: Target file path (will be created or overwritten).
|
|
data: JSON-serializable data to write.
|
|
indent: JSON indentation (default 2).
|
|
**dump_kwargs: Additional keyword args forwarded to json.dump(), such
|
|
as default=str for non-native types.
|
|
"""
|
|
path = Path(path)
|
|
path.parent.mkdir(parents=True, exist_ok=True)
|
|
|
|
fd, tmp_path = tempfile.mkstemp(
|
|
dir=str(path.parent),
|
|
prefix=f".{path.stem}_",
|
|
suffix=".tmp",
|
|
)
|
|
try:
|
|
with os.fdopen(fd, "w", encoding="utf-8") as f:
|
|
json.dump(
|
|
data,
|
|
f,
|
|
indent=indent,
|
|
ensure_ascii=False,
|
|
**dump_kwargs,
|
|
)
|
|
f.flush()
|
|
os.fsync(f.fileno())
|
|
os.replace(tmp_path, path)
|
|
except BaseException:
|
|
try:
|
|
os.unlink(tmp_path)
|
|
except OSError:
|
|
pass
|
|
raise
|
|
|
|
|
|
def atomic_yaml_write(
|
|
path: Union[str, Path],
|
|
data: Any,
|
|
*,
|
|
default_flow_style: bool = False,
|
|
sort_keys: bool = False,
|
|
extra_content: str | None = None,
|
|
) -> None:
|
|
"""Write YAML data to a file atomically.
|
|
|
|
Uses temp file + fsync + os.replace to ensure the target file is never
|
|
left in a partially-written state. If the process crashes mid-write,
|
|
the previous version of the file remains intact.
|
|
|
|
Args:
|
|
path: Target file path (will be created or overwritten).
|
|
data: YAML-serializable data to write.
|
|
default_flow_style: YAML flow style (default False).
|
|
sort_keys: Whether to sort dict keys (default False).
|
|
extra_content: Optional string to append after the YAML dump
|
|
(e.g. commented-out sections for user reference).
|
|
"""
|
|
path = Path(path)
|
|
path.parent.mkdir(parents=True, exist_ok=True)
|
|
|
|
fd, tmp_path = tempfile.mkstemp(
|
|
dir=str(path.parent),
|
|
prefix=f".{path.stem}_",
|
|
suffix=".tmp",
|
|
)
|
|
try:
|
|
with os.fdopen(fd, "w", encoding="utf-8") as f:
|
|
yaml.dump(data, f, default_flow_style=default_flow_style, sort_keys=sort_keys)
|
|
if extra_content:
|
|
f.write(extra_content)
|
|
f.flush()
|
|
os.fsync(f.fileno())
|
|
os.replace(tmp_path, path)
|
|
except BaseException:
|
|
try:
|
|
os.unlink(tmp_path)
|
|
except OSError:
|
|
pass
|
|
raise
|