(self, temp_path: Path)
| 66 | self._replace_with_retry(temp_path) |
| 67 | |
| 68 | def _replace_with_retry(self, temp_path: Path) -> None: |
| 69 | last_error: PermissionError | None = None |
| 70 | for delay in (*self._replace_retry_delays, 0): |
| 71 | try: |
| 72 | temp_path.replace(self.path) |
| 73 | return |
| 74 | except PermissionError as exc: |
| 75 | last_error = exc |
| 76 | if delay <= 0: |
| 77 | break |
| 78 | time.sleep(delay) |
| 79 | if last_error is not None: |
| 80 | raise PermissionError( |
| 81 | f"{last_error}. atomic replace failed after retries for {self.path}" |
| 82 | ) from last_error |