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