Best-effort function to write data to a path atomically. Be prepared to handle a FileExistsError if concurrent writing of the temporary file is attempted.
(path, data, mode=0o666)
| 198 | |
| 199 | |
| 200 | def _write_atomic(path, data, mode=0o666): |
| 201 | """Best-effort function to write data to a path atomically. |
| 202 | Be prepared to handle a FileExistsError if concurrent writing of the |
| 203 | temporary file is attempted.""" |
| 204 | # id() is used to generate a pseudo-random filename. |
| 205 | path_tmp = f'{path}.{id(path)}' |
| 206 | fd = _os.open(path_tmp, |
| 207 | _os.O_EXCL | _os.O_CREAT | _os.O_WRONLY, mode & 0o666) |
| 208 | try: |
| 209 | # We first write data to a temporary file, and then use os.replace() to |
| 210 | # perform an atomic rename. |
| 211 | with _io.open(fd, 'wb') as file: |
| 212 | file.write(data) |
| 213 | _os.replace(path_tmp, path) |
| 214 | except OSError: |
| 215 | try: |
| 216 | _os.unlink(path_tmp) |
| 217 | except OSError: |
| 218 | pass |
| 219 | raise |
| 220 | |
| 221 | |
| 222 | _code_type = type(_write_atomic.__code__) |