Write status file atomically to prevent corruption during writes. Args: status: Status dictionary to write
(status: dict[str, Any])
| 151 | |
| 152 | |
| 153 | def write_status_file_atomic(status: dict[str, Any]) -> None: |
| 154 | """Write status file atomically to prevent corruption during writes. |
| 155 | |
| 156 | Args: |
| 157 | status: Status dictionary to write |
| 158 | """ |
| 159 | temp_file = STATUS_FILE.with_suffix(".tmp") |
| 160 | |
| 161 | try: |
| 162 | with open(temp_file, "w") as f: |
| 163 | json.dump(status, f, indent=2) |
| 164 | |
| 165 | # Atomic rename |
| 166 | temp_file.replace(STATUS_FILE) |
| 167 | |
| 168 | except KeyboardInterrupt as ki: |
| 169 | handle_keyboard_interrupt(ki) |
| 170 | temp_file.unlink(missing_ok=True) |
| 171 | raise |
| 172 | except Exception as e: |
| 173 | logging.error(f"Failed to write status file: {e}") |
| 174 | temp_file.unlink(missing_ok=True) |
| 175 | |
| 176 | |
| 177 | def update_status(state: DaemonState, message: str, **kwargs: Any) -> None: |
no test coverage detected