Shortcuts on this page
r m x toggle line displays
j k next/prev highlighted chunk
0 (zero) top of page
1 (one) first highlighted chunk
1from __future__ import annotations
3import errno
4import os
5import sys
6from collections.abc import Iterator
7from contextlib import contextmanager
8from typing import IO, TextIO
10__all__ = ["flush_stdout"]
13def flush_stdout(stdout: TextIO, data: str) -> None:
14 # If the IO object has an `encoding` and `buffer` attribute, it means that
15 # we can access the underlying BinaryIO object and write into it in binary
16 # mode. This is preferred if possible.
17 # NOTE: When used in a Jupyter notebook, don't write binary.
18 # `ipykernel.iostream.OutStream` has an `encoding` attribute, but not
19 # a `buffer` attribute, so we can't write binary in it.
20 has_binary_io = hasattr(stdout, "encoding") and hasattr(stdout, "buffer")
22 try:
23 # Ensure that `stdout` is made blocking when writing into it.
24 # Otherwise, when uvloop is activated (which makes stdout
25 # non-blocking), and we write big amounts of text, then we get a
26 # `BlockingIOError` here.
27 with _blocking_io(stdout):
28 # (We try to encode ourself, because that way we can replace
29 # characters that don't exist in the character set, avoiding
30 # UnicodeEncodeError crashes. E.g. u'\xb7' does not appear in 'ascii'.)
31 # My Arch Linux installation of july 2015 reported 'ANSI_X3.4-1968'
32 # for sys.stdout.encoding in xterm.
33 if has_binary_io:
34 stdout.buffer.write(data.encode(stdout.encoding or "utf-8", "replace"))
35 else:
36 stdout.write(data)
38 stdout.flush()
39 except OSError as e:
40 if e.args and e.args[0] == errno.EINTR:
41 # Interrupted system call. Can happen in case of a window
42 # resize signal. (Just ignore. The resize handler will render
43 # again anyway.)
44 pass
45 elif e.args and e.args[0] == 0:
46 # This can happen when there is a lot of output and the user
47 # sends a KeyboardInterrupt by pressing Control-C. E.g. in
48 # a Python REPL when we execute "while True: print('test')".
49 # (The `ptpython` REPL uses this `Output` class instead of
50 # `stdout` directly -- in order to be network transparent.)
51 # So, just ignore.
52 pass
53 else:
54 raise
57@contextmanager
58def _blocking_io(io: IO[str]) -> Iterator[None]:
59 """
60 Ensure that the FD for `io` is set to blocking in here.
61 """
62 if sys.platform == "win32":
63 # On Windows, the `os` module doesn't have a `get/set_blocking`
64 # function.
65 yield
66 return
68 try:
69 fd = io.fileno()
70 blocking = os.get_blocking(fd)
71 except: # noqa
72 # Failed somewhere.
73 # `get_blocking` can raise `OSError`.
74 # The io object can raise `AttributeError` when no `fileno()` method is
75 # present if we're not a real file object.
76 blocking = True # Assume we're good, and don't do anything.
78 try:
79 # Make blocking if we weren't blocking yet.
80 if not blocking:
81 os.set_blocking(fd, True)
83 yield
85 finally:
86 # Restore original blocking mode.
87 if not blocking:
88 os.set_blocking(fd, blocking)