Coverage for typer / _click / _compat.py: 100%

1import codecs 7 ctx1cdabefg

2import io 7 ctx1cdabefg

3import os 7 ctx1cdabefg

4import re 7 ctx1cdabefg

5import sys 7 ctx1cdabefg

6from collections.abc import Callable, Mapping, MutableMapping 7 ctx1cdabefg

7from types import TracebackType 7 ctx1cdabefg

8from typing import ( 7 ctx1cdabefg

9 IO, 

10 Any, 

11 BinaryIO, 

12 TextIO, 

13 cast, 

14) 

15from weakref import WeakKeyDictionary 7 ctx1cdabefg

16 

17CYGWIN = sys.platform.startswith("cygwin") 7 ctx1cdabefg

18WIN = sys.platform.startswith("win") 7 ctx1cdabefg

19auto_wrap_for_ansi: Callable[[TextIO], TextIO] | None = None 7 ctx1cdabefg

20_ansi_re = re.compile(r"\033\[[;?0-9]*[a-zA-Z]") 7 ctx1cdabefg

21 

22 

23def _make_text_stream( 7 ctx1cdabefg

24 stream: BinaryIO, 

25 encoding: str | None, 

26 errors: str, 

27) -> TextIO: 

28 if encoding is None: 7 ctx1cdabefg

29 encoding = get_best_encoding(stream) 7 ctx1cdabefg

30 return _NonClosingTextIOWrapper( 7 ctx1cdabefg

31 stream, 

32 encoding, 

33 errors, 

34 line_buffering=True, 

35 ) 

36 

37 

38def is_ascii_encoding(encoding: str) -> bool: 7 ctx1cdabefg

39 """Checks if a given encoding is ascii.""" 

40 try: 7 ctx1cdabefg

41 return codecs.lookup(encoding).name == "ascii" 7 ctx1cdabefg

42 except LookupError: 7 ctx1cdabefg

43 return False 7 ctx1cdabefg

44 

45 

46def get_best_encoding(stream: IO[Any]) -> str: 7 ctx1cdabefg

47 """Returns the default stream encoding if not found.""" 

48 rv = getattr(stream, "encoding", None) or sys.getdefaultencoding() 7 ctx1cdabefg

49 if is_ascii_encoding(rv): 7 ctx1cdabefg

50 return "utf-8" 7 ctx1cdabefg

51 return rv 7 ctx1cdabefg

52 

53 

54class _NonClosingTextIOWrapper(io.TextIOWrapper): 7 ctx1cdabefg

55 def __init__( 7 ctx1cdabefg

56 self, 

57 stream: BinaryIO, 

58 encoding: str | None, 

59 errors: str | None, 

60 **extra: Any, 

61 ) -> None: 

62 self._stream = stream = cast(BinaryIO, _FixupStream(stream)) 7 ctx1cdabefg

63 super().__init__(stream, encoding, errors, **extra) 7 ctx1cdabefg

64 

65 def __del__(self) -> None: 7 ctx1cdabefg

66 try: 7 ctx1cdabefg

67 self.detach() 7 ctx1cdabefg

68 except Exception: # pragma: no cover 7 ctx1cdabefg

69 pass 7 ctx1cdabefg

70 

71 def isatty(self) -> bool: 7 ctx1cdabefg

72 # https://bitbucket.org/pypy/pypy/issue/1803 

73 return self._stream.isatty() 7 ctx1cdabefg

74 

75 

76class _FixupStream: 7 ctx1cdabefg

77 """The new io interface needs more from streams than streams 

78 traditionally implement. As such, this fix-up code is necessary in 

79 some circumstances. 

80 """ 

81 

82 def __init__( 7 ctx1cdabefg

83 self, 

84 stream: BinaryIO, 

85 ): 

86 self._stream = stream 7 ctx1cdabefg

87 

88 def __getattr__(self, name: str) -> Any: 7 ctx1cdabefg

89 return getattr(self._stream, name) 7 ctx1cdabefg

90 

91 def read1(self, size: int) -> bytes: 7 ctx1cdabefg

92 f = getattr(self._stream, "read1", None) 7 ctx1cdabefg

93 

94 if f is not None: 7 ctx1cdabefg

95 return cast(bytes, f(size)) 2 ctx1ab

96 

97 return self._stream.read(size) 7 ctx1cdabefg

98 

99 def readable(self) -> bool: 7 ctx1cdabefg

100 return True 7 ctx1cdabefg

101 

102 def writable(self) -> bool: 7 ctx1cdabefg

103 return True 7 ctx1cdabefg

104 

105 def seekable(self) -> bool: 7 ctx1cdabefg

106 x = getattr(self._stream, "seekable", None) 7 ctx1cdabefg

107 if x is not None: 7 ctx1cdabefg

108 return cast(bool, x()) 7 ctx1cdabefg

109 return False 7 ctx1cdabefg

110 

111 

112def _is_binary_reader(stream: IO[Any], default: bool = False) -> bool: 7 ctx1cdabefg

113 try: 7 ctx1cdabefg

114 return isinstance(stream.read(0), bytes) 7 ctx1cdabefg

115 except Exception: # pragma: no cover 7 ctx1cdabefg

116 return default 7 ctx1cdabefg

117 # This happens in some cases where the stream was already 

118 # closed. In this case, we assume the default. 

119 

120 

121def _is_binary_writer(stream: IO[Any], default: bool = False) -> bool: 7 ctx1cdabefg

122 try: 7 ctx1cdabefg

123 stream.write(b"") 7 ctx1cdabefg

124 except Exception: # pragma: no cover 7 ctx1cdabefg

125 try: 7 ctx1cdabefg

126 stream.write("") 7 ctx1cdabefg

127 return False 7 ctx1cdabefg

128 except Exception: 7 ctx1cdabefg

129 pass 7 ctx1cdabefg

130 return default 7 ctx1cdabefg

131 return True 7 ctx1cdabefg

132 

133 

134def _find_binary_reader(stream: IO[Any]) -> BinaryIO | None: 7 ctx1cdabefg

135 # We need to figure out if the given stream is already binary. 

136 # This can happen because the official docs recommend detaching 

137 # the streams to get binary streams. Some code might do this, so 

138 # we need to deal with this case explicitly. 

139 if _is_binary_reader(stream, False): 7 ctx1cdabefg

140 return cast(BinaryIO, stream) 7 ctx1cdabefg

141 

142 buf = getattr(stream, "buffer", None) 7 ctx1cdabefg

143 

144 # Same situation here; this time we assume that the buffer is 

145 # actually binary in case it's closed. 

146 if buf is not None and _is_binary_reader(buf, True): 7 ctx1cdabefg

147 return cast(BinaryIO, buf) 7 ctx1cdabefg

148 

149 return None 7 ctx1cdabefg

150 

151 

152def _find_binary_writer(stream: IO[Any]) -> BinaryIO | None: 7 ctx1cdabefg

153 # We need to figure out if the given stream is already binary. 

154 # This can happen because the official docs recommend detaching 

155 # the streams to get binary streams. Some code might do this, so 

156 # we need to deal with this case explicitly. 

157 if _is_binary_writer(stream, False): 7 ctx1cdabefg

158 return cast(BinaryIO, stream) 7 ctx1cdabefg

159 

160 buf = getattr(stream, "buffer", None) 7 ctx1cdabefg

161 

162 # Same situation here; this time we assume that the buffer is 

163 # actually binary in case it's closed. 

164 if buf is not None and _is_binary_writer(buf, True): 7 ctx1cdabefg

165 return cast(BinaryIO, buf) 7 ctx1cdabefg

166 

167 return None 7 ctx1cdabefg

168 

169 

170def _stream_is_misconfigured(stream: TextIO) -> bool: 7 ctx1cdabefg

171 """A stream is misconfigured if its encoding is ASCII.""" 

172 # If the stream does not have an encoding set, we assume it's set 

173 # to ASCII. This appears to happen in certain unittest 

174 # environments. It's not quite clear what the correct behavior is 

175 # but this at least will force Click to recover somehow. 

176 return is_ascii_encoding(getattr(stream, "encoding", None) or "ascii") 7 ctx1cdabefg

177 

178 

179def _is_compat_stream_attr(stream: TextIO, attr: str, value: str | None) -> bool: 7 ctx1cdabefg

180 """A stream attribute is compatible if it is equal to the 

181 desired value or the desired value is unset and the attribute 

182 has a value. 

183 """ 

184 stream_value = getattr(stream, attr, None) 7 ctx1cdabefg

185 return stream_value == value or (value is None and stream_value is not None) 7 ctx1cdabefg

186 

187 

188def _is_compatible_text_stream( 7 ctx1cdabefg

189 stream: TextIO, encoding: str | None, errors: str | None 

190) -> bool: 

191 """Check if a stream's encoding and errors attributes are 

192 compatible with the desired values. 

193 """ 

194 return _is_compat_stream_attr( 7 ctx1cdabefg

195 stream, "encoding", encoding 

196 ) and _is_compat_stream_attr(stream, "errors", errors) 

197 

198 

199def _force_correct_text_stream( 7 ctx1cdabefg

200 text_stream: IO[Any], 

201 encoding: str | None, 

202 errors: str | None, 

203 is_binary: Callable[[IO[Any], bool], bool], 

204 find_binary: Callable[[IO[Any]], BinaryIO | None], 

205) -> TextIO: 

206 if is_binary(text_stream, False): 7 ctx1cdabefg

207 binary_reader = cast(BinaryIO, text_stream) 7 ctx1cdabefg

208 else: 

209 text_stream = cast(TextIO, text_stream) 7 ctx1cdabefg

210 # If the stream looks compatible, and won't default to a 

211 # misconfigured ascii encoding, return it as-is. 

212 if _is_compatible_text_stream(text_stream, encoding, errors) and not ( 7 ctx1cdabefg

213 encoding is None and _stream_is_misconfigured(text_stream) 

214 ): 

215 return text_stream 7 ctx1cdabefg

216 

217 # Otherwise, get the underlying binary reader. 

218 possible_binary_reader = find_binary(text_stream) 7 ctx1cdabefg

219 

220 # If that's not possible, silently use the original reader 

221 # and get mojibake instead of exceptions. 

222 if possible_binary_reader is None: 7 ctx1cdabefg

223 return text_stream 7 ctx1cdabefg

224 

225 binary_reader = possible_binary_reader 7 ctx1cdabefg

226 

227 # Default errors to replace instead of strict in order to get 

228 # something that works. 

229 if errors is None: 7 ctx1cdabefg

230 errors = "replace" 7 ctx1cdabefg

231 

232 # Wrap the binary stream in a text stream with the correct 

233 # encoding parameters. 

234 return _make_text_stream( 7 ctx1cdabefg

235 binary_reader, 

236 encoding, 

237 errors, 

238 ) 

239 

240 

241def _force_correct_text_reader( 7 ctx1cdabefg

242 text_reader: IO[Any], 

243 encoding: str | None, 

244 errors: str | None, 

245) -> TextIO: 

246 return _force_correct_text_stream( 7 ctx1cdabefg

247 text_reader, 

248 encoding, 

249 errors, 

250 _is_binary_reader, 

251 _find_binary_reader, 

252 ) 

253 

254 

255def _force_correct_text_writer( 7 ctx1cdabefg

256 text_writer: IO[Any], 

257 encoding: str | None, 

258 errors: str | None, 

259) -> TextIO: 

260 return _force_correct_text_stream( 7 ctx1cdabefg

261 text_writer, 

262 encoding, 

263 errors, 

264 _is_binary_writer, 

265 _find_binary_writer, 

266 ) 

267 

268 

269def get_binary_stdin() -> BinaryIO: 7 ctx1cdabefg

270 reader = _find_binary_reader(sys.stdin) 7 ctx1cdabefg

271 if reader is None: # pragma: no cover 7 ctx1cdabefg

272 raise RuntimeError("Was not able to determine binary stream for sys.stdin.") 7 ctx1cdabefg

273 return reader 7 ctx1cdabefg

274 

275 

276def get_binary_stdout() -> BinaryIO: 7 ctx1cdabefg

277 writer = _find_binary_writer(sys.stdout) 7 ctx1cdabefg

278 if writer is None: # pragma: no cover 7 ctx1cdabefg

279 raise RuntimeError("Was not able to determine binary stream for sys.stdout.") 

280 return writer 7 ctx1cdabefg

281 

282 

283def get_binary_stderr() -> BinaryIO: 7 ctx1cdabefg

284 writer = _find_binary_writer(sys.stderr) 7 ctx1cdabefg

285 if writer is None: # pragma: no cover 7 ctx1cdabefg

286 raise RuntimeError("Was not able to determine binary stream for sys.stderr.") 

287 return writer 7 ctx1cdabefg

288 

289 

290def get_text_stdin(encoding: str | None = None, errors: str | None = None) -> TextIO: 7 ctx1cdabefg

291 rv = _get_windows_console_stream(sys.stdin, encoding, errors) 7 ctx1cdabefg

292 if rv is not None: 7 ctx1cdabefg

293 return rv 2 ctx1ab

294 return _force_correct_text_reader(sys.stdin, encoding, errors) 7 ctx1cdabefg

295 

296 

297def get_text_stdout(encoding: str | None = None, errors: str | None = None) -> TextIO: 7 ctx1cdabefg

298 rv = _get_windows_console_stream(sys.stdout, encoding, errors) 7 ctx1cdabefg

299 if rv is not None: 7 ctx1cdabefg

300 return rv 2 ctx1ab

301 return _force_correct_text_writer(sys.stdout, encoding, errors) 7 ctx1cdabefg

302 

303 

304def get_text_stderr(encoding: str | None = None, errors: str | None = None) -> TextIO: 7 ctx1cdabefg

305 rv = _get_windows_console_stream(sys.stderr, encoding, errors) 7 ctx1cdabefg

306 if rv is not None: 7 ctx1cdabefg

307 return rv 2 ctx1ab

308 return _force_correct_text_writer(sys.stderr, encoding, errors) 7 ctx1cdabefg

309 

310 

311def _wrap_io_open( 7 ctx1cdabefg

312 file: str | os.PathLike[str] | int, 

313 mode: str, 

314 encoding: str | None, 

315 errors: str | None, 

316) -> IO[Any]: 

317 """Handles not passing ``encoding`` and ``errors`` in binary mode.""" 

318 if "b" in mode: 7 ctx1cdabefg

319 return open(file, mode) 7 ctx1cdabefg

320 

321 return open(file, mode, encoding=encoding, errors=errors) 7 ctx1cdabefg

322 

323 

324def open_stream( 7 ctx1cdabefg

325 filename: str | os.PathLike[str], 

326 mode: str = "r", 

327 encoding: str | None = None, 

328 errors: str | None = "strict", 

329 atomic: bool = False, 

330) -> tuple[IO[Any], bool]: 

331 binary = "b" in mode 7 ctx1cdabefg

332 filename = os.fspath(filename) 7 ctx1cdabefg

333 

334 # Standard streams first, ignoring the atomic flag. 

335 if os.fsdecode(filename) == "-": 7 ctx1cdabefg

336 if any(m in mode for m in ["w", "a", "x"]): 7 ctx1cdabefg

337 if binary: 7 ctx1cdabefg

338 return get_binary_stdout(), False 7 ctx1cdabefg

339 return get_text_stdout(encoding=encoding, errors=errors), False 7 ctx1cdabefg

340 if binary: 7 ctx1cdabefg

341 return get_binary_stdin(), False 7 ctx1cdabefg

342 return get_text_stdin(encoding=encoding, errors=errors), False 7 ctx1cdabefg

343 

344 # Non-atomic writes directly go out through the regular open functions. 

345 if not atomic: 7 ctx1cdabefg

346 return _wrap_io_open(filename, mode, encoding, errors), True 7 ctx1cdabefg

347 

348 # Some usability stuff for atomic writes 

349 if "a" in mode: 7 ctx1cdabefg

350 raise ValueError( 7 ctx1cdabefg

351 "Appending to an existing file is not supported, because that" 

352 " would involve an expensive `copy`-operation to a temporary" 

353 " file. Open the file in normal `w`-mode and copy explicitly" 

354 " if that's what you're after." 

355 ) 

356 if "x" in mode: 7 ctx1cdabefg

357 raise ValueError("Use the `overwrite`-parameter instead.") 7 ctx1cdabefg

358 if "w" not in mode: 7 ctx1cdabefg

359 raise ValueError("Atomic writes only make sense with `w`-mode.") 7 ctx1cdabefg

360 

361 # Atomic writes are more complicated. They work by opening a file 

362 # as a proxy in the same folder and then using the fdopen 

363 # functionality to wrap it in a Python file. Then we wrap it in an 

364 # atomic file that moves the file over on close. 

365 import errno 7 ctx1cdabefg

366 import random 7 ctx1cdabefg

367 

368 try: 7 ctx1cdabefg

369 perm: int | None = os.stat(filename).st_mode 7 ctx1cdabefg

370 except OSError: # pragma: no cover 7 ctx1cdabefg

371 perm = None 7 ctx1cdabefg

372 

373 flags = os.O_RDWR | os.O_CREAT | os.O_EXCL 7 ctx1cdabefg

374 

375 if binary: 7 ctx1cdabefg

376 flags |= getattr(os, "O_BINARY", 0) 7 ctx1cdabefg

377 

378 while True: 7 ctx1cdabefg

379 tmp_filename = os.path.join( 7 ctx1cdabefg

380 os.path.dirname(filename), 

381 f".__atomic-write{random.randrange(1 << 32):08x}", 

382 ) 

383 try: 7 ctx1cdabefg

384 fd = os.open(tmp_filename, flags, 0o666 if perm is None else perm) 7 ctx1cdabefg

385 break 7 ctx1cdabefg

386 except OSError as e: # pragma: no cover 

387 if e.errno == errno.EEXIST or ( 

388 os.name == "nt" 

389 and e.errno == errno.EACCES 

390 and os.path.isdir(e.filename) 

391 and os.access(e.filename, os.W_OK) 

392 ): 

393 continue 

394 raise 

395 

396 if perm is not None: 7 ctx1cdabefg

397 os.chmod(tmp_filename, perm) # in case perm includes bits in umask 7 ctx1cdabefg

398 

399 f = _wrap_io_open(fd, mode, encoding, errors) 7 ctx1cdabefg

400 af = _AtomicFile(f, tmp_filename, os.path.realpath(filename)) 7 ctx1cdabefg

401 return cast(IO[Any], af), True 7 ctx1cdabefg

402 

403 

404class _AtomicFile: 7 ctx1cdabefg

405 def __init__(self, f: IO[Any], tmp_filename: str, real_filename: str) -> None: 7 ctx1cdabefg

406 self._f = f 7 ctx1cdabefg

407 self._tmp_filename = tmp_filename 7 ctx1cdabefg

408 self._real_filename = real_filename 7 ctx1cdabefg

409 self.closed = False 7 ctx1cdabefg

410 

411 @property 7 ctx1cdabefg

412 def name(self) -> str: 7 ctx1cdabefg

413 return self._real_filename 7 ctx1cdabefg

414 

415 def close(self, delete: bool = False) -> None: 7 ctx1cdabefg

416 if self.closed: 7 ctx1cdabefg

417 return # pragma: no cover 7 ctx1cdabefg

418 self._f.close() 7 ctx1cdabefg

419 os.replace(self._tmp_filename, self._real_filename) 7 ctx1cdabefg

420 self.closed = True 7 ctx1cdabefg

421 

422 def __getattr__(self, name: str) -> Any: 7 ctx1cdabefg

423 return getattr(self._f, name) 7 ctx1cdabefg

424 

425 def __enter__(self) -> "_AtomicFile": 7 ctx1cdabefg

426 return self 7 ctx1cdabefg

427 

428 def __exit__( 7 ctx1cdabefg

429 self, 

430 exc_type: type[BaseException] | None, 

431 exc_value: BaseException | None, 

432 tb: TracebackType | None, 

433 ) -> None: 

434 self.close(delete=exc_type is not None) 7 ctx1cdabefg

435 

436 def __repr__(self) -> str: 7 ctx1cdabefg

437 return repr(self._f) 7 ctx1cdabefg

438 

439 

440def strip_ansi(value: str) -> str: 7 ctx1cdabefg

441 return _ansi_re.sub("", value) 7 ctx1cdabefg

442 

443 

444def _is_jupyter_kernel_output(stream: IO[Any]) -> bool: 7 ctx1cdabefg

445 while isinstance(stream, (_FixupStream, _NonClosingTextIOWrapper)): 7 ctx1cdabefg

446 stream = stream._stream 7 ctx1cdabefg

447 

448 return stream.__class__.__module__.startswith("ipykernel.") 7 ctx1cdabefg

449 

450 

451def should_strip_ansi(stream: IO[Any] | None = None, color: bool | None = None) -> bool: 7 ctx1cdabefg

452 if color is None: 7 ctx1cdabefg

453 if stream is None: 7 ctx1cdabefg

454 stream = sys.stdin 7 ctx1cdabefg

455 return not isatty(stream) and not _is_jupyter_kernel_output(stream) 7 ctx1cdabefg

456 return not color 7 ctx1cdabefg

457 

458 

459# On Windows, wrap the output streams with colorama to support ANSI 

460# color codes. 

461# NOTE: double check is needed so mypy does not analyze this on Linux 

462if sys.platform.startswith("win") and WIN: 7 ctx1cdabefg

463 from ._winconsole import _get_windows_console_stream 2 ctx1ab

464 

465 def _get_argv_encoding() -> str: 2 ctx1ab

466 import locale 2 ctx1ab

467 

468 return locale.getpreferredencoding() 2 ctx1ab

469 

470 _ansi_stream_wrappers: MutableMapping[TextIO, TextIO] = WeakKeyDictionary() 2 ctx1ab

471 

472 def auto_wrap_for_ansi(stream: TextIO, color: bool | None = None) -> TextIO: 2 ctx1ab

473 """Support ANSI color and style codes on Windows by wrapping a 

474 stream with colorama. 

475 """ 

476 try: 2 ctx1ab

477 cached = _ansi_stream_wrappers.get(stream) 2 ctx1ab

478 except Exception: # pragma: no cover 

479 cached = None 

480 

481 if cached is not None: 2 ctx1ab

482 return cached 2 ctx1ab

483 

484 import colorama 2 ctx1ab

485 

486 strip = should_strip_ansi(stream, color) 2 ctx1ab

487 ansi_wrapper = colorama.AnsiToWin32(stream, strip=strip) 2 ctx1ab

488 rv = cast(TextIO, ansi_wrapper.stream) 2 ctx1ab

489 _write = rv.write 2 ctx1ab

490 

491 def _safe_write(s: str) -> int: 2 ctx1ab

492 try: 2 ctx1ab

493 return _write(s) 2 ctx1ab

494 except BaseException: # pragma: no cover 

495 ansi_wrapper.reset_all() 

496 raise 

497 

498 rv.write = _safe_write # type: ignore[method-assign] 2 ctx1ab

499 

500 try: 2 ctx1ab

501 _ansi_stream_wrappers[stream] = rv 2 ctx1ab

502 except Exception: # pragma: no cover 

503 pass 

504 

505 return rv 2 ctx1ab

506 

507else: 

508 

509 def _get_argv_encoding() -> str: 5 ctx1cdefg

510 return getattr(sys.stdin, "encoding", None) or sys.getfilesystemencoding() 2 ctx1cd

511 

512 def _get_windows_console_stream( 5 ctx1cdefg

513 f: TextIO, encoding: str | None, errors: str | None 

514 ) -> TextIO | None: 

515 return None 5 ctx1cdefg

516 

517 

518def term_len(x: str) -> int: 7 ctx1cdabefg

519 return len(strip_ansi(x)) 7 ctx1cdabefg

520 

521 

522def isatty(stream: IO[Any]) -> bool: 7 ctx1cdabefg

523 try: 7 ctx1cdabefg

524 return stream.isatty() 7 ctx1cdabefg

525 except Exception: # pragma: no cover 

526 return False 

527 

528 

529def _make_cached_stream_func( 7 ctx1cdabefg

530 src_func: Callable[[], TextIO], 

531 wrapper_func: Callable[[], TextIO], 

532) -> Callable[[], TextIO]: 

533 cache: MutableMapping[TextIO, TextIO] = WeakKeyDictionary() 7 ctx1cdabefg

534 

535 def func() -> TextIO: 7 ctx1cdabefg

536 stream = src_func() 7 ctx1cdabefg

537 

538 try: 7 ctx1cdabefg

539 rv = cache.get(stream) 7 ctx1cdabefg

540 except Exception: # pragma: no cover 7 ctx1cdabefg

541 rv = None 7 ctx1cdabefg

542 if rv is not None: 7 ctx1cdabefg

543 return rv 7 ctx1cdabefg

544 rv = wrapper_func() 7 ctx1cdabefg

545 try: 7 ctx1cdabefg

546 cache[stream] = rv 7 ctx1cdabefg

547 except Exception: # pragma: no cover 7 ctx1cdabefg

548 pass 7 ctx1cdabefg

549 return rv 7 ctx1cdabefg

550 

551 return func 7 ctx1cdabefg

552 

553 

554_default_text_stdin = _make_cached_stream_func(lambda: sys.stdin, get_text_stdin) 7 ctx1cdabefg

555_default_text_stdout = _make_cached_stream_func(lambda: sys.stdout, get_text_stdout) 7 ctx1cdabefg

556_default_text_stderr = _make_cached_stream_func(lambda: sys.stderr, get_text_stderr) 7 ctx1cdabefg

557 

558 

559binary_streams: Mapping[str, Callable[[], BinaryIO]] = { 7 ctx1cdabefg

560 "stdin": get_binary_stdin, 

561 "stdout": get_binary_stdout, 

562 "stderr": get_binary_stderr, 

563} 

564 

565text_streams: Mapping[str, Callable[[str | None, str | None], TextIO]] = { 7 ctx1cdabefg

566 "stdin": get_text_stdin, 

567 "stdout": get_text_stdout, 

568 "stderr": get_text_stderr, 

569}