Spaces:
Sleeping
Sleeping
File size: 3,530 Bytes
990895d | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 | """Provide locked JSON/JSONL persistence and atomic file replacement.
Mutations use a sibling lock file so append and full rewrites cannot overlap.
"""
import json
import os
from contextlib import contextmanager
from pathlib import Path
from typing import Any, Iterator
@contextmanager
def file_lock(path: Path) -> Iterator[None]:
"""Hold an exclusive process lock for a data path."""
lock_path = path.with_name(f"{path.name}.lock")
lock_path.parent.mkdir(parents=True, exist_ok=True)
with lock_path.open("a+b") as handle:
if os.name == "nt":
import msvcrt
if handle.tell() == 0:
handle.write(b"\0")
handle.flush()
handle.seek(0)
msvcrt.locking(handle.fileno(), msvcrt.LK_LOCK, 1)
else:
import fcntl
fcntl.flock(handle.fileno(), fcntl.LOCK_EX)
try:
yield
finally:
if os.name == "nt":
handle.seek(0)
msvcrt.locking(handle.fileno(), msvcrt.LK_UNLCK, 1)
else:
fcntl.flock(handle.fileno(), fcntl.LOCK_UN)
def atomic_write_text(path: Path, text: str) -> None:
"""Replace a file atomically after flushing its contents to disk."""
path.parent.mkdir(parents=True, exist_ok=True)
temporary = path.with_name(f"{path.name}.tmp")
with temporary.open("w", encoding="utf-8", newline="\n") as handle:
handle.write(text)
handle.flush()
os.fsync(handle.fileno())
os.replace(temporary, path)
def read_json(path: Path) -> dict[str, Any] | None:
"""Read a JSON object, returning None when the file is absent."""
if not path.exists():
return None
with path.open("r", encoding="utf-8") as handle:
value = json.load(handle)
if not isinstance(value, dict):
raise ValueError(f"{path.name} must contain a JSON object")
return value
def write_json(path: Path, value: dict[str, Any]) -> None:
"""Write a JSON object under a lock using atomic replacement."""
payload = json.dumps(value, ensure_ascii=False, indent=2) + "\n"
with file_lock(path):
atomic_write_text(path, payload)
def read_jsonl(path: Path) -> list[dict[str, Any]]:
"""Read non-empty JSONL records in file order."""
if not path.exists():
return []
records: list[dict[str, Any]] = []
with path.open("r", encoding="utf-8") as handle:
for line in handle:
if line.strip():
value = json.loads(line)
if not isinstance(value, dict):
raise ValueError(f"{path.name} contains a non-object record")
records.append(value)
return records
def append_jsonl(path: Path, value: dict[str, Any]) -> None:
"""Append one durable JSONL record while holding the file lock."""
path.parent.mkdir(parents=True, exist_ok=True)
line = json.dumps(value, ensure_ascii=False, separators=(",", ":")) + "\n"
with file_lock(path):
with path.open("a", encoding="utf-8", newline="\n") as handle:
handle.write(line)
handle.flush()
os.fsync(handle.fileno())
def rewrite_jsonl(path: Path, values: list[dict[str, Any]]) -> None:
"""Atomically replace a JSONL file while holding its lock."""
text = "".join(
json.dumps(value, ensure_ascii=False, separators=(",", ":")) + "\n"
for value in values
)
with file_lock(path):
atomic_write_text(path, text)
|