File manager - Edit - /home/newsbmcs.com/public_html/static/img/logo/click.tar
Back
_unicodefun.py 0000644 00000006201 15030210742 0007405 0 ustar 00 import codecs import os from gettext import gettext as _ def _verify_python_env() -> None: """Ensures that the environment is good for Unicode.""" try: from locale import getpreferredencoding fs_enc = codecs.lookup(getpreferredencoding()).name except Exception: fs_enc = "ascii" if fs_enc != "ascii": return extra = [ _( "Click will abort further execution because Python was" " configured to use ASCII as encoding for the environment." " Consult https://click.palletsprojects.com/unicode-support/" " for mitigation steps." ) ] if os.name == "posix": import subprocess try: rv = subprocess.Popen( ["locale", "-a"], stdout=subprocess.PIPE, stderr=subprocess.PIPE, encoding="ascii", errors="replace", ).communicate()[0] except OSError: rv = "" good_locales = set() has_c_utf8 = False for line in rv.splitlines(): locale = line.strip() if locale.lower().endswith((".utf-8", ".utf8")): good_locales.add(locale) if locale.lower() in ("c.utf8", "c.utf-8"): has_c_utf8 = True if not good_locales: extra.append( _( "Additional information: on this system no suitable" " UTF-8 locales were discovered. This most likely" " requires resolving by reconfiguring the locale" " system." ) ) elif has_c_utf8: extra.append( _( "This system supports the C.UTF-8 locale which is" " recommended. You might be able to resolve your" " issue by exporting the following environment" " variables:" ) ) extra.append(" export LC_ALL=C.UTF-8\n export LANG=C.UTF-8") else: extra.append( _( "This system lists some UTF-8 supporting locales" " that you can pick from. The following suitable" " locales were discovered: {locales}" ).format(locales=", ".join(sorted(good_locales))) ) bad_locale = None for env_locale in os.environ.get("LC_ALL"), os.environ.get("LANG"): if env_locale and env_locale.lower().endswith((".utf-8", ".utf8")): bad_locale = env_locale if env_locale is not None: break if bad_locale is not None: extra.append( _( "Click discovered that you exported a UTF-8 locale" " but the locale system could not pick up from it" " because it does not exist. The exported locale is" " {locale!r} but it is not supported." ).format(locale=bad_locale) ) raise RuntimeError("\n\n".join(extra)) globals.py 0000644 00000003701 15030210742 0006534 0 ustar 00 import typing import typing as t from threading import local if t.TYPE_CHECKING: import typing_extensions as te from .core import Context _local = local() @typing.overload def get_current_context(silent: "te.Literal[False]" = False) -> "Context": ... @typing.overload def get_current_context(silent: bool = ...) -> t.Optional["Context"]: ... def get_current_context(silent: bool = False) -> t.Optional["Context"]: """Returns the current click context. This can be used as a way to access the current context object from anywhere. This is a more implicit alternative to the :func:`pass_context` decorator. This function is primarily useful for helpers such as :func:`echo` which might be interested in changing its behavior based on the current context. To push the current context, :meth:`Context.scope` can be used. .. versionadded:: 5.0 :param silent: if set to `True` the return value is `None` if no context is available. The default behavior is to raise a :exc:`RuntimeError`. """ try: return t.cast("Context", _local.stack[-1]) except (AttributeError, IndexError) as e: if not silent: raise RuntimeError("There is no active click context.") from e return None def push_context(ctx: "Context") -> None: """Pushes a new context to the current stack.""" _local.__dict__.setdefault("stack", []).append(ctx) def pop_context() -> None: """Removes the top level from the stack.""" _local.stack.pop() def resolve_color_default(color: t.Optional[bool] = None) -> t.Optional[bool]: """Internal helper to get the default value of the color flag. If a value is passed it's returned unchanged, otherwise it's looked up from the current context. """ if color is not None: return color ctx = get_current_context(silent=True) if ctx is not None: return ctx.color return None _termui_impl.py 0000644 00000055634 15030210742 0007612 0 ustar 00 """ This module contains implementations for the termui module. To keep the import time of Click down, some infrequently used functionality is placed in this module and only imported as needed. """ import contextlib import math import os import sys import time import typing as t from gettext import gettext as _ from ._compat import _default_text_stdout from ._compat import CYGWIN from ._compat import get_best_encoding from ._compat import isatty from ._compat import open_stream from ._compat import strip_ansi from ._compat import term_len from ._compat import WIN from .exceptions import ClickException from .utils import echo V = t.TypeVar("V") if os.name == "nt": BEFORE_BAR = "\r" AFTER_BAR = "\n" else: BEFORE_BAR = "\r\033[?25l" AFTER_BAR = "\033[?25h\n" class ProgressBar(t.Generic[V]): def __init__( self, iterable: t.Optional[t.Iterable[V]], length: t.Optional[int] = None, fill_char: str = "#", empty_char: str = " ", bar_template: str = "%(bar)s", info_sep: str = " ", show_eta: bool = True, show_percent: t.Optional[bool] = None, show_pos: bool = False, item_show_func: t.Optional[t.Callable[[t.Optional[V]], t.Optional[str]]] = None, label: t.Optional[str] = None, file: t.Optional[t.TextIO] = None, color: t.Optional[bool] = None, update_min_steps: int = 1, width: int = 30, ) -> None: self.fill_char = fill_char self.empty_char = empty_char self.bar_template = bar_template self.info_sep = info_sep self.show_eta = show_eta self.show_percent = show_percent self.show_pos = show_pos self.item_show_func = item_show_func self.label = label or "" if file is None: file = _default_text_stdout() self.file = file self.color = color self.update_min_steps = update_min_steps self._completed_intervals = 0 self.width = width self.autowidth = width == 0 if length is None: from operator import length_hint length = length_hint(iterable, -1) if length == -1: length = None if iterable is None: if length is None: raise TypeError("iterable or length is required") iterable = t.cast(t.Iterable[V], range(length)) self.iter = iter(iterable) self.length = length self.pos = 0 self.avg: t.List[float] = [] self.start = self.last_eta = time.time() self.eta_known = False self.finished = False self.max_width: t.Optional[int] = None self.entered = False self.current_item: t.Optional[V] = None self.is_hidden = not isatty(self.file) self._last_line: t.Optional[str] = None def __enter__(self) -> "ProgressBar": self.entered = True self.render_progress() return self def __exit__(self, exc_type, exc_value, tb): # type: ignore self.render_finish() def __iter__(self) -> t.Iterator[V]: if not self.entered: raise RuntimeError("You need to use progress bars in a with block.") self.render_progress() return self.generator() def __next__(self) -> V: # Iteration is defined in terms of a generator function, # returned by iter(self); use that to define next(). This works # because `self.iter` is an iterable consumed by that generator, # so it is re-entry safe. Calling `next(self.generator())` # twice works and does "what you want". return next(iter(self)) def render_finish(self) -> None: if self.is_hidden: return self.file.write(AFTER_BAR) self.file.flush() @property def pct(self) -> float: if self.finished: return 1.0 return min(self.pos / (float(self.length or 1) or 1), 1.0) @property def time_per_iteration(self) -> float: if not self.avg: return 0.0 return sum(self.avg) / float(len(self.avg)) @property def eta(self) -> float: if self.length is not None and not self.finished: return self.time_per_iteration * (self.length - self.pos) return 0.0 def format_eta(self) -> str: if self.eta_known: t = int(self.eta) seconds = t % 60 t //= 60 minutes = t % 60 t //= 60 hours = t % 24 t //= 24 if t > 0: return f"{t}d {hours:02}:{minutes:02}:{seconds:02}" else: return f"{hours:02}:{minutes:02}:{seconds:02}" return "" def format_pos(self) -> str: pos = str(self.pos) if self.length is not None: pos += f"/{self.length}" return pos def format_pct(self) -> str: return f"{int(self.pct * 100): 4}%"[1:] def format_bar(self) -> str: if self.length is not None: bar_length = int(self.pct * self.width) bar = self.fill_char * bar_length bar += self.empty_char * (self.width - bar_length) elif self.finished: bar = self.fill_char * self.width else: chars = list(self.empty_char * (self.width or 1)) if self.time_per_iteration != 0: chars[ int( (math.cos(self.pos * self.time_per_iteration) / 2.0 + 0.5) * self.width ) ] = self.fill_char bar = "".join(chars) return bar def format_progress_line(self) -> str: show_percent = self.show_percent info_bits = [] if self.length is not None and show_percent is None: show_percent = not self.show_pos if self.show_pos: info_bits.append(self.format_pos()) if show_percent: info_bits.append(self.format_pct()) if self.show_eta and self.eta_known and not self.finished: info_bits.append(self.format_eta()) if self.item_show_func is not None: item_info = self.item_show_func(self.current_item) if item_info is not None: info_bits.append(item_info) return ( self.bar_template % { "label": self.label, "bar": self.format_bar(), "info": self.info_sep.join(info_bits), } ).rstrip() def render_progress(self) -> None: import shutil if self.is_hidden: # Only output the label as it changes if the output is not a # TTY. Use file=stderr if you expect to be piping stdout. if self._last_line != self.label: self._last_line = self.label echo(self.label, file=self.file, color=self.color) return buf = [] # Update width in case the terminal has been resized if self.autowidth: old_width = self.width self.width = 0 clutter_length = term_len(self.format_progress_line()) new_width = max(0, shutil.get_terminal_size().columns - clutter_length) if new_width < old_width: buf.append(BEFORE_BAR) buf.append(" " * self.max_width) # type: ignore self.max_width = new_width self.width = new_width clear_width = self.width if self.max_width is not None: clear_width = self.max_width buf.append(BEFORE_BAR) line = self.format_progress_line() line_len = term_len(line) if self.max_width is None or self.max_width < line_len: self.max_width = line_len buf.append(line) buf.append(" " * (clear_width - line_len)) line = "".join(buf) # Render the line only if it changed. if line != self._last_line: self._last_line = line echo(line, file=self.file, color=self.color, nl=False) self.file.flush() def make_step(self, n_steps: int) -> None: self.pos += n_steps if self.length is not None and self.pos >= self.length: self.finished = True if (time.time() - self.last_eta) < 1.0: return self.last_eta = time.time() # self.avg is a rolling list of length <= 7 of steps where steps are # defined as time elapsed divided by the total progress through # self.length. if self.pos: step = (time.time() - self.start) / self.pos else: step = time.time() - self.start self.avg = self.avg[-6:] + [step] self.eta_known = self.length is not None def update(self, n_steps: int, current_item: t.Optional[V] = None) -> None: """Update the progress bar by advancing a specified number of steps, and optionally set the ``current_item`` for this new position. :param n_steps: Number of steps to advance. :param current_item: Optional item to set as ``current_item`` for the updated position. .. versionchanged:: 8.0 Added the ``current_item`` optional parameter. .. versionchanged:: 8.0 Only render when the number of steps meets the ``update_min_steps`` threshold. """ if current_item is not None: self.current_item = current_item self._completed_intervals += n_steps if self._completed_intervals >= self.update_min_steps: self.make_step(self._completed_intervals) self.render_progress() self._completed_intervals = 0 def finish(self) -> None: self.eta_known = False self.current_item = None self.finished = True def generator(self) -> t.Iterator[V]: """Return a generator which yields the items added to the bar during construction, and updates the progress bar *after* the yielded block returns. """ # WARNING: the iterator interface for `ProgressBar` relies on # this and only works because this is a simple generator which # doesn't create or manage additional state. If this function # changes, the impact should be evaluated both against # `iter(bar)` and `next(bar)`. `next()` in particular may call # `self.generator()` repeatedly, and this must remain safe in # order for that interface to work. if not self.entered: raise RuntimeError("You need to use progress bars in a with block.") if self.is_hidden: yield from self.iter else: for rv in self.iter: self.current_item = rv # This allows show_item_func to be updated before the # item is processed. Only trigger at the beginning of # the update interval. if self._completed_intervals == 0: self.render_progress() yield rv self.update(1) self.finish() self.render_progress() def pager(generator: t.Iterable[str], color: t.Optional[bool] = None) -> None: """Decide what method to use for paging through text.""" stdout = _default_text_stdout() if not isatty(sys.stdin) or not isatty(stdout): return _nullpager(stdout, generator, color) pager_cmd = (os.environ.get("PAGER", None) or "").strip() if pager_cmd: if WIN: return _tempfilepager(generator, pager_cmd, color) return _pipepager(generator, pager_cmd, color) if os.environ.get("TERM") in ("dumb", "emacs"): return _nullpager(stdout, generator, color) if WIN or sys.platform.startswith("os2"): return _tempfilepager(generator, "more <", color) if hasattr(os, "system") and os.system("(less) 2>/dev/null") == 0: return _pipepager(generator, "less", color) import tempfile fd, filename = tempfile.mkstemp() os.close(fd) try: if hasattr(os, "system") and os.system(f'more "{filename}"') == 0: return _pipepager(generator, "more", color) return _nullpager(stdout, generator, color) finally: os.unlink(filename) def _pipepager(generator: t.Iterable[str], cmd: str, color: t.Optional[bool]) -> None: """Page through text by feeding it to another program. Invoking a pager through this might support colors. """ import subprocess env = dict(os.environ) # If we're piping to less we might support colors under the # condition that cmd_detail = cmd.rsplit("/", 1)[-1].split() if color is None and cmd_detail[0] == "less": less_flags = f"{os.environ.get('LESS', '')}{' '.join(cmd_detail[1:])}" if not less_flags: env["LESS"] = "-R" color = True elif "r" in less_flags or "R" in less_flags: color = True c = subprocess.Popen(cmd, shell=True, stdin=subprocess.PIPE, env=env) stdin = t.cast(t.BinaryIO, c.stdin) encoding = get_best_encoding(stdin) try: for text in generator: if not color: text = strip_ansi(text) stdin.write(text.encode(encoding, "replace")) except (OSError, KeyboardInterrupt): pass else: stdin.close() # Less doesn't respect ^C, but catches it for its own UI purposes (aborting # search or other commands inside less). # # That means when the user hits ^C, the parent process (click) terminates, # but less is still alive, paging the output and messing up the terminal. # # If the user wants to make the pager exit on ^C, they should set # `LESS='-K'`. It's not our decision to make. while True: try: c.wait() except KeyboardInterrupt: pass else: break def _tempfilepager( generator: t.Iterable[str], cmd: str, color: t.Optional[bool] ) -> None: """Page through text by invoking a program on a temporary file.""" import tempfile fd, filename = tempfile.mkstemp() # TODO: This never terminates if the passed generator never terminates. text = "".join(generator) if not color: text = strip_ansi(text) encoding = get_best_encoding(sys.stdout) with open_stream(filename, "wb")[0] as f: f.write(text.encode(encoding)) try: os.system(f'{cmd} "{filename}"') finally: os.close(fd) os.unlink(filename) def _nullpager( stream: t.TextIO, generator: t.Iterable[str], color: t.Optional[bool] ) -> None: """Simply print unformatted text. This is the ultimate fallback.""" for text in generator: if not color: text = strip_ansi(text) stream.write(text) class Editor: def __init__( self, editor: t.Optional[str] = None, env: t.Optional[t.Mapping[str, str]] = None, require_save: bool = True, extension: str = ".txt", ) -> None: self.editor = editor self.env = env self.require_save = require_save self.extension = extension def get_editor(self) -> str: if self.editor is not None: return self.editor for key in "VISUAL", "EDITOR": rv = os.environ.get(key) if rv: return rv if WIN: return "notepad" for editor in "sensible-editor", "vim", "nano": if os.system(f"which {editor} >/dev/null 2>&1") == 0: return editor return "vi" def edit_file(self, filename: str) -> None: import subprocess editor = self.get_editor() environ: t.Optional[t.Dict[str, str]] = None if self.env: environ = os.environ.copy() environ.update(self.env) try: c = subprocess.Popen(f'{editor} "{filename}"', env=environ, shell=True) exit_code = c.wait() if exit_code != 0: raise ClickException( _("{editor}: Editing failed").format(editor=editor) ) except OSError as e: raise ClickException( _("{editor}: Editing failed: {e}").format(editor=editor, e=e) ) from e def edit(self, text: t.Optional[t.AnyStr]) -> t.Optional[t.AnyStr]: import tempfile if not text: data = b"" elif isinstance(text, (bytes, bytearray)): data = text else: if text and not text.endswith("\n"): text += "\n" if WIN: data = text.replace("\n", "\r\n").encode("utf-8-sig") else: data = text.encode("utf-8") fd, name = tempfile.mkstemp(prefix="editor-", suffix=self.extension) f: t.BinaryIO try: with os.fdopen(fd, "wb") as f: f.write(data) # If the filesystem resolution is 1 second, like Mac OS # 10.12 Extended, or 2 seconds, like FAT32, and the editor # closes very fast, require_save can fail. Set the modified # time to be 2 seconds in the past to work around this. os.utime(name, (os.path.getatime(name), os.path.getmtime(name) - 2)) # Depending on the resolution, the exact value might not be # recorded, so get the new recorded value. timestamp = os.path.getmtime(name) self.edit_file(name) if self.require_save and os.path.getmtime(name) == timestamp: return None with open(name, "rb") as f: rv = f.read() if isinstance(text, (bytes, bytearray)): return rv return rv.decode("utf-8-sig").replace("\r\n", "\n") # type: ignore finally: os.unlink(name) def open_url(url: str, wait: bool = False, locate: bool = False) -> int: import subprocess def _unquote_file(url: str) -> str: from urllib.parse import unquote if url.startswith("file://"): url = unquote(url[7:]) return url if sys.platform == "darwin": args = ["open"] if wait: args.append("-W") if locate: args.append("-R") args.append(_unquote_file(url)) null = open("/dev/null", "w") try: return subprocess.Popen(args, stderr=null).wait() finally: null.close() elif WIN: if locate: url = _unquote_file(url.replace('"', "")) args = f'explorer /select,"{url}"' else: url = url.replace('"', "") wait_str = "/WAIT" if wait else "" args = f'start {wait_str} "" "{url}"' return os.system(args) elif CYGWIN: if locate: url = os.path.dirname(_unquote_file(url).replace('"', "")) args = f'cygstart "{url}"' else: url = url.replace('"', "") wait_str = "-w" if wait else "" args = f'cygstart {wait_str} "{url}"' return os.system(args) try: if locate: url = os.path.dirname(_unquote_file(url)) or "." else: url = _unquote_file(url) c = subprocess.Popen(["xdg-open", url]) if wait: return c.wait() return 0 except OSError: if url.startswith(("http://", "https://")) and not locate and not wait: import webbrowser webbrowser.open(url) return 0 return 1 def _translate_ch_to_exc(ch: str) -> t.Optional[BaseException]: if ch == "\x03": raise KeyboardInterrupt() if ch == "\x04" and not WIN: # Unix-like, Ctrl+D raise EOFError() if ch == "\x1a" and WIN: # Windows, Ctrl+Z raise EOFError() return None if WIN: import msvcrt @contextlib.contextmanager def raw_terminal() -> t.Iterator[int]: yield -1 def getchar(echo: bool) -> str: # The function `getch` will return a bytes object corresponding to # the pressed character. Since Windows 10 build 1803, it will also # return \x00 when called a second time after pressing a regular key. # # `getwch` does not share this probably-bugged behavior. Moreover, it # returns a Unicode object by default, which is what we want. # # Either of these functions will return \x00 or \xe0 to indicate # a special key, and you need to call the same function again to get # the "rest" of the code. The fun part is that \u00e0 is # "latin small letter a with grave", so if you type that on a French # keyboard, you _also_ get a \xe0. # E.g., consider the Up arrow. This returns \xe0 and then \x48. The # resulting Unicode string reads as "a with grave" + "capital H". # This is indistinguishable from when the user actually types # "a with grave" and then "capital H". # # When \xe0 is returned, we assume it's part of a special-key sequence # and call `getwch` again, but that means that when the user types # the \u00e0 character, `getchar` doesn't return until a second # character is typed. # The alternative is returning immediately, but that would mess up # cross-platform handling of arrow keys and others that start with # \xe0. Another option is using `getch`, but then we can't reliably # read non-ASCII characters, because return values of `getch` are # limited to the current 8-bit codepage. # # Anyway, Click doesn't claim to do this Right(tm), and using `getwch` # is doing the right thing in more situations than with `getch`. func: t.Callable[[], str] if echo: func = msvcrt.getwche # type: ignore else: func = msvcrt.getwch # type: ignore rv = func() if rv in ("\x00", "\xe0"): # \x00 and \xe0 are control characters that indicate special key, # see above. rv += func() _translate_ch_to_exc(rv) return rv else: import tty import termios @contextlib.contextmanager def raw_terminal() -> t.Iterator[int]: f: t.Optional[t.TextIO] fd: int if not isatty(sys.stdin): f = open("/dev/tty") fd = f.fileno() else: fd = sys.stdin.fileno() f = None try: old_settings = termios.tcgetattr(fd) try: tty.setraw(fd) yield fd finally: termios.tcsetattr(fd, termios.TCSADRAIN, old_settings) sys.stdout.flush() if f is not None: f.close() except termios.error: pass def getchar(echo: bool) -> str: with raw_terminal() as fd: ch = os.read(fd, 32).decode(get_best_encoding(sys.stdin), "replace") if echo and isatty(sys.stdout): sys.stdout.write(ch) _translate_ch_to_exc(ch) return ch parser.py 0000644 00000045144 15030210742 0006414 0 ustar 00 """ This module started out as largely a copy paste from the stdlib's optparse module with the features removed that we do not need from optparse because we implement them in Click on a higher level (for instance type handling, help formatting and a lot more). The plan is to remove more and more from here over time. The reason this is a different module and not optparse from the stdlib is that there are differences in 2.x and 3.x about the error messages generated and optparse in the stdlib uses gettext for no good reason and might cause us issues. Click uses parts of optparse written by Gregory P. Ward and maintained by the Python Software Foundation. This is limited to code in parser.py. Copyright 2001-2006 Gregory P. Ward. All rights reserved. Copyright 2002-2006 Python Software Foundation. All rights reserved. """ # This code uses parts of optparse written by Gregory P. Ward and # maintained by the Python Software Foundation. # Copyright 2001-2006 Gregory P. Ward # Copyright 2002-2006 Python Software Foundation import typing as t from collections import deque from gettext import gettext as _ from gettext import ngettext from .exceptions import BadArgumentUsage from .exceptions import BadOptionUsage from .exceptions import NoSuchOption from .exceptions import UsageError if t.TYPE_CHECKING: import typing_extensions as te from .core import Argument as CoreArgument from .core import Context from .core import Option as CoreOption from .core import Parameter as CoreParameter V = t.TypeVar("V") # Sentinel value that indicates an option was passed as a flag without a # value but is not a flag option. Option.consume_value uses this to # prompt or use the flag_value. _flag_needs_value = object() def _unpack_args( args: t.Sequence[str], nargs_spec: t.Sequence[int] ) -> t.Tuple[t.Sequence[t.Union[str, t.Sequence[t.Optional[str]], None]], t.List[str]]: """Given an iterable of arguments and an iterable of nargs specifications, it returns a tuple with all the unpacked arguments at the first index and all remaining arguments as the second. The nargs specification is the number of arguments that should be consumed or `-1` to indicate that this position should eat up all the remainders. Missing items are filled with `None`. """ args = deque(args) nargs_spec = deque(nargs_spec) rv: t.List[t.Union[str, t.Tuple[t.Optional[str], ...], None]] = [] spos: t.Optional[int] = None def _fetch(c: "te.Deque[V]") -> t.Optional[V]: try: if spos is None: return c.popleft() else: return c.pop() except IndexError: return None while nargs_spec: nargs = _fetch(nargs_spec) if nargs is None: continue if nargs == 1: rv.append(_fetch(args)) elif nargs > 1: x = [_fetch(args) for _ in range(nargs)] # If we're reversed, we're pulling in the arguments in reverse, # so we need to turn them around. if spos is not None: x.reverse() rv.append(tuple(x)) elif nargs < 0: if spos is not None: raise TypeError("Cannot have two nargs < 0") spos = len(rv) rv.append(None) # spos is the position of the wildcard (star). If it's not `None`, # we fill it with the remainder. if spos is not None: rv[spos] = tuple(args) args = [] rv[spos + 1 :] = reversed(rv[spos + 1 :]) return tuple(rv), list(args) def split_opt(opt: str) -> t.Tuple[str, str]: first = opt[:1] if first.isalnum(): return "", opt if opt[1:2] == first: return opt[:2], opt[2:] return first, opt[1:] def normalize_opt(opt: str, ctx: t.Optional["Context"]) -> str: if ctx is None or ctx.token_normalize_func is None: return opt prefix, opt = split_opt(opt) return f"{prefix}{ctx.token_normalize_func(opt)}" def split_arg_string(string: str) -> t.List[str]: """Split an argument string as with :func:`shlex.split`, but don't fail if the string is incomplete. Ignores a missing closing quote or incomplete escape sequence and uses the partial token as-is. .. code-block:: python split_arg_string("example 'my file") ["example", "my file"] split_arg_string("example my\\") ["example", "my"] :param string: String to split. """ import shlex lex = shlex.shlex(string, posix=True) lex.whitespace_split = True lex.commenters = "" out = [] try: for token in lex: out.append(token) except ValueError: # Raised when end-of-string is reached in an invalid state. Use # the partial token as-is. The quote or escape character is in # lex.state, not lex.token. out.append(lex.token) return out class Option: def __init__( self, obj: "CoreOption", opts: t.Sequence[str], dest: t.Optional[str], action: t.Optional[str] = None, nargs: int = 1, const: t.Optional[t.Any] = None, ): self._short_opts = [] self._long_opts = [] self.prefixes = set() for opt in opts: prefix, value = split_opt(opt) if not prefix: raise ValueError(f"Invalid start character for option ({opt})") self.prefixes.add(prefix[0]) if len(prefix) == 1 and len(value) == 1: self._short_opts.append(opt) else: self._long_opts.append(opt) self.prefixes.add(prefix) if action is None: action = "store" self.dest = dest self.action = action self.nargs = nargs self.const = const self.obj = obj @property def takes_value(self) -> bool: return self.action in ("store", "append") def process(self, value: str, state: "ParsingState") -> None: if self.action == "store": state.opts[self.dest] = value # type: ignore elif self.action == "store_const": state.opts[self.dest] = self.const # type: ignore elif self.action == "append": state.opts.setdefault(self.dest, []).append(value) # type: ignore elif self.action == "append_const": state.opts.setdefault(self.dest, []).append(self.const) # type: ignore elif self.action == "count": state.opts[self.dest] = state.opts.get(self.dest, 0) + 1 # type: ignore else: raise ValueError(f"unknown action '{self.action}'") state.order.append(self.obj) class Argument: def __init__(self, obj: "CoreArgument", dest: t.Optional[str], nargs: int = 1): self.dest = dest self.nargs = nargs self.obj = obj def process( self, value: t.Union[t.Optional[str], t.Sequence[t.Optional[str]]], state: "ParsingState", ) -> None: if self.nargs > 1: assert value is not None holes = sum(1 for x in value if x is None) if holes == len(value): value = None elif holes != 0: raise BadArgumentUsage( _("Argument {name!r} takes {nargs} values.").format( name=self.dest, nargs=self.nargs ) ) if self.nargs == -1 and self.obj.envvar is not None and value == (): # Replace empty tuple with None so that a value from the # environment may be tried. value = None state.opts[self.dest] = value # type: ignore state.order.append(self.obj) class ParsingState: def __init__(self, rargs: t.List[str]) -> None: self.opts: t.Dict[str, t.Any] = {} self.largs: t.List[str] = [] self.rargs = rargs self.order: t.List["CoreParameter"] = [] class OptionParser: """The option parser is an internal class that is ultimately used to parse options and arguments. It's modelled after optparse and brings a similar but vastly simplified API. It should generally not be used directly as the high level Click classes wrap it for you. It's not nearly as extensible as optparse or argparse as it does not implement features that are implemented on a higher level (such as types or defaults). :param ctx: optionally the :class:`~click.Context` where this parser should go with. """ def __init__(self, ctx: t.Optional["Context"] = None) -> None: #: The :class:`~click.Context` for this parser. This might be #: `None` for some advanced use cases. self.ctx = ctx #: This controls how the parser deals with interspersed arguments. #: If this is set to `False`, the parser will stop on the first #: non-option. Click uses this to implement nested subcommands #: safely. self.allow_interspersed_args = True #: This tells the parser how to deal with unknown options. By #: default it will error out (which is sensible), but there is a #: second mode where it will ignore it and continue processing #: after shifting all the unknown options into the resulting args. self.ignore_unknown_options = False if ctx is not None: self.allow_interspersed_args = ctx.allow_interspersed_args self.ignore_unknown_options = ctx.ignore_unknown_options self._short_opt: t.Dict[str, Option] = {} self._long_opt: t.Dict[str, Option] = {} self._opt_prefixes = {"-", "--"} self._args: t.List[Argument] = [] def add_option( self, obj: "CoreOption", opts: t.Sequence[str], dest: t.Optional[str], action: t.Optional[str] = None, nargs: int = 1, const: t.Optional[t.Any] = None, ) -> None: """Adds a new option named `dest` to the parser. The destination is not inferred (unlike with optparse) and needs to be explicitly provided. Action can be any of ``store``, ``store_const``, ``append``, ``append_const`` or ``count``. The `obj` can be used to identify the option in the order list that is returned from the parser. """ opts = [normalize_opt(opt, self.ctx) for opt in opts] option = Option(obj, opts, dest, action=action, nargs=nargs, const=const) self._opt_prefixes.update(option.prefixes) for opt in option._short_opts: self._short_opt[opt] = option for opt in option._long_opts: self._long_opt[opt] = option def add_argument( self, obj: "CoreArgument", dest: t.Optional[str], nargs: int = 1 ) -> None: """Adds a positional argument named `dest` to the parser. The `obj` can be used to identify the option in the order list that is returned from the parser. """ self._args.append(Argument(obj, dest=dest, nargs=nargs)) def parse_args( self, args: t.List[str] ) -> t.Tuple[t.Dict[str, t.Any], t.List[str], t.List["CoreParameter"]]: """Parses positional arguments and returns ``(values, args, order)`` for the parsed options and arguments as well as the leftover arguments if there are any. The order is a list of objects as they appear on the command line. If arguments appear multiple times they will be memorized multiple times as well. """ state = ParsingState(args) try: self._process_args_for_options(state) self._process_args_for_args(state) except UsageError: if self.ctx is None or not self.ctx.resilient_parsing: raise return state.opts, state.largs, state.order def _process_args_for_args(self, state: ParsingState) -> None: pargs, args = _unpack_args( state.largs + state.rargs, [x.nargs for x in self._args] ) for idx, arg in enumerate(self._args): arg.process(pargs[idx], state) state.largs = args state.rargs = [] def _process_args_for_options(self, state: ParsingState) -> None: while state.rargs: arg = state.rargs.pop(0) arglen = len(arg) # Double dashes always handled explicitly regardless of what # prefixes are valid. if arg == "--": return elif arg[:1] in self._opt_prefixes and arglen > 1: self._process_opts(arg, state) elif self.allow_interspersed_args: state.largs.append(arg) else: state.rargs.insert(0, arg) return # Say this is the original argument list: # [arg0, arg1, ..., arg(i-1), arg(i), arg(i+1), ..., arg(N-1)] # ^ # (we are about to process arg(i)). # # Then rargs is [arg(i), ..., arg(N-1)] and largs is a *subset* of # [arg0, ..., arg(i-1)] (any options and their arguments will have # been removed from largs). # # The while loop will usually consume 1 or more arguments per pass. # If it consumes 1 (eg. arg is an option that takes no arguments), # then after _process_arg() is done the situation is: # # largs = subset of [arg0, ..., arg(i)] # rargs = [arg(i+1), ..., arg(N-1)] # # If allow_interspersed_args is false, largs will always be # *empty* -- still a subset of [arg0, ..., arg(i-1)], but # not a very interesting subset! def _match_long_opt( self, opt: str, explicit_value: t.Optional[str], state: ParsingState ) -> None: if opt not in self._long_opt: from difflib import get_close_matches possibilities = get_close_matches(opt, self._long_opt) raise NoSuchOption(opt, possibilities=possibilities, ctx=self.ctx) option = self._long_opt[opt] if option.takes_value: # At this point it's safe to modify rargs by injecting the # explicit value, because no exception is raised in this # branch. This means that the inserted value will be fully # consumed. if explicit_value is not None: state.rargs.insert(0, explicit_value) value = self._get_value_from_state(opt, option, state) elif explicit_value is not None: raise BadOptionUsage( opt, _("Option {name!r} does not take a value.").format(name=opt) ) else: value = None option.process(value, state) def _match_short_opt(self, arg: str, state: ParsingState) -> None: stop = False i = 1 prefix = arg[0] unknown_options = [] for ch in arg[1:]: opt = normalize_opt(f"{prefix}{ch}", self.ctx) option = self._short_opt.get(opt) i += 1 if not option: if self.ignore_unknown_options: unknown_options.append(ch) continue raise NoSuchOption(opt, ctx=self.ctx) if option.takes_value: # Any characters left in arg? Pretend they're the # next arg, and stop consuming characters of arg. if i < len(arg): state.rargs.insert(0, arg[i:]) stop = True value = self._get_value_from_state(opt, option, state) else: value = None option.process(value, state) if stop: break # If we got any unknown options we re-combinate the string of the # remaining options and re-attach the prefix, then report that # to the state as new larg. This way there is basic combinatorics # that can be achieved while still ignoring unknown arguments. if self.ignore_unknown_options and unknown_options: state.largs.append(f"{prefix}{''.join(unknown_options)}") def _get_value_from_state( self, option_name: str, option: Option, state: ParsingState ) -> t.Any: nargs = option.nargs if len(state.rargs) < nargs: if option.obj._flag_needs_value: # Option allows omitting the value. value = _flag_needs_value else: raise BadOptionUsage( option_name, ngettext( "Option {name!r} requires an argument.", "Option {name!r} requires {nargs} arguments.", nargs, ).format(name=option_name, nargs=nargs), ) elif nargs == 1: next_rarg = state.rargs[0] if ( option.obj._flag_needs_value and isinstance(next_rarg, str) and next_rarg[:1] in self._opt_prefixes and len(next_rarg) > 1 ): # The next arg looks like the start of an option, don't # use it as the value if omitting the value is allowed. value = _flag_needs_value else: value = state.rargs.pop(0) else: value = tuple(state.rargs[:nargs]) del state.rargs[:nargs] return value def _process_opts(self, arg: str, state: ParsingState) -> None: explicit_value = None # Long option handling happens in two parts. The first part is # supporting explicitly attached values. In any case, we will try # to long match the option first. if "=" in arg: long_opt, explicit_value = arg.split("=", 1) else: long_opt = arg norm_long_opt = normalize_opt(long_opt, self.ctx) # At this point we will match the (assumed) long option through # the long option matching code. Note that this allows options # like "-foo" to be matched as long options. try: self._match_long_opt(norm_long_opt, explicit_value, state) except NoSuchOption: # At this point the long option matching failed, and we need # to try with short options. However there is a special rule # which says, that if we have a two character options prefix # (applies to "--foo" for instance), we do not dispatch to the # short option code and will instead raise the no option # error. if arg[:2] not in self._opt_prefixes: self._match_short_opt(arg, state) return if not self.ignore_unknown_options: raise state.largs.append(arg) decorators.py 0000644 00000035020 15030210742 0007255 0 ustar 00 import inspect import types import typing as t from functools import update_wrapper from gettext import gettext as _ from .core import Argument from .core import Command from .core import Context from .core import Group from .core import Option from .core import Parameter from .globals import get_current_context from .utils import echo F = t.TypeVar("F", bound=t.Callable[..., t.Any]) FC = t.TypeVar("FC", t.Callable[..., t.Any], Command) def pass_context(f: F) -> F: """Marks a callback as wanting to receive the current context object as first argument. """ def new_func(*args, **kwargs): # type: ignore return f(get_current_context(), *args, **kwargs) return update_wrapper(t.cast(F, new_func), f) def pass_obj(f: F) -> F: """Similar to :func:`pass_context`, but only pass the object on the context onwards (:attr:`Context.obj`). This is useful if that object represents the state of a nested system. """ def new_func(*args, **kwargs): # type: ignore return f(get_current_context().obj, *args, **kwargs) return update_wrapper(t.cast(F, new_func), f) def make_pass_decorator( object_type: t.Type, ensure: bool = False ) -> "t.Callable[[F], F]": """Given an object type this creates a decorator that will work similar to :func:`pass_obj` but instead of passing the object of the current context, it will find the innermost context of type :func:`object_type`. This generates a decorator that works roughly like this:: from functools import update_wrapper def decorator(f): @pass_context def new_func(ctx, *args, **kwargs): obj = ctx.find_object(object_type) return ctx.invoke(f, obj, *args, **kwargs) return update_wrapper(new_func, f) return decorator :param object_type: the type of the object to pass. :param ensure: if set to `True`, a new object will be created and remembered on the context if it's not there yet. """ def decorator(f: F) -> F: def new_func(*args, **kwargs): # type: ignore ctx = get_current_context() if ensure: obj = ctx.ensure_object(object_type) else: obj = ctx.find_object(object_type) if obj is None: raise RuntimeError( "Managed to invoke callback without a context" f" object of type {object_type.__name__!r}" " existing." ) return ctx.invoke(f, obj, *args, **kwargs) return update_wrapper(t.cast(F, new_func), f) return decorator def pass_meta_key( key: str, *, doc_description: t.Optional[str] = None ) -> "t.Callable[[F], F]": """Create a decorator that passes a key from :attr:`click.Context.meta` as the first argument to the decorated function. :param key: Key in ``Context.meta`` to pass. :param doc_description: Description of the object being passed, inserted into the decorator's docstring. Defaults to "the 'key' key from Context.meta". .. versionadded:: 8.0 """ def decorator(f: F) -> F: def new_func(*args, **kwargs): # type: ignore ctx = get_current_context() obj = ctx.meta[key] return ctx.invoke(f, obj, *args, **kwargs) return update_wrapper(t.cast(F, new_func), f) if doc_description is None: doc_description = f"the {key!r} key from :attr:`click.Context.meta`" decorator.__doc__ = ( f"Decorator that passes {doc_description} as the first argument" " to the decorated function." ) return decorator def _make_command( f: F, name: t.Optional[str], attrs: t.MutableMapping[str, t.Any], cls: t.Type[Command], ) -> Command: if isinstance(f, Command): raise TypeError("Attempted to convert a callback into a command twice.") try: params = f.__click_params__ # type: ignore params.reverse() del f.__click_params__ # type: ignore except AttributeError: params = [] help = attrs.get("help") if help is None: help = inspect.getdoc(f) else: help = inspect.cleandoc(help) attrs["help"] = help return cls( name=name or f.__name__.lower().replace("_", "-"), callback=f, params=params, **attrs, ) def command( name: t.Optional[str] = None, cls: t.Optional[t.Type[Command]] = None, **attrs: t.Any, ) -> t.Callable[[F], Command]: r"""Creates a new :class:`Command` and uses the decorated function as callback. This will also automatically attach all decorated :func:`option`\s and :func:`argument`\s as parameters to the command. The name of the command defaults to the name of the function with underscores replaced by dashes. If you want to change that, you can pass the intended name as the first argument. All keyword arguments are forwarded to the underlying command class. Once decorated the function turns into a :class:`Command` instance that can be invoked as a command line utility or be attached to a command :class:`Group`. :param name: the name of the command. This defaults to the function name with underscores replaced by dashes. :param cls: the command class to instantiate. This defaults to :class:`Command`. """ if cls is None: cls = Command def decorator(f: t.Callable[..., t.Any]) -> Command: cmd = _make_command(f, name, attrs, cls) # type: ignore cmd.__doc__ = f.__doc__ return cmd return decorator def group(name: t.Optional[str] = None, **attrs: t.Any) -> t.Callable[[F], Group]: """Creates a new :class:`Group` with a function as callback. This works otherwise the same as :func:`command` just that the `cls` parameter is set to :class:`Group`. """ attrs.setdefault("cls", Group) return t.cast(Group, command(name, **attrs)) def _param_memo(f: FC, param: Parameter) -> None: if isinstance(f, Command): f.params.append(param) else: if not hasattr(f, "__click_params__"): f.__click_params__ = [] # type: ignore f.__click_params__.append(param) # type: ignore def argument(*param_decls: str, **attrs: t.Any) -> t.Callable[[FC], FC]: """Attaches an argument to the command. All positional arguments are passed as parameter declarations to :class:`Argument`; all keyword arguments are forwarded unchanged (except ``cls``). This is equivalent to creating an :class:`Argument` instance manually and attaching it to the :attr:`Command.params` list. :param cls: the argument class to instantiate. This defaults to :class:`Argument`. """ def decorator(f: FC) -> FC: ArgumentClass = attrs.pop("cls", Argument) _param_memo(f, ArgumentClass(param_decls, **attrs)) return f return decorator def option(*param_decls: str, **attrs: t.Any) -> t.Callable[[FC], FC]: """Attaches an option to the command. All positional arguments are passed as parameter declarations to :class:`Option`; all keyword arguments are forwarded unchanged (except ``cls``). This is equivalent to creating an :class:`Option` instance manually and attaching it to the :attr:`Command.params` list. :param cls: the option class to instantiate. This defaults to :class:`Option`. """ def decorator(f: FC) -> FC: # Issue 926, copy attrs, so pre-defined options can re-use the same cls= option_attrs = attrs.copy() if "help" in option_attrs: option_attrs["help"] = inspect.cleandoc(option_attrs["help"]) OptionClass = option_attrs.pop("cls", Option) _param_memo(f, OptionClass(param_decls, **option_attrs)) return f return decorator def confirmation_option(*param_decls: str, **kwargs: t.Any) -> t.Callable[[FC], FC]: """Add a ``--yes`` option which shows a prompt before continuing if not passed. If the prompt is declined, the program will exit. :param param_decls: One or more option names. Defaults to the single value ``"--yes"``. :param kwargs: Extra arguments are passed to :func:`option`. """ def callback(ctx: Context, param: Parameter, value: bool) -> None: if not value: ctx.abort() if not param_decls: param_decls = ("--yes",) kwargs.setdefault("is_flag", True) kwargs.setdefault("callback", callback) kwargs.setdefault("expose_value", False) kwargs.setdefault("prompt", "Do you want to continue?") kwargs.setdefault("help", "Confirm the action without prompting.") return option(*param_decls, **kwargs) def password_option(*param_decls: str, **kwargs: t.Any) -> t.Callable[[FC], FC]: """Add a ``--password`` option which prompts for a password, hiding input and asking to enter the value again for confirmation. :param param_decls: One or more option names. Defaults to the single value ``"--password"``. :param kwargs: Extra arguments are passed to :func:`option`. """ if not param_decls: param_decls = ("--password",) kwargs.setdefault("prompt", True) kwargs.setdefault("confirmation_prompt", True) kwargs.setdefault("hide_input", True) return option(*param_decls, **kwargs) def version_option( version: t.Optional[str] = None, *param_decls: str, package_name: t.Optional[str] = None, prog_name: t.Optional[str] = None, message: t.Optional[str] = None, **kwargs: t.Any, ) -> t.Callable[[FC], FC]: """Add a ``--version`` option which immediately prints the version number and exits the program. If ``version`` is not provided, Click will try to detect it using :func:`importlib.metadata.version` to get the version for the ``package_name``. On Python < 3.8, the ``importlib_metadata`` backport must be installed. If ``package_name`` is not provided, Click will try to detect it by inspecting the stack frames. This will be used to detect the version, so it must match the name of the installed package. :param version: The version number to show. If not provided, Click will try to detect it. :param param_decls: One or more option names. Defaults to the single value ``"--version"``. :param package_name: The package name to detect the version from. If not provided, Click will try to detect it. :param prog_name: The name of the CLI to show in the message. If not provided, it will be detected from the command. :param message: The message to show. The values ``%(prog)s``, ``%(package)s``, and ``%(version)s`` are available. Defaults to ``"%(prog)s, version %(version)s"``. :param kwargs: Extra arguments are passed to :func:`option`. :raise RuntimeError: ``version`` could not be detected. .. versionchanged:: 8.0 Add the ``package_name`` parameter, and the ``%(package)s`` value for messages. .. versionchanged:: 8.0 Use :mod:`importlib.metadata` instead of ``pkg_resources``. The version is detected based on the package name, not the entry point name. The Python package name must match the installed package name, or be passed with ``package_name=``. """ if message is None: message = _("%(prog)s, version %(version)s") if version is None and package_name is None: frame = inspect.currentframe() f_back = frame.f_back if frame is not None else None f_globals = f_back.f_globals if f_back is not None else None # break reference cycle # https://docs.python.org/3/library/inspect.html#the-interpreter-stack del frame if f_globals is not None: package_name = f_globals.get("__name__") if package_name == "__main__": package_name = f_globals.get("__package__") if package_name: package_name = package_name.partition(".")[0] def callback(ctx: Context, param: Parameter, value: bool) -> None: if not value or ctx.resilient_parsing: return nonlocal prog_name nonlocal version if prog_name is None: prog_name = ctx.find_root().info_name if version is None and package_name is not None: metadata: t.Optional[types.ModuleType] try: from importlib import metadata # type: ignore except ImportError: # Python < 3.8 import importlib_metadata as metadata # type: ignore try: version = metadata.version(package_name) # type: ignore except metadata.PackageNotFoundError: # type: ignore raise RuntimeError( f"{package_name!r} is not installed. Try passing" " 'package_name' instead." ) from None if version is None: raise RuntimeError( f"Could not determine the version for {package_name!r} automatically." ) echo( t.cast(str, message) % {"prog": prog_name, "package": package_name, "version": version}, color=ctx.color, ) ctx.exit() if not param_decls: param_decls = ("--version",) kwargs.setdefault("is_flag", True) kwargs.setdefault("expose_value", False) kwargs.setdefault("is_eager", True) kwargs.setdefault("help", _("Show the version and exit.")) kwargs["callback"] = callback return option(*param_decls, **kwargs) def help_option(*param_decls: str, **kwargs: t.Any) -> t.Callable[[FC], FC]: """Add a ``--help`` option which immediately prints the help page and exits the program. This is usually unnecessary, as the ``--help`` option is added to each command automatically unless ``add_help_option=False`` is passed. :param param_decls: One or more option names. Defaults to the single value ``"--help"``. :param kwargs: Extra arguments are passed to :func:`option`. """ def callback(ctx: Context, param: Parameter, value: bool) -> None: if not value or ctx.resilient_parsing: return echo(ctx.get_help(), color=ctx.color) ctx.exit() if not param_decls: param_decls = ("--help",) kwargs.setdefault("is_flag", True) kwargs.setdefault("expose_value", False) kwargs.setdefault("is_eager", True) kwargs.setdefault("help", _("Show this message and exit.")) kwargs["callback"] = callback return option(*param_decls, **kwargs) __pycache__/decorators.cpython-310.pyc 0000644 00000033671 15030210742 0013626 0 ustar 00 o �+ca: � @ s� d dl Z d dlZd dlZd dlmZ d dlmZ ddlm Z ddlm Z ddlmZ ddlmZ dd lm Z dd lmZ ddlmZ ddlmZ ejd ejdejf d�Ze�dejdejf e �Zdedefdd�Zdedefdd�Z dAdejdeddfdd�Zdd�dedeje ddfd d!�Zded"eje d#ej eejf d$eje de f d%d&�Z! dBd"eje d$ejeje d#ejdejege f fd'd(�Z"dCd"eje d#ejdejegef fd)d*�Z#ded+eddfd,d-�Z$d.ed#ejdejegef fd/d0�Z%d.ed#ejdejegef fd1d2�Z&d.ed3ejdejegef fd4d5�Z'd.ed3ejdejegef fd6d7�Z( dCdddd8�d9eje d.ed:eje d;eje d<eje d3ejdejegef fd=d>�Z)d.ed3ejdejegef fd?d@�Z*dS )D� N)�update_wrapper)�gettext� )�Argument)�Command)�Context)�Group)�Option)� Parameter��get_current_context)�echo�F.)�bound�FC�f�returnc � � fdd�}t t�t|�� �S )z]Marks a callback as wanting to receive the current context object as first argument. c s � t � g| �R i |��S �Nr ��args�kwargs�r � �2/usr/lib/python3/dist-packages/click/decorators.py�new_func s zpass_context.<locals>.new_func�r �t�castr �r r r r r �pass_context s r c r )z�Similar to :func:`pass_context`, but only pass the object on the context onwards (:attr:`Context.obj`). This is useful if that object represents the state of a nested system. c s � t � jg| �R i |��S r )r �objr r r r r % s zpass_obj.<locals>.new_funcr r r r r �pass_obj s r"