Frank Schwenk
b09f90bfcc
CI / lint-and-test (pull_request) Failing after 9s
Prevent worker/web clobbering of meta variants via flock and merge-by-id, make filter timeouts thread-local, harden job-id/stem sanitization, migrate TemplateResponse API, and remove the compare-bg experiment. Co-authored-by: Cursor <cursoragent@cursor.com>
43 lines
1.2 KiB
Python
43 lines
1.2 KiB
Python
"""Small shared I/O helpers."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import fcntl
|
|
import json
|
|
import os
|
|
from collections.abc import Iterator
|
|
from contextlib import contextmanager
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
|
|
def read_json(path: Path, *, default: Any = None) -> Any:
|
|
"""Read JSON from *path*, returning *default* when missing or corrupt."""
|
|
if not path.exists():
|
|
return default
|
|
try:
|
|
return json.loads(path.read_text(encoding="utf-8"))
|
|
except (OSError, ValueError):
|
|
return default
|
|
|
|
|
|
def write_json_atomic(path: Path, data: Any) -> None:
|
|
"""Write JSON via a temp file and atomic replace."""
|
|
path.parent.mkdir(parents=True, exist_ok=True)
|
|
tmp = path.with_suffix(path.suffix + ".tmp")
|
|
tmp.write_text(json.dumps(data, indent=2, ensure_ascii=False), encoding="utf-8")
|
|
tmp.replace(path)
|
|
|
|
|
|
@contextmanager
|
|
def file_lock(lock_path: Path) -> Iterator[None]:
|
|
"""Exclusive advisory lock via ``fcntl.flock`` (cross-process)."""
|
|
lock_path.parent.mkdir(parents=True, exist_ok=True)
|
|
fd = os.open(lock_path, os.O_CREAT | os.O_RDWR, 0o644)
|
|
try:
|
|
fcntl.flock(fd, fcntl.LOCK_EX)
|
|
yield
|
|
finally:
|
|
fcntl.flock(fd, fcntl.LOCK_UN)
|
|
os.close(fd)
|