(
filename: str | os.PathLike[str],
mode: str = "r",
encoding: str | None = None,
errors: str | None = "strict",
atomic: bool = False,
)
| 369 | |
| 370 | |
| 371 | def open_stream( |
| 372 | filename: str | os.PathLike[str], |
| 373 | mode: str = "r", |
| 374 | encoding: str | None = None, |
| 375 | errors: str | None = "strict", |
| 376 | atomic: bool = False, |
| 377 | ) -> tuple[t.IO[t.Any], bool]: |
| 378 | binary = "b" in mode |
| 379 | filename = os.fspath(filename) |
| 380 | |
| 381 | # Standard streams first. These are simple because they ignore the |
| 382 | # atomic flag. Use fsdecode to handle Path("-"). |
| 383 | if os.fsdecode(filename) == "-": |
| 384 | if any(m in mode for m in ["w", "a", "x"]): |
| 385 | if binary: |
| 386 | return get_binary_stdout(), False |
| 387 | return get_text_stdout(encoding=encoding, errors=errors), False |
| 388 | if binary: |
| 389 | return get_binary_stdin(), False |
| 390 | return get_text_stdin(encoding=encoding, errors=errors), False |
| 391 | |
| 392 | # Non-atomic writes directly go out through the regular open functions. |
| 393 | if not atomic: |
| 394 | return _wrap_io_open(filename, mode, encoding, errors), True |
| 395 | |
| 396 | # Some usability stuff for atomic writes |
| 397 | if "a" in mode: |
| 398 | raise ValueError( |
| 399 | "Appending to an existing file is not supported, because that" |
| 400 | " would involve an expensive `copy`-operation to a temporary" |
| 401 | " file. Open the file in normal `w`-mode and copy explicitly" |
| 402 | " if that's what you're after." |
| 403 | ) |
| 404 | if "x" in mode: |
| 405 | raise ValueError("Use the `overwrite`-parameter instead.") |
| 406 | if "w" not in mode: |
| 407 | raise ValueError("Atomic writes only make sense with `w`-mode.") |
| 408 | |
| 409 | # Atomic writes are more complicated. They work by opening a file |
| 410 | # as a proxy in the same folder and then using the fdopen |
| 411 | # functionality to wrap it in a Python file. Then we wrap it in an |
| 412 | # atomic file that moves the file over on close. |
| 413 | import errno |
| 414 | import random |
| 415 | |
| 416 | try: |
| 417 | perm: int | None = os.stat(filename).st_mode |
| 418 | except OSError: |
| 419 | perm = None |
| 420 | |
| 421 | flags = os.O_RDWR | os.O_CREAT | os.O_EXCL |
| 422 | |
| 423 | if binary: |
| 424 | flags |= getattr(os, "O_BINARY", 0) |
| 425 | |
| 426 | while True: |
| 427 | tmp_filename = os.path.join( |
| 428 | os.path.dirname(filename), |
no test coverage detected