Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.11/site-packages/click/_compat.py: 24%
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 codecs
4import collections.abc as cabc
5import io
6import os
7import re
8import sys
9import typing as t
10from types import TracebackType
11from weakref import WeakKeyDictionary
13CYGWIN = sys.platform.startswith("cygwin")
14WIN = sys.platform.startswith("win")
15# One CSI escape sequence per the ECMA-48 grammar: parameter bytes (0x30-0x3F),
16# intermediate bytes (0x20-0x2F), then a final byte (0x40-0x7E). Broader than the
17# SGR codes Click emits, so foreign sequences (colon-delimited true-color, mouse
18# reporting) are stripped too.
19_ansi_re = re.compile(r"\033\[[0-?]*[ -/]*[@-~]")
22def _make_text_stream(
23 stream: t.BinaryIO,
24 encoding: str | None,
25 errors: str | None,
26 force_readable: bool = False,
27 force_writable: bool = False,
28) -> t.TextIO:
29 if encoding is None:
30 encoding = get_best_encoding(stream)
31 if errors is None:
32 errors = "replace"
33 return _NonClosingTextIOWrapper(
34 stream,
35 encoding,
36 errors,
37 line_buffering=True,
38 force_readable=force_readable,
39 force_writable=force_writable,
40 )
43def is_ascii_encoding(encoding: str) -> bool:
44 """Checks if a given encoding is ascii."""
45 try:
46 return codecs.lookup(encoding).name == "ascii"
47 except LookupError:
48 return False
51def get_best_encoding(stream: t.IO[t.Any]) -> str:
52 """Returns the default stream encoding if not found."""
53 rv = getattr(stream, "encoding", None) or sys.getdefaultencoding()
54 if is_ascii_encoding(rv):
55 return "utf-8"
56 return rv
59class _NonClosingTextIOWrapper(io.TextIOWrapper):
60 def __init__(
61 self,
62 stream: t.BinaryIO,
63 encoding: str | None,
64 errors: str | None,
65 force_readable: bool = False,
66 force_writable: bool = False,
67 **extra: t.Any,
68 ) -> None:
69 self._stream = stream = t.cast(
70 t.BinaryIO, _FixupStream(stream, force_readable, force_writable)
71 )
72 super().__init__(stream, encoding, errors, **extra)
74 def __del__(self) -> None:
75 try:
76 self.detach()
77 except Exception:
78 pass
80 def isatty(self) -> bool:
81 # https://bitbucket.org/pypy/pypy/issue/1803
82 return self._stream.isatty()
85class _FixupStream:
86 """The new io interface needs more from streams than streams
87 traditionally implement. As such, this fix-up code is necessary in
88 some circumstances.
90 The forcing of readable and writable flags are there because some tools
91 put badly patched objects on sys (one such offender are certain version
92 of jupyter notebook).
93 """
95 def __init__(
96 self,
97 stream: t.BinaryIO,
98 force_readable: bool = False,
99 force_writable: bool = False,
100 ):
101 self._stream = stream
102 self._force_readable = force_readable
103 self._force_writable = force_writable
105 def __getattr__(self, name: str) -> t.Any:
106 return getattr(self._stream, name)
108 def read1(self, size: int) -> bytes:
109 f = getattr(self._stream, "read1", None)
111 if f is not None:
112 return t.cast(bytes, f(size))
114 return self._stream.read(size)
116 def readable(self) -> bool:
117 if self._force_readable:
118 return True
119 x = getattr(self._stream, "readable", None)
120 if x is not None:
121 return t.cast(bool, x())
122 try:
123 self._stream.read(0)
124 except Exception:
125 return False
126 return True
128 def writable(self) -> bool:
129 if self._force_writable:
130 return True
131 x = getattr(self._stream, "writable", None)
132 if x is not None:
133 return t.cast(bool, x())
134 try:
135 self._stream.write(b"")
136 except Exception:
137 try:
138 self._stream.write(b"")
139 except Exception:
140 return False
141 return True
143 def seekable(self) -> bool:
144 x = getattr(self._stream, "seekable", None)
145 if x is not None:
146 return t.cast(bool, x())
147 try:
148 self._stream.seek(self._stream.tell())
149 except Exception:
150 return False
151 return True
154def _is_binary_reader(stream: t.IO[t.Any], default: bool = False) -> bool:
155 try:
156 return isinstance(stream.read(0), bytes)
157 except Exception:
158 return default
159 # This happens in some cases where the stream was already
160 # closed. In this case, we assume the default.
163def _is_binary_writer(stream: t.IO[t.Any], default: bool = False) -> bool:
164 try:
165 stream.write(b"")
166 except Exception:
167 try:
168 stream.write("")
169 return False
170 except Exception:
171 pass
172 return default
173 return True
176def _find_binary_reader(stream: t.IO[t.Any]) -> t.BinaryIO | None:
177 # We need to figure out if the given stream is already binary.
178 # This can happen because the official docs recommend detaching
179 # the streams to get binary streams. Some code might do this, so
180 # we need to deal with this case explicitly.
181 if _is_binary_reader(stream, False):
182 return t.cast(t.BinaryIO, stream)
184 buf = getattr(stream, "buffer", None)
186 # Same situation here; this time we assume that the buffer is
187 # actually binary in case it's closed.
188 if buf is not None and _is_binary_reader(buf, True):
189 return t.cast(t.BinaryIO, buf)
191 return None
194def _find_binary_writer(stream: t.IO[t.Any]) -> t.BinaryIO | None:
195 # We need to figure out if the given stream is already binary.
196 # This can happen because the official docs recommend detaching
197 # the streams to get binary streams. Some code might do this, so
198 # we need to deal with this case explicitly.
199 if _is_binary_writer(stream, False):
200 return t.cast(t.BinaryIO, stream)
202 buf = getattr(stream, "buffer", None)
204 # Same situation here; this time we assume that the buffer is
205 # actually binary in case it's closed.
206 if buf is not None and _is_binary_writer(buf, True):
207 return t.cast(t.BinaryIO, buf)
209 return None
212def _stream_is_misconfigured(stream: t.TextIO) -> bool:
213 """A stream is misconfigured if its encoding is ASCII."""
214 # If the stream does not have an encoding set, we assume it's set
215 # to ASCII. This appears to happen in certain unittest
216 # environments. It's not quite clear what the correct behavior is
217 # but this at least will force Click to recover somehow.
218 return is_ascii_encoding(getattr(stream, "encoding", None) or "ascii")
221def _is_compat_stream_attr(stream: t.TextIO, attr: str, value: str | None) -> bool:
222 """A stream attribute is compatible if it is equal to the
223 desired value or the desired value is unset and the attribute
224 has a value.
225 """
226 stream_value = getattr(stream, attr, None)
227 return stream_value == value or (value is None and stream_value is not None)
230def _is_compatible_text_stream(
231 stream: t.TextIO, encoding: str | None, errors: str | None
232) -> bool:
233 """Check if a stream's encoding and errors attributes are
234 compatible with the desired values.
235 """
236 return _is_compat_stream_attr(
237 stream, "encoding", encoding
238 ) and _is_compat_stream_attr(stream, "errors", errors)
241def _force_correct_text_stream(
242 text_stream: t.IO[t.Any],
243 encoding: str | None,
244 errors: str | None,
245 is_binary: t.Callable[[t.IO[t.Any], bool], bool],
246 find_binary: t.Callable[[t.IO[t.Any]], t.BinaryIO | None],
247 force_readable: bool = False,
248 force_writable: bool = False,
249) -> t.TextIO:
250 if is_binary(text_stream, False):
251 binary_reader = t.cast(t.BinaryIO, text_stream)
252 else:
253 text_stream = t.cast(t.TextIO, text_stream)
254 # If the stream looks compatible, and won't default to a
255 # misconfigured ascii encoding, return it as-is.
256 if _is_compatible_text_stream(text_stream, encoding, errors) and not (
257 encoding is None and _stream_is_misconfigured(text_stream)
258 ):
259 return text_stream
261 # Otherwise, get the underlying binary reader.
262 possible_binary_reader = find_binary(text_stream)
264 # If that's not possible, silently use the original reader
265 # and get mojibake instead of exceptions.
266 if possible_binary_reader is None:
267 return text_stream
269 binary_reader = possible_binary_reader
271 # Default errors to replace instead of strict in order to get
272 # something that works.
273 if errors is None:
274 errors = "replace"
276 # Wrap the binary stream in a text stream with the correct
277 # encoding parameters.
278 return _make_text_stream(
279 binary_reader,
280 encoding,
281 errors,
282 force_readable=force_readable,
283 force_writable=force_writable,
284 )
287def _force_correct_text_reader(
288 text_reader: t.IO[t.Any],
289 encoding: str | None,
290 errors: str | None,
291 force_readable: bool = False,
292) -> t.TextIO:
293 return _force_correct_text_stream(
294 text_reader,
295 encoding,
296 errors,
297 _is_binary_reader,
298 _find_binary_reader,
299 force_readable=force_readable,
300 )
303def _force_correct_text_writer(
304 text_writer: t.IO[t.Any],
305 encoding: str | None,
306 errors: str | None,
307 force_writable: bool = False,
308) -> t.TextIO:
309 return _force_correct_text_stream(
310 text_writer,
311 encoding,
312 errors,
313 _is_binary_writer,
314 _find_binary_writer,
315 force_writable=force_writable,
316 )
319def get_binary_stdin() -> t.BinaryIO:
320 reader = _find_binary_reader(sys.stdin)
321 if reader is None:
322 raise RuntimeError("Was not able to determine binary stream for sys.stdin.")
323 return reader
326def get_binary_stdout() -> t.BinaryIO:
327 writer = _find_binary_writer(sys.stdout)
328 if writer is None:
329 raise RuntimeError("Was not able to determine binary stream for sys.stdout.")
330 return writer
333def get_binary_stderr() -> t.BinaryIO:
334 writer = _find_binary_writer(sys.stderr)
335 if writer is None:
336 raise RuntimeError("Was not able to determine binary stream for sys.stderr.")
337 return writer
340def get_text_stdin(encoding: str | None = None, errors: str | None = None) -> t.TextIO:
341 rv = _get_windows_console_stream(sys.stdin, encoding, errors)
342 if rv is not None:
343 return rv
344 return _force_correct_text_reader(sys.stdin, encoding, errors, force_readable=True)
347def get_text_stdout(encoding: str | None = None, errors: str | None = None) -> t.TextIO:
348 rv = _get_windows_console_stream(sys.stdout, encoding, errors)
349 if rv is not None:
350 return rv
351 return _force_correct_text_writer(sys.stdout, encoding, errors, force_writable=True)
354def get_text_stderr(encoding: str | None = None, errors: str | None = None) -> t.TextIO:
355 rv = _get_windows_console_stream(sys.stderr, encoding, errors)
356 if rv is not None:
357 return rv
358 return _force_correct_text_writer(sys.stderr, encoding, errors, force_writable=True)
361def _wrap_io_open(
362 file: str | os.PathLike[str] | int,
363 mode: str,
364 encoding: str | None,
365 errors: str | None,
366) -> t.IO[t.Any]:
367 """Handles not passing ``encoding`` and ``errors`` in binary mode."""
368 if "b" in mode:
369 return open(file, mode)
371 return open(file, mode, encoding=encoding, errors=errors)
374def open_stream(
375 filename: str | os.PathLike[str],
376 mode: str = "r",
377 encoding: str | None = None,
378 errors: str | None = "strict",
379 atomic: bool = False,
380) -> tuple[t.IO[t.Any], bool]:
381 binary = "b" in mode
382 filename = os.fspath(filename)
384 # Standard streams first. These are simple because they ignore the
385 # atomic flag. Use fsdecode to handle Path("-").
386 if os.fsdecode(filename) == "-":
387 if any(m in mode for m in ["w", "a", "x"]):
388 if binary:
389 return get_binary_stdout(), False
390 return get_text_stdout(encoding=encoding, errors=errors), False
391 if binary:
392 return get_binary_stdin(), False
393 return get_text_stdin(encoding=encoding, errors=errors), False
395 # Non-atomic writes directly go out through the regular open functions.
396 if not atomic:
397 return _wrap_io_open(filename, mode, encoding, errors), True
399 # Some usability stuff for atomic writes
400 if "a" in mode:
401 raise ValueError(
402 "Appending to an existing file is not supported, because that"
403 " would involve an expensive `copy`-operation to a temporary"
404 " file. Open the file in normal `w`-mode and copy explicitly"
405 " if that's what you're after."
406 )
407 if "x" in mode:
408 raise ValueError("Use the `overwrite`-parameter instead.")
409 if "w" not in mode:
410 raise ValueError("Atomic writes only make sense with `w`-mode.")
412 # Atomic writes are more complicated. They work by opening a file
413 # as a proxy in the same folder and then using the fdopen
414 # functionality to wrap it in a Python file. Then we wrap it in an
415 # atomic file that moves the file over on close.
416 import errno
417 import random
419 try:
420 perm: int | None = os.stat(filename).st_mode
421 except OSError:
422 perm = None
424 flags = os.O_RDWR | os.O_CREAT | os.O_EXCL
426 if binary:
427 flags |= getattr(os, "O_BINARY", 0)
429 while True:
430 tmp_filename = os.path.join(
431 os.path.dirname(filename),
432 f".__atomic-write{random.randrange(1 << 32):08x}",
433 )
434 try:
435 fd = os.open(tmp_filename, flags, 0o666 if perm is None else perm)
436 break
437 except OSError as e:
438 if e.errno == errno.EEXIST or (
439 os.name == "nt"
440 and e.errno == errno.EACCES
441 and os.path.isdir(e.filename)
442 and os.access(e.filename, os.W_OK)
443 ):
444 continue
445 raise
447 if perm is not None:
448 os.chmod(tmp_filename, perm) # in case perm includes bits in umask
450 f = _wrap_io_open(fd, mode, encoding, errors)
451 af = _AtomicFile(f, tmp_filename, os.path.realpath(filename))
452 return t.cast(t.IO[t.Any], af), True
455class _AtomicFile:
456 def __init__(self, f: t.IO[t.Any], tmp_filename: str, real_filename: str) -> None:
457 self._f = f
458 self._tmp_filename = tmp_filename
459 self._real_filename = real_filename
460 self.closed = False
462 @property
463 def name(self) -> str:
464 return self._real_filename
466 def close(self, delete: bool = False) -> None:
467 if self.closed:
468 return
469 self._f.close()
470 os.replace(self._tmp_filename, self._real_filename)
471 self.closed = True
473 def __getattr__(self, name: str) -> t.Any:
474 return getattr(self._f, name)
476 def __enter__(self) -> _AtomicFile:
477 return self
479 def __exit__(
480 self,
481 exc_type: type[BaseException] | None,
482 exc_value: BaseException | None,
483 tb: TracebackType | None,
484 ) -> None:
485 self.close(delete=exc_type is not None)
487 def __repr__(self) -> str:
488 return repr(self._f)
491def strip_ansi(value: str) -> str:
492 return _ansi_re.sub("", value)
495def _is_jupyter_kernel_output(stream: t.IO[t.Any]) -> bool:
496 while isinstance(stream, (_FixupStream, _NonClosingTextIOWrapper)):
497 stream = stream._stream
499 return stream.__class__.__module__.startswith("ipykernel.")
502def should_strip_ansi(
503 stream: t.IO[t.Any] | None = None, color: bool | None = None
504) -> bool:
505 if color is None:
506 if stream is None:
507 stream = sys.stdin
508 elif hasattr(stream, "color"):
509 # ._termui_impl.MaybeStripAnsi handles stripping ansi itself,
510 # so we don't need to strip it here
511 return False
512 return not isatty(stream) and not _is_jupyter_kernel_output(stream)
513 return not color
516# Double check is needed so mypy does not analyze this on Linux.
517if sys.platform.startswith("win") and WIN:
518 from ._winconsole import _get_windows_console_stream
520 def _get_argv_encoding() -> str:
521 import locale
523 return locale.getpreferredencoding()
525else:
527 def _get_argv_encoding() -> str:
528 return getattr(sys.stdin, "encoding", None) or sys.getfilesystemencoding()
530 def _get_windows_console_stream(
531 f: t.TextIO, encoding: str | None, errors: str | None
532 ) -> t.TextIO | None:
533 return None
536def term_len(x: str) -> int:
537 return len(strip_ansi(x))
540def isatty(stream: t.IO[t.Any]) -> bool:
541 try:
542 return stream.isatty()
543 except Exception:
544 return False
547def _make_cached_stream_func(
548 src_func: t.Callable[[], t.TextIO | None],
549 wrapper_func: t.Callable[[], t.TextIO],
550) -> t.Callable[[], t.TextIO | None]:
551 cache: cabc.MutableMapping[t.TextIO, t.TextIO] = WeakKeyDictionary()
553 def func() -> t.TextIO | None:
554 stream = src_func()
556 if stream is None:
557 return None
559 try:
560 rv = cache.get(stream)
561 except Exception:
562 rv = None
563 if rv is not None:
564 return rv
565 rv = wrapper_func()
566 try:
567 cache[stream] = rv
568 except Exception:
569 pass
570 return rv
572 return func
575_default_text_stdin = _make_cached_stream_func(lambda: sys.stdin, get_text_stdin)
576_default_text_stdout = _make_cached_stream_func(lambda: sys.stdout, get_text_stdout)
577_default_text_stderr = _make_cached_stream_func(lambda: sys.stderr, get_text_stderr)
580binary_streams: cabc.Mapping[str, t.Callable[[], t.BinaryIO]] = {
581 "stdin": get_binary_stdin,
582 "stdout": get_binary_stdout,
583 "stderr": get_binary_stderr,
584}
586text_streams: cabc.Mapping[str, t.Callable[[str | None, str | None], t.TextIO]] = {
587 "stdin": get_text_stdin,
588 "stdout": get_text_stdout,
589 "stderr": get_text_stderr,
590}