Coverage for little_loops / file_utils.py: 92%

1"""Shared file I/O utilities for little-loops.""" 

2 

3from __future__ import annotations 

4 

5import fcntl 

6import json 

7import os 

8import tempfile 

9import time 

10from collections.abc import Generator 

11from contextlib import contextmanager 

12from pathlib import Path 

13from typing import Any 

14 

15 

16def atomic_write(path: Path, content: str, encoding: str = "utf-8") -> None: 

17 """Write *content* to *path* atomically using tempfile + os.replace. 

18 

19 Writes to a sibling temp file in the same directory (same filesystem), 

20 then renames it over the target so readers never observe a partial file. 

21 """ 

22 tmp_fd, tmp_path = tempfile.mkstemp(dir=path.parent, suffix=".tmp") 

23 try: 

24 with os.fdopen(tmp_fd, "w", encoding=encoding) as f: 

25 f.write(content) 

26 os.replace(tmp_path, path) 

27 except Exception: 

28 try: 

29 os.unlink(tmp_path) 

30 except FileNotFoundError: 

31 pass 

32 raise 

33 

34 

35def atomic_write_json(path: Path, data: Any) -> None: 

36 """Atomically write *data* as JSON to *path* (Python port of ``atomic_write_json``). 

37 

38 Mirrors ``hooks/scripts/lib/common.sh:atomic_write_json``: ensures the parent 

39 directory exists, validates the serialized JSON via a ``json.loads`` round-trip 

40 (the bash version uses ``jq empty``), then writes to a sibling tempfile and 

41 ``os.replace``-renames it over the target. 

42 

43 Raises: 

44 ValueError: if the serialized payload fails the round-trip validation 

45 (e.g. NaN/Infinity rejected by ``json.loads``). 

46 """ 

47 path.parent.mkdir(parents=True, exist_ok=True) 

48 # allow_nan=False makes json.dumps raise ValueError on NaN/Infinity, matching 

49 # bash `jq empty`'s rejection of non-strict-JSON values. 

50 payload = json.dumps(data, indent=2, allow_nan=False) 

51 # Defensive round-trip: catches any divergence between dumps output and 

52 # strict-mode parse expectations (parity with `jq empty` validation). 

53 try: 

54 json.loads(payload) 

55 except json.JSONDecodeError as exc: # pragma: no cover — defense in depth 

56 raise ValueError(f"atomic_write_json round-trip validation failed: {exc}") from exc 

57 atomic_write(path, payload) 

58 

59 

60@contextmanager 

61def acquire_lock(path: Path, timeout: float = 10.0) -> Generator[None, None, None]: 

62 """Acquire an exclusive advisory lock on *path*, polled up to *timeout* seconds. 

63 

64 Python port of ``hooks/scripts/lib/common.sh:acquire_lock``. Uses 

65 ``fcntl.flock(LOCK_EX | LOCK_NB)`` in a 0.05s polling loop bounded by 

66 *timeout*; the lock is released when the file descriptor is closed on 

67 context-manager exit (no explicit ``release_lock`` needed). 

68 

69 The bash adapter calls this with ``timeout=3.0`` from precompact and falls 

70 back to a best-effort unlocked write on ``TimeoutError`` to preserve the 

71 bash caller's existing semantics. 

72 

73 Raises: 

74 TimeoutError: if the lock cannot be acquired within *timeout* seconds. 

75 """ 

76 path.parent.mkdir(parents=True, exist_ok=True) 

77 deadline = time.monotonic() + timeout 

78 poll_interval = 0.05 

79 with open(path, "w") as lock_fd: 

80 while True: 

81 try: 

82 fcntl.flock(lock_fd, fcntl.LOCK_EX | fcntl.LOCK_NB) 

83 break 

84 except BlockingIOError: 

85 if time.monotonic() >= deadline: 

86 raise TimeoutError( 

87 f"acquire_lock: could not acquire {path} within {timeout}s" 

88 ) from None 

89 time.sleep(poll_interval) 

90 try: 

91 yield 

92 finally: 

93 try: 

94 fcntl.flock(lock_fd, fcntl.LOCK_UN) 

95 except OSError: 

96 pass