Write breadcrumb data to a JSON file with file locking. Automatically adds/updates timestamp if not present in data. Thread-safe: Uses file locking with 5-second timeout and stale lock recovery. Args: breadcrumb_path: Path to the breadcrumb file data: Dictionary wi
(breadcrumb_path: Path, data: dict[str, Any])
| 58 | |
| 59 | |
| 60 | def write_breadcrumb(breadcrumb_path: Path, data: dict[str, Any]) -> bool: |
| 61 | """ |
| 62 | Write breadcrumb data to a JSON file with file locking. |
| 63 | |
| 64 | Automatically adds/updates timestamp if not present in data. |
| 65 | Thread-safe: Uses file locking with 5-second timeout and stale lock recovery. |
| 66 | |
| 67 | Args: |
| 68 | breadcrumb_path: Path to the breadcrumb file |
| 69 | data: Dictionary with breadcrumb data to write |
| 70 | |
| 71 | Returns: |
| 72 | True if write succeeded, False otherwise |
| 73 | """ |
| 74 | try: |
| 75 | # Ensure parent directory exists |
| 76 | breadcrumb_path.parent.mkdir(parents=True, exist_ok=True) |
| 77 | |
| 78 | # Add timestamp if not present |
| 79 | if "timestamp" not in data: |
| 80 | data["timestamp"] = datetime.now().isoformat() |
| 81 | |
| 82 | # Use file locking with 5-second timeout to prevent concurrent write corruption |
| 83 | with write_lock(breadcrumb_path, timeout=5.0, operation="breadcrumb_write"): |
| 84 | with open(breadcrumb_path, "w") as f: |
| 85 | json.dump(data, f, indent=2) |
| 86 | |
| 87 | return True |
| 88 | |
| 89 | except TimeoutError as e: |
| 90 | logger.error(f"Lock timeout writing breadcrumb {breadcrumb_path}: {e}") |
| 91 | return False |
| 92 | except OSError as e: |
| 93 | logger.warning(f"Failed to write breadcrumb {breadcrumb_path}: {e}") |
| 94 | return False |
| 95 | |
| 96 | |
| 97 | def is_download_complete(breadcrumb_path: Path) -> bool: |
no test coverage detected