(self, path: str, contents: str | bytes)
| 24 | return os.path.join(self.root, path) |
| 25 | |
| 26 | def write(self, path: str, contents: str | bytes) -> None: |
| 27 | full_path = self.get_full_path(path) |
| 28 | os.makedirs(os.path.dirname(full_path), exist_ok=True) |
| 29 | mode = 'w' if isinstance(contents, str) else 'wb' |
| 30 | |
| 31 | # Use atomic write: write to temp file, then rename |
| 32 | # This prevents race conditions where concurrent writes could corrupt the file |
| 33 | temp_path = f'{full_path}.tmp.{os.getpid()}.{threading.get_ident()}' |
| 34 | try: |
| 35 | with open(temp_path, mode) as f: |
| 36 | f.write(contents) |
| 37 | f.flush() |
| 38 | os.fsync(f.fileno()) |
| 39 | os.replace(temp_path, full_path) |
| 40 | except Exception: |
| 41 | if os.path.exists(temp_path): |
| 42 | os.remove(temp_path) |
| 43 | raise |
| 44 | |
| 45 | def write_from_path(self, path: str, source_path: str) -> None: |
| 46 | # shutil.copyfile streams in chunks (never the whole file in RAM); keep |
nothing calls this directly
no test coverage detected