Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.11/site-packages/IPython/utils/io.py: 31%

1# encoding: utf-8 

2""" 

3IO related utilities. 

4""" 

5 

6# Copyright (c) IPython Development Team. 

7# Distributed under the terms of the Modified BSD License. 

8 

9 

10 

11import atexit 

12import os 

13import sys 

14import tempfile 

15from pathlib import Path 

16from warnings import warn 

17 

18from IPython.utils.decorators import undoc 

19from .capture import CapturedIO, capture_output 

20from io import StringIO 

21from typing import Union 

22 

23 

24class Tee: 

25 """A class to duplicate an output stream to stdout/err. 

26 

27 This works in a manner very similar to the Unix 'tee' command. 

28 

29 When the object is closed or deleted, it closes the original file given to 

30 it for duplication. 

31 """ 

32 # Inspired by: 

33 # http://mail.python.org/pipermail/python-list/2007-May/442737.html 

34 

35 def __init__(self, file_or_name: Union[str, StringIO], mode: str="w", channel: str='stdout'): 

36 """Construct a new Tee object. 

37 

38 Parameters 

39 ---------- 

40 file_or_name : filename or open filehandle (writable) 

41 File that will be duplicated 

42 mode : optional, valid mode for open(). 

43 If a filename was give, open with this mode. 

44 channel : str, one of ['stdout', 'stderr'] 

45 """ 

46 if channel not in ['stdout', 'stderr']: 

47 raise ValueError('Invalid channel spec %s' % channel) 

48 

49 if hasattr(file_or_name, 'write') and hasattr(file_or_name, 'seek'): 

50 self.file = file_or_name 

51 else: 

52 encoding = None if "b" in mode else "utf-8" 

53 self.file = open(file_or_name, mode, encoding=encoding) 

54 self.channel = channel 

55 self.ostream = getattr(sys, channel) 

56 setattr(sys, channel, self) 

57 self._closed = False 

58 

59 def close(self): 

60 """Close the file and restore the channel.""" 

61 self.flush() 

62 setattr(sys, self.channel, self.ostream) 

63 self.file.close() 

64 self._closed = True 

65 

66 def write(self, data): 

67 """Write data to both channels.""" 

68 self.file.write(data) 

69 self.ostream.write(data) 

70 self.ostream.flush() 

71 

72 def flush(self): 

73 """Flush both channels.""" 

74 self.file.flush() 

75 self.ostream.flush() 

76 

77 def __del__(self): 

78 if not self._closed: 

79 self.close() 

80 

81 def isatty(self): 

82 return False 

83 

84def ask_yes_no(prompt, default=None, interrupt=None): 

85 """Asks a question and returns a boolean (y/n) answer. 

86 

87 If default is given (one of 'y','n'), it is used if the user input is 

88 empty. If interrupt is given (one of 'y','n'), it is used if the user 

89 presses Ctrl-C. Otherwise the question is repeated until an answer is 

90 given. 

91 

92 An EOF is treated as the default answer. If there is no default, an 

93 exception is raised to prevent infinite loops. 

94 

95 Valid answers are: y/yes/n/no (match is not case sensitive).""" 

96 

97 answers = {'y':True,'n':False,'yes':True,'no':False} 

98 ans = None 

99 while ans not in answers.keys(): 

100 try: 

101 ans = input(prompt+' ').lower() 

102 if not ans: # response was an empty string 

103 ans = default 

104 except KeyboardInterrupt: 

105 if interrupt: 

106 ans = interrupt 

107 print("\r") 

108 except EOFError: 

109 if default in answers.keys(): 

110 ans = default 

111 print() 

112 else: 

113 raise 

114 

115 return answers[ans] 

116 

117 

118def temp_pyfile(src: str, ext: str='.py') -> str: 

119 """Make a temporary python file, return filename and filehandle. 

120 

121 Parameters 

122 ---------- 

123 src : string or list of strings (no need for ending newlines if list) 

124 Source code to be written to the file. 

125 ext : optional, string 

126 Extension for the generated file. 

127 

128 Returns 

129 ------- 

130 (filename, open filehandle) 

131 It is the caller's responsibility to close the open file and unlink it. 

132 """ 

133 fname = tempfile.mkstemp(ext)[1] 

134 with open(Path(fname), "w", encoding="utf-8") as f: 

135 f.write(src) 

136 f.flush() 

137 return fname