Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.8/site-packages/prompt_toolkit/output/flush_stdout.py: 27%

1from __future__ import annotations 

2 

3import errno 

4import os 

5import sys 

6from contextlib import contextmanager 

7from typing import IO, Iterator, TextIO 

8 

9__all__ = ["flush_stdout"] 

10 

11 

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") 

20 

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) 

36 

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 

54 

55 

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 

66 

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. 

76 

77 try: 

78 # Make blocking if we weren't blocking yet. 

79 if not blocking: 

80 os.set_blocking(fd, True) 

81 

82 yield 

83 

84 finally: 

85 # Restore original blocking mode. 

86 if not blocking: 

87 os.set_blocking(fd, blocking)