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

1""" 

2IO related utilities. 

3""" 

4 

5# Copyright (c) IPython Development Team. 

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

7 

8 

9 

10import sys 

11from pathlib import Path 

12 

13from .capture import CapturedIO, capture_output 

14from io import StringIO 

15 

16 

17class Tee: 

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

19 

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

21 

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

23 it for duplication. 

24 """ 

25 # Inspired by: 

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

27 

28 def __init__(self, file_or_name: str | StringIO, mode: str="w", channel: str='stdout'): 

29 """Construct a new Tee object. 

30 

31 Parameters 

32 ---------- 

33 file_or_name : filename or open filehandle (writable) 

34 File that will be duplicated 

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

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

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

38 """ 

39 self._closed = True 

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

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

42 

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

44 self.file = file_or_name 

45 else: 

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

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

48 self.channel = channel 

49 self.ostream = getattr(sys, channel) 

50 setattr(sys, channel, self) 

51 self._closed = False # fully initialized, mark as open 

52 

53 def close(self): 

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

55 self.flush() 

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

57 self.file.close() 

58 self._closed = True 

59 

60 def write(self, data): 

61 """Write data to both channels.""" 

62 self.file.write(data) 

63 self.ostream.write(data) 

64 self.ostream.flush() 

65 

66 def flush(self): 

67 """Flush both channels.""" 

68 self.file.flush() 

69 self.ostream.flush() 

70 

71 def __del__(self): 

72 if not self._closed: 

73 self.close() 

74 

75 def isatty(self): 

76 return False 

77 

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

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

80 

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

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

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

84 given. 

85 

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

87 exception is raised to prevent infinite loops. 

88 

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

90 

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

92 ans = None 

93 while ans not in answers.keys(): 

94 try: 

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

96 if not ans: # response was an empty string 

97 ans = default 

98 except KeyboardInterrupt: 

99 if interrupt: 

100 ans = interrupt 

101 print("\r") 

102 except EOFError: 

103 if default in answers.keys(): 

104 ans = default 

105 print() 

106 else: 

107 raise 

108 

109 return answers[ans] 

110 

111 

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

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

114 

115 Parameters 

116 ---------- 

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

118 Source code to be written to the file. 

119 ext : optional, string 

120 Extension for the generated file. 

121 

122 Returns 

123 ------- 

124 (filename, open filehandle) 

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

126 """ 

127 import tempfile 

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

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

130 f.write(src) 

131 f.flush() 

132 return fname