(self, content: str)
| 331 | return "\n".join(lines) + "\n" |
| 332 | |
| 333 | def _atomic_write(self, content: str) -> None: |
| 334 | tmp = tempfile.NamedTemporaryFile( |
| 335 | mode="w", |
| 336 | encoding="utf-8", |
| 337 | dir=str(self.config_path.parent), |
| 338 | delete=False, |
| 339 | ) |
| 340 | tmp_path = Path(tmp.name) |
| 341 | try: |
| 342 | tmp.write(content) |
| 343 | tmp.flush() |
| 344 | os.fsync(tmp.fileno()) |
| 345 | tmp.close() |
| 346 | if self.config_path.exists(): |
| 347 | st = self.config_path.stat() |
| 348 | os.chmod(tmp_path, stat.S_IMODE(st.st_mode)) |
| 349 | else: |
| 350 | os.chmod(tmp_path, 0o600) |
| 351 | os.replace(tmp_path, self.config_path) |
| 352 | except Exception: |
| 353 | try: |
| 354 | tmp.close() |
| 355 | except Exception: |
| 356 | pass |
| 357 | try: |
| 358 | tmp_path.unlink(missing_ok=True) |
| 359 | except Exception: |
| 360 | pass |
| 361 | raise |
no test coverage detected