Primer commit

This commit is contained in:
Mongar28
2024-02-03 21:52:43 -05:00
commit b5c17f85f1
3325 changed files with 855400 additions and 0 deletions

View File

@@ -0,0 +1,13 @@
# Copyright 2012-2023, Andrey Kislyuk and argcomplete contributors.
# Licensed under the Apache License. See https://github.com/kislyuk/argcomplete for more info.
from . import completers
from .completers import ChoicesCompleter, DirectoriesCompleter, EnvironCompleter, FilesCompleter, SuppressCompleter
from .exceptions import ArgcompleteException
from .finders import CompletionFinder, ExclusiveCompletionFinder, safe_actions
from .io import debug, mute_stderr, warn
from .lexers import split_line
from .shell_integration import shellcode
autocomplete = CompletionFinder()
autocomplete.__doc__ = """ Use this to access argcomplete. See :meth:`argcomplete.CompletionFinder.__call__()`. """

View File

@@ -0,0 +1,64 @@
"""
Utility for locating the module (or package's __init__.py)
associated with a given console_script name
and verifying it contains the PYTHON_ARGCOMPLETE_OK marker.
Such scripts are automatically generated and cannot contain
the marker themselves, so we defer to the containing module or package.
For more information on setuptools console_scripts, see
https://setuptools.readthedocs.io/en/latest/setuptools.html#automatic-script-creation
Intended to be invoked by argcomplete's global completion function.
"""
import os
import sys
from importlib.metadata import EntryPoint
from importlib.metadata import entry_points as importlib_entry_points
from typing import Iterable
from ._check_module import ArgcompleteMarkerNotFound, find
def main():
# Argument is the full path to the console script.
script_path = sys.argv[1]
# Find the module and function names that correspond to this
# assuming it is actually a console script.
name = os.path.basename(script_path)
entry_points: Iterable[EntryPoint] = importlib_entry_points() # type:ignore
# Python 3.12+ returns a tuple of entry point objects
# whereas <=3.11 returns a SelectableGroups object
if sys.version_info < (3, 12):
entry_points = entry_points["console_scripts"] # type:ignore
entry_points = [ep for ep in entry_points if ep.name == name and ep.group == "console_scripts"] # type:ignore
if not entry_points:
raise ArgcompleteMarkerNotFound("no entry point found matching script")
entry_point = entry_points[0]
module_name, function_name = entry_point.value.split(":", 1)
# Check this looks like the script we really expected.
with open(script_path) as f:
script = f.read()
if "from {} import {}".format(module_name, function_name) not in script:
raise ArgcompleteMarkerNotFound("does not appear to be a console script")
if "sys.exit({}())".format(function_name) not in script:
raise ArgcompleteMarkerNotFound("does not appear to be a console script")
# Look for the argcomplete marker in the script it imports.
with open(find(module_name, return_package=True)) as f:
head = f.read(1024)
if "PYTHON_ARGCOMPLETE_OK" not in head:
raise ArgcompleteMarkerNotFound("marker not found")
if __name__ == "__main__":
try:
main()
except ArgcompleteMarkerNotFound as e:
sys.exit(str(e))

View File

@@ -0,0 +1,89 @@
"""
Utility for locating a module (or package's __main__.py) with a given name
and verifying it contains the PYTHON_ARGCOMPLETE_OK marker.
The module name should be specified in a form usable with `python -m`.
Intended to be invoked by argcomplete's global completion function.
"""
import os
import sys
import tokenize
try:
from importlib.util import find_spec # type:ignore
except ImportError:
import typing as t
from collections import namedtuple
from imp import find_module # type:ignore
ModuleSpec = namedtuple("ModuleSpec", ["origin", "has_location", "submodule_search_locations"])
def find_spec( # type:ignore
name: str,
package: t.Optional[str] = None,
) -> t.Optional[ModuleSpec]:
"""Minimal implementation as required by `find`."""
try:
f, path, _ = find_module(name)
except ImportError:
return None
has_location = path is not None
if f is None:
return ModuleSpec(None, has_location, [path])
f.close()
return ModuleSpec(path, has_location, None)
class ArgcompleteMarkerNotFound(RuntimeError):
pass
def find(name, return_package=False):
names = name.split(".")
spec = find_spec(names[0])
if spec is None:
raise ArgcompleteMarkerNotFound('no module named "{}"'.format(names[0]))
if not spec.has_location:
raise ArgcompleteMarkerNotFound("cannot locate file")
if spec.submodule_search_locations is None:
if len(names) != 1:
raise ArgcompleteMarkerNotFound("{} is not a package".format(names[0]))
return spec.origin
if len(spec.submodule_search_locations) != 1:
raise ArgcompleteMarkerNotFound("expecting one search location")
path = os.path.join(spec.submodule_search_locations[0], *names[1:])
if os.path.isdir(path):
filename = "__main__.py"
if return_package:
filename = "__init__.py"
return os.path.join(path, filename)
else:
return path + ".py"
def main():
try:
name = sys.argv[1]
except IndexError:
raise ArgcompleteMarkerNotFound("missing argument on the command line")
filename = find(name)
try:
fp = tokenize.open(filename)
except OSError:
raise ArgcompleteMarkerNotFound("cannot open file")
with fp:
head = fp.read(1024)
if "PYTHON_ARGCOMPLETE_OK" not in head:
raise ArgcompleteMarkerNotFound("marker not found")
if __name__ == "__main__":
try:
main()
except ArgcompleteMarkerNotFound as e:
sys.exit(str(e))

View File

@@ -0,0 +1,261 @@
#compdef -default-
# Copyright 2012-2023, Andrey Kislyuk and argcomplete contributors.
# Licensed under the Apache License. See https://github.com/kislyuk/argcomplete for more info.
# Note: both the leading underscore in the name of this file and the first line (compdef) are required by zsh
# In zsh, this file is autoloaded and used as the default completer (_default).
# There are many other special contexts we don't want to override
# (as would be the case with `#compdef -P *`).
# https://zsh.sourceforge.io/Doc/Release/Completion-System.html
# Copy of __expand_tilde_by_ref from bash-completion
# ZSH implementation added
__python_argcomplete_expand_tilde_by_ref () {
if [ -n "${ZSH_VERSION-}" ]; then
if [ "${(P)1[1]}" = "~" ]; then
eval $1="${(P)1/#\~/$HOME}";
fi
else
if [ "${!1:0:1}" = "~" ]; then
if [ "${!1}" != "${!1//\/}" ]; then
eval $1="${!1/%\/*}"/'${!1#*/}';
else
eval $1="${!1}";
fi;
fi
fi
}
# Run something, muting output or redirecting it to the debug stream
# depending on the value of _ARC_DEBUG.
# If ARGCOMPLETE_USE_TEMPFILES is set, use tempfiles for IPC.
__python_argcomplete_run() {
if [[ -z "${ARGCOMPLETE_USE_TEMPFILES-}" ]]; then
__python_argcomplete_run_inner "$@"
return
fi
local tmpfile="$(mktemp)"
_ARGCOMPLETE_STDOUT_FILENAME="$tmpfile" __python_argcomplete_run_inner "$@"
local code=$?
cat "$tmpfile"
rm "$tmpfile"
return $code
}
__python_argcomplete_run_inner() {
if [[ -z "${_ARC_DEBUG-}" ]]; then
"$@" 8>&1 9>&2 1>/dev/null 2>&1
else
"$@" 8>&1 9>&2 1>&9 2>&1
fi
}
__python_argcomplete_upshift_bash_rematch() {
if [[ -z "${ZSH_VERSION-}" ]]; then
_BASH_REMATCH=( "" "${BASH_REMATCH[@]}" )
else
_BASH_REMATCH=( "${BASH_REMATCH[@]}" )
fi
}
# This function scans the beginning of an executable file provided as the first
# argument ($1) for certain indicators, specified by the second argument ($2),
# or the "target". There are three possible targets: "interpreter",
# "magic_string", and "easy_install". If the target is "interpreter", the
# function matches the interpreter line, alongside any optional interpreter
# arguments. If the target is "magic_string", a match is attempted for the
# "PYTHON_ARGCOMPLETE_OK" magic string, indicating that the file should be run
# to get completions. If the target is "easy_install", the function matches either
# "PBR Generated" or any of the "EASY-INSTALL" scripts (either SCRIPT,
# ENTRY-SCRIPT, or DEV-SCRIPT). In all cases, only the first kilobyte of
# the file is searched. The regex matches are returned in BASH_REMATCH,
# indexed starting at 1, regardless of the shell in use.
__python_argcomplete_scan_head() {
local file="$1"
local target="$2"
if [[ -n "${ZSH_VERSION-}" ]]; then
read -r -k 1024 -u 0 < "$file";
else
read -r -N 1024 < "$file"
fi
# Since ZSH does not support a -n option, we
# trim all characters after the first line in both shells
if [[ "$target" = "interpreter" ]]; then
read -r <<< "$REPLY"
fi
local regex
case "$target" in
magic_string) regex='PYTHON_ARGCOMPLETE_OK' ;;
easy_install) regex="(PBR Generated)|(EASY-INSTALL-(SCRIPT|ENTRY-SCRIPT|DEV-SCRIPT))" ;;
asdf) regex="asdf exec " ;;
interpreter) regex='^#!(.*)$' ;;
esac
local ret=""
if [[ "$REPLY" =~ $regex ]]; then
ret=1
fi
__python_argcomplete_upshift_bash_rematch
[[ -n $ret ]]
}
__python_argcomplete_scan_head_noerr() {
__python_argcomplete_scan_head "$@" 2>/dev/null
}
__python_argcomplete_which() {
if [[ -n "${ZSH_VERSION-}" ]]; then
whence -p "$@"
else
type -P "$@"
fi
}
_python_argcomplete_global() {
if [[ -n "${ZSH_VERSION-}" ]]; then
# Store result of a regex match in the
# BASH_REMATCH variable rather than MATCH
setopt local_options BASH_REMATCH
fi
# 1-based version of BASH_REMATCH. Modifying BASH_REMATCH
# directly causes older versions of Bash to exit
local _BASH_REMATCH="";
local executable=""
# req_argv contains the arguments to the completion
# indexed from 1 (regardless of the shell.) In Bash,
# the zeroth index is empty
local req_argv=()
if [[ -z "${ZSH_VERSION-}" ]]; then
executable=$1
req_argv=( "" "${COMP_WORDS[@]:1}" )
__python_argcomplete_expand_tilde_by_ref executable
else
executable="${words[1]}"
__python_argcomplete_expand_tilde_by_ref executable
req_argv=( "${words[@]:1}" )
fi
local ARGCOMPLETE=0
if [[ "$executable" == python* ]] || [[ "$executable" == pypy* ]]; then
if [[ "${req_argv[1]}" == -m ]]; then
if __python_argcomplete_run "$executable" -m argcomplete._check_module "${req_argv[2]}"; then
ARGCOMPLETE=3
else
return
fi
fi
if [[ $ARGCOMPLETE == 0 ]]; then
local potential_path="${req_argv[1]}"
__python_argcomplete_expand_tilde_by_ref potential_path
if [[ -f "$potential_path" ]] && __python_argcomplete_scan_head_noerr "$potential_path" magic_string; then
req_argv[1]="$potential_path" # not expanded in __python_argcomplete_run
ARGCOMPLETE=2
else
return
fi
fi
elif __python_argcomplete_which "$executable" >/dev/null 2>&1; then
local SCRIPT_NAME=$(__python_argcomplete_which "$executable")
__python_argcomplete_scan_head_noerr "$SCRIPT_NAME" interpreter
if (__python_argcomplete_which pyenv && [[ "$SCRIPT_NAME" = $(pyenv root)/shims/* ]]) >/dev/null 2>&1; then
local SCRIPT_NAME=$(pyenv which "$executable")
fi
if (__python_argcomplete_which asdf && __python_argcomplete_scan_head_noerr "$SCRIPT_NAME" asdf) >/dev/null 2>&1; then
local SCRIPT_NAME=$(asdf which "$executable")
fi
if __python_argcomplete_scan_head_noerr "$SCRIPT_NAME" magic_string; then
ARGCOMPLETE=1
elif __python_argcomplete_scan_head_noerr "$SCRIPT_NAME" interpreter; then
__python_argcomplete_upshift_bash_rematch
local interpreter="${_BASH_REMATCH[2]}"
if [[ -n "${ZSH_VERSION-}" ]]; then
interpreter=($=interpreter)
else
interpreter=($interpreter)
fi
if (__python_argcomplete_scan_head_noerr "$SCRIPT_NAME" easy_install \
&& "${interpreter[@]}" "$(__python_argcomplete_which python-argcomplete-check-easy-install-script)" "$SCRIPT_NAME") >/dev/null 2>&1; then
ARGCOMPLETE=1
elif __python_argcomplete_run "${interpreter[@]}" -m argcomplete._check_console_script "$SCRIPT_NAME"; then
ARGCOMPLETE=1
fi
fi
fi
if [[ $ARGCOMPLETE != 0 ]]; then
local IFS=$'\013'
if [[ -n "${ZSH_VERSION-}" ]]; then
local completions
completions=($(IFS="$IFS" \
COMP_LINE="$BUFFER" \
COMP_POINT="$CURSOR" \
_ARGCOMPLETE=$ARGCOMPLETE \
_ARGCOMPLETE_SHELL="zsh" \
_ARGCOMPLETE_SUPPRESS_SPACE=1 \
__python_argcomplete_run "$executable" "${(@)req_argv[1, ${ARGCOMPLETE}-1]}"))
local nosort=()
local nospace=()
if is-at-least 5.8; then
nosort=(-o nosort)
fi
if [[ "${completions-}" =~ ([^\\]): && "${BASH_REMATCH[2]}" =~ [=/:] ]]; then
nospace=(-S '')
fi
_describe "$executable" completions "${nosort[@]}" "${nospace[@]}"
else
COMPREPLY=($(IFS="$IFS" \
COMP_LINE="$COMP_LINE" \
COMP_POINT="$COMP_POINT" \
COMP_TYPE="$COMP_TYPE" \
_ARGCOMPLETE_COMP_WORDBREAKS="$COMP_WORDBREAKS" \
_ARGCOMPLETE=$ARGCOMPLETE \
_ARGCOMPLETE_SHELL="bash" \
_ARGCOMPLETE_SUPPRESS_SPACE=1 \
__python_argcomplete_run "$executable" "${req_argv[@]:1:${ARGCOMPLETE}-1}"))
if [[ $? != 0 ]]; then
unset COMPREPLY
elif [[ "${COMPREPLY-}" =~ [=/:]$ ]]; then
compopt -o nospace
fi
fi
elif [[ -n "${ZSH_VERSION-}" ]]; then
_default
else
type -t _completion_loader | grep -q 'function' && _completion_loader "$@"
fi
}
if [[ -z "${ZSH_VERSION-}" ]]; then
complete -o default -o bashdefault -D -F _python_argcomplete_global
else
# -Uz is recommended for the use of functions supplied with the zsh distribution.
# https://unix.stackexchange.com/a/214306
autoload -Uz is-at-least
# If this is being implicitly loaded because we placed it on fpath,
# the comment at the top of this file causes zsh to invoke this script directly,
# so we must explicitly call the global completion function.
# Note $service should only ever be -default- because the comment at the top
# registers this script as the default completer (#compdef -default-).
if [[ $service == -default- ]]; then
_python_argcomplete_global
fi
# If this has been executed directly (e.g. `eval "$(activate-global-python-argcomplete --dest=-)"`)
# we need to explicitly call compdef to register the completion function.
# If we have been implicitly loaded, we still call compdef as a slight optimisation
# (there is no need to execute any top-level code more than once).
compdef _python_argcomplete_global -default-
fi

View File

@@ -0,0 +1,124 @@
# Copyright 2012-2023, Andrey Kislyuk and argcomplete contributors.
# Licensed under the Apache License. See https://github.com/kislyuk/argcomplete for more info.
import argparse
import os
import subprocess
def _call(*args, **kwargs):
# TODO: replace "universal_newlines" with "text" once 3.6 support is dropped
kwargs["universal_newlines"] = True
try:
return subprocess.check_output(*args, **kwargs).splitlines()
except subprocess.CalledProcessError:
return []
class BaseCompleter:
"""
This is the base class that all argcomplete completers should subclass.
"""
def __call__(
self, *, prefix: str, action: argparse.Action, parser: argparse.ArgumentParser, parsed_args: argparse.Namespace
):
raise NotImplementedError("This method should be implemented by a subclass.")
class ChoicesCompleter(BaseCompleter):
def __init__(self, choices):
self.choices = choices
def _convert(self, choice):
if not isinstance(choice, str):
choice = str(choice)
return choice
def __call__(self, **kwargs):
return (self._convert(c) for c in self.choices)
EnvironCompleter = ChoicesCompleter(os.environ)
class FilesCompleter(BaseCompleter):
"""
File completer class, optionally takes a list of allowed extensions
"""
def __init__(self, allowednames=(), directories=True):
# Fix if someone passes in a string instead of a list
if isinstance(allowednames, (str, bytes)):
allowednames = [allowednames]
self.allowednames = [x.lstrip("*").lstrip(".") for x in allowednames]
self.directories = directories
def __call__(self, prefix, **kwargs):
completion = []
if self.allowednames:
if self.directories:
files = _call(["bash", "-c", "compgen -A directory -- '{p}'".format(p=prefix)])
completion += [f + "/" for f in files]
for x in self.allowednames:
completion += _call(["bash", "-c", "compgen -A file -X '!*.{0}' -- '{p}'".format(x, p=prefix)])
else:
completion += _call(["bash", "-c", "compgen -A file -- '{p}'".format(p=prefix)])
anticomp = _call(["bash", "-c", "compgen -A directory -- '{p}'".format(p=prefix)])
completion = list(set(completion) - set(anticomp))
if self.directories:
completion += [f + "/" for f in anticomp]
return completion
class _FilteredFilesCompleter(BaseCompleter):
def __init__(self, predicate):
"""
Create the completer
A predicate accepts as its only argument a candidate path and either
accepts it or rejects it.
"""
assert predicate, "Expected a callable predicate"
self.predicate = predicate
def __call__(self, prefix, **kwargs):
"""
Provide completions on prefix
"""
target_dir = os.path.dirname(prefix)
try:
names = os.listdir(target_dir or ".")
except Exception:
return # empty iterator
incomplete_part = os.path.basename(prefix)
# Iterate on target_dir entries and filter on given predicate
for name in names:
if not name.startswith(incomplete_part):
continue
candidate = os.path.join(target_dir, name)
if not self.predicate(candidate):
continue
yield candidate + "/" if os.path.isdir(candidate) else candidate
class DirectoriesCompleter(_FilteredFilesCompleter):
def __init__(self):
_FilteredFilesCompleter.__init__(self, predicate=os.path.isdir)
class SuppressCompleter(BaseCompleter):
"""
A completer used to suppress the completion of specific arguments
"""
def __init__(self):
pass
def suppress(self):
"""
Decide if the completion should be suppressed
"""
return True

View File

@@ -0,0 +1,2 @@
class ArgcompleteException(Exception):
"Exception raised when the shell argument completion process fails."

View File

@@ -0,0 +1,600 @@
# Copyright 2012-2023, Andrey Kislyuk and argcomplete contributors. Licensed under the terms of the
# `Apache License, Version 2.0 <http://www.apache.org/licenses/LICENSE-2.0>`_. Distribution of the LICENSE and NOTICE
# files with source copies of this package and derivative works is **REQUIRED** as specified by the Apache License.
# See https://github.com/kislyuk/argcomplete for more info.
import argparse
import os
import sys
from collections.abc import Mapping
from typing import Callable, Dict, List, Optional, Sequence, Union
from . import io as _io
from .completers import ChoicesCompleter, FilesCompleter, SuppressCompleter
from .io import debug, mute_stderr
from .lexers import split_line
from .packages._argparse import IntrospectiveArgumentParser, action_is_greedy, action_is_open, action_is_satisfied
safe_actions = {
argparse._StoreAction,
argparse._StoreConstAction,
argparse._StoreTrueAction,
argparse._StoreFalseAction,
argparse._AppendAction,
argparse._AppendConstAction,
argparse._CountAction,
}
def default_validator(completion, prefix):
return completion.startswith(prefix)
class CompletionFinder(object):
"""
Inherit from this class if you wish to override any of the stages below. Otherwise, use
``argcomplete.autocomplete()`` directly (it's a convenience instance of this class). It has the same signature as
:meth:`CompletionFinder.__call__()`.
"""
def __init__(
self,
argument_parser=None,
always_complete_options=True,
exclude=None,
validator=None,
print_suppressed=False,
default_completer=FilesCompleter(),
append_space=None,
):
self._parser = argument_parser
self.always_complete_options = always_complete_options
self.exclude = exclude
if validator is None:
validator = default_validator
self.validator = validator
self.print_suppressed = print_suppressed
self.completing = False
self._display_completions: Dict[str, str] = {}
self.default_completer = default_completer
if append_space is None:
append_space = os.environ.get("_ARGCOMPLETE_SUPPRESS_SPACE") != "1"
self.append_space = append_space
def __call__(
self,
argument_parser: argparse.ArgumentParser,
always_complete_options: Union[bool, str] = True,
exit_method: Callable = os._exit,
output_stream=None,
exclude: Optional[Sequence[str]] = None,
validator: Optional[Callable] = None,
print_suppressed: bool = False,
append_space: Optional[bool] = None,
default_completer=FilesCompleter(),
):
"""
:param argument_parser: The argument parser to autocomplete on
:param always_complete_options:
Controls the autocompletion of option strings if an option string opening character (normally ``-``) has not
been entered. If ``True`` (default), both short (``-x``) and long (``--x``) option strings will be
suggested. If ``False``, no option strings will be suggested. If ``long``, long options and short options
with no long variant will be suggested. If ``short``, short options and long options with no short variant
will be suggested.
:param exit_method:
Method used to stop the program after printing completions. Defaults to :meth:`os._exit`. If you want to
perform a normal exit that calls exit handlers, use :meth:`sys.exit`.
:param exclude: List of strings representing options to be omitted from autocompletion
:param validator:
Function to filter all completions through before returning (called with two string arguments, completion
and prefix; return value is evaluated as a boolean)
:param print_suppressed:
Whether or not to autocomplete options that have the ``help=argparse.SUPPRESS`` keyword argument set.
:param append_space:
Whether to append a space to unique matches. The default is ``True``.
.. note::
If you are not subclassing CompletionFinder to override its behaviors,
use :meth:`argcomplete.autocomplete()` directly. It has the same signature as this method.
Produces tab completions for ``argument_parser``. See module docs for more info.
Argcomplete only executes actions if their class is known not to have side effects. Custom action classes can be
added to argcomplete.safe_actions, if their values are wanted in the ``parsed_args`` completer argument, or
their execution is otherwise desirable.
"""
self.__init__( # type: ignore
argument_parser,
always_complete_options=always_complete_options,
exclude=exclude,
validator=validator,
print_suppressed=print_suppressed,
append_space=append_space,
default_completer=default_completer,
)
if "_ARGCOMPLETE" not in os.environ:
# not an argument completion invocation
return
try:
_io.debug_stream = os.fdopen(9, "w")
except Exception:
_io.debug_stream = sys.stderr
debug()
if output_stream is None:
filename = os.environ.get("_ARGCOMPLETE_STDOUT_FILENAME")
if filename is not None:
debug("Using output file {}".format(filename))
output_stream = open(filename, "w")
if output_stream is None:
try:
output_stream = os.fdopen(8, "w")
except Exception:
debug("Unable to open fd 8 for writing, quitting")
exit_method(1)
ifs = os.environ.get("_ARGCOMPLETE_IFS", "\013")
if len(ifs) != 1:
debug("Invalid value for IFS, quitting [{v}]".format(v=ifs))
exit_method(1)
dfs = os.environ.get("_ARGCOMPLETE_DFS")
if dfs and len(dfs) != 1:
debug("Invalid value for DFS, quitting [{v}]".format(v=dfs))
exit_method(1)
comp_line = os.environ["COMP_LINE"]
comp_point = int(os.environ["COMP_POINT"])
cword_prequote, cword_prefix, cword_suffix, comp_words, last_wordbreak_pos = split_line(comp_line, comp_point)
# _ARGCOMPLETE is set by the shell script to tell us where comp_words
# should start, based on what we're completing.
# 1: <script> [args]
# 2: python <script> [args]
# 3: python -m <module> [args]
start = int(os.environ["_ARGCOMPLETE"]) - 1
comp_words = comp_words[start:]
if cword_prefix and cword_prefix[0] in self._parser.prefix_chars and "=" in cword_prefix:
# Special case for when the current word is "--optional=PARTIAL_VALUE". Give the optional to the parser.
comp_words.append(cword_prefix.split("=", 1)[0])
debug(
"\nLINE: {!r}".format(comp_line),
"\nPOINT: {!r}".format(comp_point),
"\nPREQUOTE: {!r}".format(cword_prequote),
"\nPREFIX: {!r}".format(cword_prefix),
"\nSUFFIX: {!r}".format(cword_suffix),
"\nWORDS:",
comp_words,
)
completions = self._get_completions(comp_words, cword_prefix, cword_prequote, last_wordbreak_pos)
if dfs:
display_completions = {
key: value.replace(ifs, " ") if value else "" for key, value in self._display_completions.items()
}
completions = [dfs.join((key, display_completions.get(key) or "")) for key in completions]
if os.environ.get("_ARGCOMPLETE_SHELL") == "zsh":
completions = [f"{c}:{self._display_completions.get(c)}" for c in completions]
debug("\nReturning completions:", completions)
output_stream.write(ifs.join(completions))
output_stream.flush()
_io.debug_stream.flush()
exit_method(0)
def _get_completions(self, comp_words, cword_prefix, cword_prequote, last_wordbreak_pos):
active_parsers = self._patch_argument_parser()
parsed_args = argparse.Namespace()
self.completing = True
try:
debug("invoking parser with", comp_words[1:])
with mute_stderr():
a = self._parser.parse_known_args(comp_words[1:], namespace=parsed_args)
debug("parsed args:", a)
except BaseException as e:
debug("\nexception", type(e), str(e), "while parsing args")
self.completing = False
if "--" in comp_words:
self.always_complete_options = False
completions = self.collect_completions(active_parsers, parsed_args, cword_prefix)
completions = self.filter_completions(completions)
completions = self.quote_completions(completions, cword_prequote, last_wordbreak_pos)
return completions
def _patch_argument_parser(self):
"""
Since argparse doesn't support much introspection, we monkey-patch it to replace the parse_known_args method and
all actions with hooks that tell us which action was last taken or about to be taken, and let us have the parser
figure out which subparsers need to be activated (then recursively monkey-patch those).
We save all active ArgumentParsers to extract all their possible option names later.
"""
self.active_parsers: List[argparse.ArgumentParser] = []
self.visited_positionals: List[argparse.Action] = []
completer = self
def patch(parser):
completer.visited_positionals.append(parser)
completer.active_parsers.append(parser)
if isinstance(parser, IntrospectiveArgumentParser):
return
classname = "MonkeyPatchedIntrospectiveArgumentParser"
parser.__class__ = type(classname, (IntrospectiveArgumentParser, parser.__class__), {})
for action in parser._actions:
if hasattr(action, "_orig_class"):
continue
# TODO: accomplish this with super
class IntrospectAction(action.__class__): # type: ignore
def __call__(self, parser, namespace, values, option_string=None):
debug("Action stub called on", self)
debug("\targs:", parser, namespace, values, option_string)
debug("\torig class:", self._orig_class)
debug("\torig callable:", self._orig_callable)
if not completer.completing:
self._orig_callable(parser, namespace, values, option_string=option_string)
elif issubclass(self._orig_class, argparse._SubParsersAction):
debug("orig class is a subparsers action: patching and running it")
patch(self._name_parser_map[values[0]])
self._orig_callable(parser, namespace, values, option_string=option_string)
elif self._orig_class in safe_actions:
if not self.option_strings:
completer.visited_positionals.append(self)
self._orig_callable(parser, namespace, values, option_string=option_string)
action._orig_class = action.__class__
action._orig_callable = action.__call__
action.__class__ = IntrospectAction
patch(self._parser)
debug("Active parsers:", self.active_parsers)
debug("Visited positionals:", self.visited_positionals)
return self.active_parsers
def _get_subparser_completions(self, parser, cword_prefix):
aliases_by_parser: Dict[argparse.ArgumentParser, List[str]] = {}
for key in parser.choices.keys():
p = parser.choices[key]
aliases_by_parser.setdefault(p, []).append(key)
for action in parser._get_subactions():
for alias in aliases_by_parser[parser.choices[action.dest]]:
if alias.startswith(cword_prefix):
self._display_completions[alias] = action.help or ""
completions = [subcmd for subcmd in parser.choices.keys() if subcmd.startswith(cword_prefix)]
return completions
def _include_options(self, action, cword_prefix):
if len(cword_prefix) > 0 or self.always_complete_options is True:
return [opt for opt in action.option_strings if opt.startswith(cword_prefix)]
long_opts = [opt for opt in action.option_strings if len(opt) > 2]
short_opts = [opt for opt in action.option_strings if len(opt) <= 2]
if self.always_complete_options == "long":
return long_opts if long_opts else short_opts
elif self.always_complete_options == "short":
return short_opts if short_opts else long_opts
return []
def _get_option_completions(self, parser, cword_prefix):
for action in parser._actions:
if action.option_strings:
for option_string in action.option_strings:
if option_string.startswith(cword_prefix):
self._display_completions[option_string] = action.help or ""
option_completions = []
for action in parser._actions:
if not self.print_suppressed:
completer = getattr(action, "completer", None)
if isinstance(completer, SuppressCompleter) and completer.suppress():
continue
if action.help == argparse.SUPPRESS:
continue
if not self._action_allowed(action, parser):
continue
if not isinstance(action, argparse._SubParsersAction):
option_completions += self._include_options(action, cword_prefix)
return option_completions
@staticmethod
def _action_allowed(action, parser):
# Logic adapted from take_action in ArgumentParser._parse_known_args
# (members are saved by vendor._argparse.IntrospectiveArgumentParser)
for conflict_action in parser._action_conflicts.get(action, []):
if conflict_action in parser._seen_non_default_actions:
return False
return True
def _complete_active_option(self, parser, next_positional, cword_prefix, parsed_args, completions):
debug("Active actions (L={l}): {a}".format(l=len(parser.active_actions), a=parser.active_actions))
isoptional = cword_prefix and cword_prefix[0] in parser.prefix_chars
optional_prefix = ""
greedy_actions = [x for x in parser.active_actions if action_is_greedy(x, isoptional)]
if greedy_actions:
assert len(greedy_actions) == 1, "expect at most 1 greedy action"
# This means the action will fail to parse if the word under the cursor is not given
# to it, so give it exclusive control over completions (flush previous completions)
debug("Resetting completions because", greedy_actions[0], "must consume the next argument")
self._display_completions = {}
completions = []
elif isoptional:
if "=" in cword_prefix:
# Special case for when the current word is "--optional=PARTIAL_VALUE".
# The completer runs on PARTIAL_VALUE. The prefix is added back to the completions
# (and chopped back off later in quote_completions() by the COMP_WORDBREAKS logic).
optional_prefix, _, cword_prefix = cword_prefix.partition("=")
else:
# Only run completers if current word does not start with - (is not an optional)
return completions
complete_remaining_positionals = False
# Use the single greedy action (if there is one) or all active actions.
for active_action in greedy_actions or parser.active_actions:
if not active_action.option_strings: # action is a positional
if action_is_open(active_action):
# Any positional arguments after this may slide down into this action
# if more arguments are added (since the user may not be done yet),
# so it is extremely difficult to tell which completers to run.
# Running all remaining completers will probably show more than the user wants
# but it also guarantees we won't miss anything.
complete_remaining_positionals = True
if not complete_remaining_positionals:
if action_is_satisfied(active_action) and not action_is_open(active_action):
debug("Skipping", active_action)
continue
debug("Activating completion for", active_action, active_action._orig_class)
# completer = getattr(active_action, "completer", DefaultCompleter())
completer = getattr(active_action, "completer", None)
if completer is None:
if active_action.choices is not None and not isinstance(active_action, argparse._SubParsersAction):
completer = ChoicesCompleter(active_action.choices)
elif not isinstance(active_action, argparse._SubParsersAction):
completer = self.default_completer
if completer:
if isinstance(completer, SuppressCompleter) and completer.suppress():
continue
if callable(completer):
completer_output = completer(
prefix=cword_prefix, action=active_action, parser=parser, parsed_args=parsed_args
)
if isinstance(completer_output, Mapping):
for completion, description in completer_output.items():
if self.validator(completion, cword_prefix):
completions.append(completion)
self._display_completions[completion] = description
else:
for completion in completer_output:
if self.validator(completion, cword_prefix):
completions.append(completion)
if isinstance(completer, ChoicesCompleter):
self._display_completions[completion] = active_action.help or ""
else:
self._display_completions[completion] = ""
else:
debug("Completer is not callable, trying the readline completer protocol instead")
for i in range(9999):
next_completion = completer.complete(cword_prefix, i) # type: ignore
if next_completion is None:
break
if self.validator(next_completion, cword_prefix):
self._display_completions[next_completion] = ""
completions.append(next_completion)
if optional_prefix:
completions = [optional_prefix + "=" + completion for completion in completions]
debug("Completions:", completions)
return completions
def collect_completions(
self, active_parsers: List[argparse.ArgumentParser], parsed_args: argparse.Namespace, cword_prefix: str
) -> List[str]:
"""
Visits the active parsers and their actions, executes their completers or introspects them to collect their
option strings. Returns the resulting completions as a list of strings.
This method is exposed for overriding in subclasses; there is no need to use it directly.
"""
completions: List[str] = []
debug("all active parsers:", active_parsers)
active_parser = active_parsers[-1]
debug("active_parser:", active_parser)
if self.always_complete_options or (len(cword_prefix) > 0 and cword_prefix[0] in active_parser.prefix_chars):
completions += self._get_option_completions(active_parser, cword_prefix)
debug("optional options:", completions)
next_positional = self._get_next_positional()
debug("next_positional:", next_positional)
if isinstance(next_positional, argparse._SubParsersAction):
completions += self._get_subparser_completions(next_positional, cword_prefix)
completions = self._complete_active_option(
active_parser, next_positional, cword_prefix, parsed_args, completions
)
debug("active options:", completions)
debug("display completions:", self._display_completions)
return completions
def _get_next_positional(self):
"""
Get the next positional action if it exists.
"""
active_parser = self.active_parsers[-1]
last_positional = self.visited_positionals[-1]
all_positionals = active_parser._get_positional_actions()
if not all_positionals:
return None
if active_parser == last_positional:
return all_positionals[0]
i = 0
for i in range(len(all_positionals)):
if all_positionals[i] == last_positional:
break
if i + 1 < len(all_positionals):
return all_positionals[i + 1]
return None
def filter_completions(self, completions: List[str]) -> List[str]:
"""
De-duplicates completions and excludes those specified by ``exclude``.
Returns the filtered completions as a list.
This method is exposed for overriding in subclasses; there is no need to use it directly.
"""
filtered_completions = []
for completion in completions:
if self.exclude is not None:
if completion in self.exclude:
continue
if completion not in filtered_completions:
filtered_completions.append(completion)
return filtered_completions
def quote_completions(
self, completions: List[str], cword_prequote: str, last_wordbreak_pos: Optional[int]
) -> List[str]:
"""
If the word under the cursor started with a quote (as indicated by a nonempty ``cword_prequote``), escapes
occurrences of that quote character in the completions, and adds the quote to the beginning of each completion.
Otherwise, escapes all characters that bash splits words on (``COMP_WORDBREAKS``), and removes portions of
completions before the first colon if (``COMP_WORDBREAKS``) contains a colon.
If there is only one completion, and it doesn't end with a **continuation character** (``/``, ``:``, or ``=``),
adds a space after the completion.
This method is exposed for overriding in subclasses; there is no need to use it directly.
"""
special_chars = "\\"
# If the word under the cursor was quoted, escape the quote char.
# Otherwise, escape all special characters and specially handle all COMP_WORDBREAKS chars.
if cword_prequote == "":
# Bash mangles completions which contain characters in COMP_WORDBREAKS.
# This workaround has the same effect as __ltrim_colon_completions in bash_completion
# (extended to characters other than the colon).
if last_wordbreak_pos:
completions = [c[last_wordbreak_pos + 1 :] for c in completions]
special_chars += "();<>|&!`$* \t\n\"'"
elif cword_prequote == '"':
special_chars += '"`$!'
if os.environ.get("_ARGCOMPLETE_SHELL") in ("tcsh", "fish"):
# tcsh and fish escapes special characters itself.
special_chars = ""
elif cword_prequote == "'":
# Nothing can be escaped in single quotes, so we need to close
# the string, escape the single quote, then open a new string.
special_chars = ""
completions = [c.replace("'", r"'\''") for c in completions]
# PowerShell uses ` as escape character.
if os.environ.get("_ARGCOMPLETE_SHELL") == "powershell":
escape_char = '`'
special_chars = special_chars.replace('`', '')
else:
escape_char = "\\"
if os.environ.get("_ARGCOMPLETE_SHELL") == "zsh":
# zsh uses colon as a separator between a completion and its description.
special_chars += ":"
escaped_completions = []
for completion in completions:
escaped_completion = completion
for char in special_chars:
escaped_completion = escaped_completion.replace(char, escape_char + char)
escaped_completions.append(escaped_completion)
if completion in self._display_completions:
self._display_completions[escaped_completion] = self._display_completions[completion]
if self.append_space:
# Similar functionality in bash was previously turned off by supplying the "-o nospace" option to complete.
# Now it is conditionally disabled using "compopt -o nospace" if the match ends in a continuation character.
# This code is retained for environments where this isn't done natively.
continuation_chars = "=/:"
if len(escaped_completions) == 1 and escaped_completions[0][-1] not in continuation_chars:
if cword_prequote == "":
escaped_completions[0] += " "
return escaped_completions
def rl_complete(self, text, state):
"""
Alternate entry point for using the argcomplete completer in a readline-based REPL. See also
`rlcompleter <https://docs.python.org/3/library/rlcompleter.html#completer-objects>`_.
Usage:
.. code-block:: python
import argcomplete, argparse, readline
parser = argparse.ArgumentParser()
...
completer = argcomplete.CompletionFinder(parser)
readline.set_completer_delims("")
readline.set_completer(completer.rl_complete)
readline.parse_and_bind("tab: complete")
result = input("prompt> ")
"""
if state == 0:
cword_prequote, cword_prefix, cword_suffix, comp_words, first_colon_pos = split_line(text)
comp_words.insert(0, sys.argv[0])
matches = self._get_completions(comp_words, cword_prefix, cword_prequote, first_colon_pos)
self._rl_matches = [text + match[len(cword_prefix) :] for match in matches]
if state < len(self._rl_matches):
return self._rl_matches[state]
else:
return None
def get_display_completions(self):
"""
This function returns a mapping of completions to their help strings for displaying to the user.
"""
return self._display_completions
class ExclusiveCompletionFinder(CompletionFinder):
@staticmethod
def _action_allowed(action, parser):
if not CompletionFinder._action_allowed(action, parser):
return False
append_classes = (argparse._AppendAction, argparse._AppendConstAction)
if action._orig_class in append_classes:
return True
if action not in parser._seen_non_default_actions:
return True
return False

View File

@@ -0,0 +1,42 @@
import contextlib
import os
import sys
_DEBUG = "_ARC_DEBUG" in os.environ
debug_stream = sys.stderr
def debug(*args):
if _DEBUG:
print(file=debug_stream, *args)
@contextlib.contextmanager
def mute_stdout():
stdout = sys.stdout
sys.stdout = open(os.devnull, "w")
try:
yield
finally:
sys.stdout = stdout
@contextlib.contextmanager
def mute_stderr():
stderr = sys.stderr
sys.stderr = open(os.devnull, "w")
try:
yield
finally:
sys.stderr.close()
sys.stderr = stderr
def warn(*args):
"""
Prints **args** to standard error when running completions. This will interrupt the user's command line interaction;
use it to indicate an error condition that is preventing your completer from working.
"""
print(file=debug_stream)
print(file=debug_stream, *args)

View File

@@ -0,0 +1,57 @@
import os
from .exceptions import ArgcompleteException
from .io import debug
from .packages import _shlex
def split_line(line, point=None):
if point is None:
point = len(line)
line = line[:point]
lexer = _shlex.shlex(line, posix=True)
lexer.whitespace_split = True
lexer.wordbreaks = os.environ.get("_ARGCOMPLETE_COMP_WORDBREAKS", "")
words = []
def split_word(word):
# TODO: make this less ugly
point_in_word = len(word) + point - lexer.instream.tell()
if isinstance(lexer.state, (str, bytes)) and lexer.state in lexer.whitespace:
point_in_word += 1
if point_in_word > len(word):
debug("In trailing whitespace")
words.append(word)
word = ""
prefix, suffix = word[:point_in_word], word[point_in_word:]
prequote = ""
# posix
if lexer.state is not None and lexer.state in lexer.quotes:
prequote = lexer.state
# non-posix
# if len(prefix) > 0 and prefix[0] in lexer.quotes:
# prequote, prefix = prefix[0], prefix[1:]
return prequote, prefix, suffix, words, lexer.last_wordbreak_pos
while True:
try:
word = lexer.get_token()
if word == lexer.eof:
# TODO: check if this is ever unsafe
# raise ArgcompleteException("Unexpected end of input")
return "", "", "", words, None
if lexer.instream.tell() >= point:
debug("word", word, "split, lexer state: '{s}'".format(s=lexer.state))
return split_word(word)
words.append(word)
except ValueError:
debug("word", lexer.token, "split (lexer stopped, state: '{s}')".format(s=lexer.state))
if lexer.instream.tell() >= point:
return split_word(lexer.token)
else:
msg = (
"Unexpected internal state. "
"Please report this bug at https://github.com/kislyuk/argcomplete/issues."
)
raise ArgcompleteException(msg)

View File

@@ -0,0 +1,337 @@
# Copyright 2012-2023, Andrey Kislyuk and argcomplete contributors. Licensed under the terms of the
# `Apache License, Version 2.0 <http://www.apache.org/licenses/LICENSE-2.0>`_. Distribution of the LICENSE and NOTICE
# files with source copies of this package and derivative works is **REQUIRED** as specified by the Apache License.
# See https://github.com/kislyuk/argcomplete for more info.
# This file contains argparse introspection utilities used in the course of argcomplete execution.
from argparse import (
ONE_OR_MORE,
OPTIONAL,
PARSER,
REMAINDER,
SUPPRESS,
ZERO_OR_MORE,
Action,
ArgumentError,
ArgumentParser,
_get_action_name,
_SubParsersAction,
)
from gettext import gettext
from typing import Dict, List, Set, Tuple
_num_consumed_args: Dict[Action, int] = {}
def action_is_satisfied(action):
'''Returns False if the parse would raise an error if no more arguments are given to this action, True otherwise.'''
num_consumed_args = _num_consumed_args.get(action, 0)
if action.nargs in [OPTIONAL, ZERO_OR_MORE, REMAINDER]:
return True
if action.nargs == ONE_OR_MORE:
return num_consumed_args >= 1
if action.nargs == PARSER:
# Not sure what this should be, but this previously always returned False
# so at least this won't break anything that wasn't already broken.
return False
if action.nargs is None:
return num_consumed_args == 1
assert isinstance(action.nargs, int), 'failed to handle a possible nargs value: %r' % action.nargs
return num_consumed_args == action.nargs
def action_is_open(action):
'''Returns True if action could consume more arguments (i.e., its pattern is open).'''
num_consumed_args = _num_consumed_args.get(action, 0)
if action.nargs in [ZERO_OR_MORE, ONE_OR_MORE, PARSER, REMAINDER]:
return True
if action.nargs == OPTIONAL or action.nargs is None:
return num_consumed_args == 0
assert isinstance(action.nargs, int), 'failed to handle a possible nargs value: %r' % action.nargs
return num_consumed_args < action.nargs
def action_is_greedy(action, isoptional=False):
'''Returns True if action will necessarily consume the next argument.
isoptional indicates whether the argument is an optional (starts with -).
'''
num_consumed_args = _num_consumed_args.get(action, 0)
if action.option_strings:
if not isoptional and not action_is_satisfied(action):
return True
return action.nargs == REMAINDER
else:
return action.nargs == REMAINDER and num_consumed_args >= 1
class IntrospectiveArgumentParser(ArgumentParser):
'''The following is a verbatim copy of ArgumentParser._parse_known_args (Python 2.7.3),
except for the lines that contain the string "Added by argcomplete".
'''
def _parse_known_args(self, arg_strings, namespace):
_num_consumed_args.clear() # Added by argcomplete
self._argcomplete_namespace = namespace
self.active_actions: List[Action] = [] # Added by argcomplete
# replace arg strings that are file references
if self.fromfile_prefix_chars is not None:
arg_strings = self._read_args_from_files(arg_strings)
# map all mutually exclusive arguments to the other arguments
# they can't occur with
action_conflicts: Dict[Action, List[Action]] = {}
self._action_conflicts = action_conflicts # Added by argcomplete
for mutex_group in self._mutually_exclusive_groups:
group_actions = mutex_group._group_actions
for i, mutex_action in enumerate(mutex_group._group_actions):
conflicts = action_conflicts.setdefault(mutex_action, [])
conflicts.extend(group_actions[:i])
conflicts.extend(group_actions[i + 1 :])
# find all option indices, and determine the arg_string_pattern
# which has an 'O' if there is an option at an index,
# an 'A' if there is an argument, or a '-' if there is a '--'
option_string_indices = {}
arg_string_pattern_parts = []
arg_strings_iter = iter(arg_strings)
for i, arg_string in enumerate(arg_strings_iter):
# all args after -- are non-options
if arg_string == '--':
arg_string_pattern_parts.append('-')
for arg_string in arg_strings_iter:
arg_string_pattern_parts.append('A')
# otherwise, add the arg to the arg strings
# and note the index if it was an option
else:
option_tuple = self._parse_optional(arg_string)
if option_tuple is None:
pattern = 'A'
else:
option_string_indices[i] = option_tuple
pattern = 'O'
arg_string_pattern_parts.append(pattern)
# join the pieces together to form the pattern
arg_strings_pattern = ''.join(arg_string_pattern_parts)
# converts arg strings to the appropriate and then takes the action
seen_actions: Set[Action] = set()
seen_non_default_actions: Set[Action] = set()
self._seen_non_default_actions = seen_non_default_actions # Added by argcomplete
def take_action(action, argument_strings, option_string=None):
seen_actions.add(action)
argument_values = self._get_values(action, argument_strings)
# error if this argument is not allowed with other previously
# seen arguments, assuming that actions that use the default
# value don't really count as "present"
if argument_values is not action.default:
seen_non_default_actions.add(action)
for conflict_action in action_conflicts.get(action, []):
if conflict_action in seen_non_default_actions:
msg = gettext('not allowed with argument %s')
action_name = _get_action_name(conflict_action)
raise ArgumentError(action, msg % action_name)
# take the action if we didn't receive a SUPPRESS value
# (e.g. from a default)
if argument_values is not SUPPRESS or isinstance(action, _SubParsersAction):
try:
action(self, namespace, argument_values, option_string)
except BaseException:
# Begin added by argcomplete
# When a subparser action is taken and fails due to incomplete arguments, it does not merge the
# contents of its parsed namespace into the parent namespace. Do that here to allow completers to
# access the partially parsed arguments for the subparser.
if isinstance(action, _SubParsersAction):
subnamespace = action._name_parser_map[argument_values[0]]._argcomplete_namespace
for key, value in vars(subnamespace).items():
setattr(namespace, key, value)
# End added by argcomplete
raise
# function to convert arg_strings into an optional action
def consume_optional(start_index):
# get the optional identified at this index
option_tuple = option_string_indices[start_index]
action, option_string, explicit_arg = option_tuple
# identify additional optionals in the same arg string
# (e.g. -xyz is the same as -x -y -z if no args are required)
match_argument = self._match_argument
action_tuples: List[Tuple[Action, List[str], str]] = []
while True:
# if we found no optional action, skip it
if action is None:
extras.append(arg_strings[start_index])
return start_index + 1
# if there is an explicit argument, try to match the
# optional's string arguments to only this
if explicit_arg is not None:
arg_count = match_argument(action, 'A')
# if the action is a single-dash option and takes no
# arguments, try to parse more single-dash options out
# of the tail of the option string
chars = self.prefix_chars
if arg_count == 0 and option_string[1] not in chars:
action_tuples.append((action, [], option_string))
char = option_string[0]
option_string = char + explicit_arg[0]
new_explicit_arg = explicit_arg[1:] or None
optionals_map = self._option_string_actions
if option_string in optionals_map:
action = optionals_map[option_string]
explicit_arg = new_explicit_arg
else:
msg = gettext('ignored explicit argument %r')
raise ArgumentError(action, msg % explicit_arg)
# if the action expect exactly one argument, we've
# successfully matched the option; exit the loop
elif arg_count == 1:
stop = start_index + 1
args = [explicit_arg]
action_tuples.append((action, args, option_string))
break
# error if a double-dash option did not use the
# explicit argument
else:
msg = gettext('ignored explicit argument %r')
raise ArgumentError(action, msg % explicit_arg)
# if there is no explicit argument, try to match the
# optional's string arguments with the following strings
# if successful, exit the loop
else:
start = start_index + 1
selected_patterns = arg_strings_pattern[start:]
self.active_actions = [action] # Added by argcomplete
_num_consumed_args[action] = 0 # Added by argcomplete
arg_count = match_argument(action, selected_patterns)
stop = start + arg_count
args = arg_strings[start:stop]
# Begin added by argcomplete
# If the pattern is not open (e.g. no + at the end), remove the action from active actions (since
# it wouldn't be able to consume any more args)
_num_consumed_args[action] = len(args)
if not action_is_open(action):
self.active_actions.remove(action)
# End added by argcomplete
action_tuples.append((action, args, option_string))
break
# add the Optional to the list and return the index at which
# the Optional's string args stopped
assert action_tuples
for action, args, option_string in action_tuples:
take_action(action, args, option_string)
return stop
# the list of Positionals left to be parsed; this is modified
# by consume_positionals()
positionals = self._get_positional_actions()
# function to convert arg_strings into positional actions
def consume_positionals(start_index):
# match as many Positionals as possible
match_partial = self._match_arguments_partial
selected_pattern = arg_strings_pattern[start_index:]
arg_counts = match_partial(positionals, selected_pattern)
# slice off the appropriate arg strings for each Positional
# and add the Positional and its args to the list
for action, arg_count in zip(positionals, arg_counts): # Added by argcomplete
self.active_actions.append(action) # Added by argcomplete
for action, arg_count in zip(positionals, arg_counts):
args = arg_strings[start_index : start_index + arg_count]
start_index += arg_count
_num_consumed_args[action] = len(args) # Added by argcomplete
take_action(action, args)
# slice off the Positionals that we just parsed and return the
# index at which the Positionals' string args stopped
positionals[:] = positionals[len(arg_counts) :]
return start_index
# consume Positionals and Optionals alternately, until we have
# passed the last option string
extras = []
start_index = 0
if option_string_indices:
max_option_string_index = max(option_string_indices)
else:
max_option_string_index = -1
while start_index <= max_option_string_index:
# consume any Positionals preceding the next option
next_option_string_index = min([index for index in option_string_indices if index >= start_index])
if start_index != next_option_string_index:
positionals_end_index = consume_positionals(start_index)
# only try to parse the next optional if we didn't consume
# the option string during the positionals parsing
if positionals_end_index > start_index:
start_index = positionals_end_index
continue
else:
start_index = positionals_end_index
# if we consumed all the positionals we could and we're not
# at the index of an option string, there were extra arguments
if start_index not in option_string_indices:
strings = arg_strings[start_index:next_option_string_index]
extras.extend(strings)
start_index = next_option_string_index
# consume the next optional and any arguments for it
start_index = consume_optional(start_index)
# consume any positionals following the last Optional
stop_index = consume_positionals(start_index)
# if we didn't consume all the argument strings, there were extras
extras.extend(arg_strings[stop_index:])
# if we didn't use all the Positional objects, there were too few
# arg strings supplied.
if positionals:
self.active_actions.append(positionals[0]) # Added by argcomplete
self.error(gettext('too few arguments'))
# make sure all required actions were present
for action in self._actions:
if action.required:
if action not in seen_actions:
name = _get_action_name(action)
self.error(gettext('argument %s is required') % name)
# make sure all required groups had one option present
for group in self._mutually_exclusive_groups:
if group.required:
for action in group._group_actions:
if action in seen_non_default_actions:
break
# if no actions were used, report the error
else:
names = [
str(_get_action_name(action)) for action in group._group_actions if action.help is not SUPPRESS
]
msg = gettext('one of the arguments %s is required')
self.error(msg % ' '.join(names))
# return the updated namespace and the extra arguments
return namespace, extras

View File

@@ -0,0 +1,310 @@
# This copy of shlex.py from Python 3.6 is distributed with argcomplete.
# It contains only the shlex class, with modifications as noted.
"""A lexical analyzer class for simple shell-like syntaxes."""
# Module and documentation by Eric S. Raymond, 21 Dec 1998
# Input stacking and error message cleanup added by ESR, March 2000
# push_source() and pop_source() made explicit by ESR, January 2001.
# Posix compliance, split(), string arguments, and
# iterator interface by Gustavo Niemeyer, April 2003.
# changes to tokenize more like Posix shells by Vinay Sajip, July 2016.
import os
import sys
from collections import deque
from io import StringIO
from typing import Optional
class shlex:
"A lexical analyzer class for simple shell-like syntaxes."
def __init__(self, instream=None, infile=None, posix=False, punctuation_chars=False):
# Modified by argcomplete: 2/3 compatibility
if isinstance(instream, str):
instream = StringIO(instream)
if instream is not None:
self.instream = instream
self.infile = infile
else:
self.instream = sys.stdin
self.infile = None
self.posix = posix
if posix:
self.eof = None
else:
self.eof = ''
self.commenters = '#'
self.wordchars = 'abcdfeghijklmnopqrstuvwxyz' 'ABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789_'
# Modified by argcomplete: 2/3 compatibility
# if self.posix:
# self.wordchars += ('ßàáâãäåæçèéêëìíîïðñòóôõöøùúûüýþÿ'
# 'ÀÁÂÃÄÅÆÇÈÉÊËÌÍÎÏÐÑÒÓÔÕÖØÙÚÛÜÝÞ')
self.whitespace = ' \t\r\n'
self.whitespace_split = False
self.quotes = '\'"'
self.escape = '\\'
self.escapedquotes = '"'
self.state: Optional[str] = ' '
self.pushback: deque = deque()
self.lineno = 1
self.debug = 0
self.token = ''
self.filestack: deque = deque()
self.source = None
if not punctuation_chars:
punctuation_chars = ''
elif punctuation_chars is True:
punctuation_chars = '();<>|&'
self.punctuation_chars = punctuation_chars
if punctuation_chars:
# _pushback_chars is a push back queue used by lookahead logic
self._pushback_chars: deque = deque()
# these chars added because allowed in file names, args, wildcards
self.wordchars += '~-./*?='
# remove any punctuation chars from wordchars
t = self.wordchars.maketrans(dict.fromkeys(punctuation_chars))
self.wordchars = self.wordchars.translate(t)
# Modified by argcomplete: Record last wordbreak position
self.last_wordbreak_pos = None
self.wordbreaks = ''
def push_token(self, tok):
"Push a token onto the stack popped by the get_token method"
if self.debug >= 1:
print("shlex: pushing token " + repr(tok))
self.pushback.appendleft(tok)
def push_source(self, newstream, newfile=None):
"Push an input source onto the lexer's input source stack."
# Modified by argcomplete: 2/3 compatibility
if isinstance(newstream, str):
newstream = StringIO(newstream)
self.filestack.appendleft((self.infile, self.instream, self.lineno))
self.infile = newfile
self.instream = newstream
self.lineno = 1
if self.debug:
if newfile is not None:
print('shlex: pushing to file %s' % (self.infile,))
else:
print('shlex: pushing to stream %s' % (self.instream,))
def pop_source(self):
"Pop the input source stack."
self.instream.close()
(self.infile, self.instream, self.lineno) = self.filestack.popleft()
if self.debug:
print('shlex: popping to %s, line %d' % (self.instream, self.lineno))
self.state = ' '
def get_token(self):
"Get a token from the input stream (or from stack if it's nonempty)"
if self.pushback:
tok = self.pushback.popleft()
if self.debug >= 1:
print("shlex: popping token " + repr(tok))
return tok
# No pushback. Get a token.
raw = self.read_token()
# Handle inclusions
if self.source is not None:
while raw == self.source:
spec = self.sourcehook(self.read_token())
if spec:
(newfile, newstream) = spec
self.push_source(newstream, newfile)
raw = self.get_token()
# Maybe we got EOF instead?
while raw == self.eof:
if not self.filestack:
return self.eof
else:
self.pop_source()
raw = self.get_token()
# Neither inclusion nor EOF
if self.debug >= 1:
if raw != self.eof:
print("shlex: token=" + repr(raw))
else:
print("shlex: token=EOF")
return raw
def read_token(self):
quoted = False
escapedstate = ' '
while True:
if self.punctuation_chars and self._pushback_chars:
nextchar = self._pushback_chars.pop()
else:
nextchar = self.instream.read(1)
if nextchar == '\n':
self.lineno += 1
if self.debug >= 3:
print("shlex: in state %r I see character: %r" % (self.state, nextchar))
if self.state is None:
self.token = '' # past end of file
break
elif self.state == ' ':
if not nextchar:
self.state = None # end of file
break
elif nextchar in self.whitespace:
if self.debug >= 2:
print("shlex: I see whitespace in whitespace state")
if self.token or (self.posix and quoted):
break # emit current token
else:
continue
elif nextchar in self.commenters:
self.instream.readline()
self.lineno += 1
elif self.posix and nextchar in self.escape:
escapedstate = 'a'
self.state = nextchar
elif nextchar in self.wordchars:
self.token = nextchar
self.state = 'a'
elif nextchar in self.punctuation_chars:
self.token = nextchar
self.state = 'c'
elif nextchar in self.quotes:
if not self.posix:
self.token = nextchar
self.state = nextchar
elif self.whitespace_split:
self.token = nextchar
self.state = 'a'
else:
self.token = nextchar
if self.token or (self.posix and quoted):
break # emit current token
else:
continue
elif self.state in self.quotes:
quoted = True
if not nextchar: # end of file
if self.debug >= 2:
print("shlex: I see EOF in quotes state")
# XXX what error should be raised here?
raise ValueError("No closing quotation")
if nextchar == self.state:
if not self.posix:
self.token += nextchar
self.state = ' '
break
else:
self.state = 'a'
elif self.posix and nextchar in self.escape and self.state in self.escapedquotes:
escapedstate = self.state
self.state = nextchar
else:
self.token += nextchar
elif self.state in self.escape:
if not nextchar: # end of file
if self.debug >= 2:
print("shlex: I see EOF in escape state")
# XXX what error should be raised here?
raise ValueError("No escaped character")
# In posix shells, only the quote itself or the escape
# character may be escaped within quotes.
if escapedstate in self.quotes and nextchar != self.state and nextchar != escapedstate:
self.token += self.state
self.token += nextchar
self.state = escapedstate
elif self.state in ('a', 'c'):
if not nextchar:
self.state = None # end of file
break
elif nextchar in self.whitespace:
if self.debug >= 2:
print("shlex: I see whitespace in word state")
self.state = ' '
if self.token or (self.posix and quoted):
break # emit current token
else:
continue
elif nextchar in self.commenters:
self.instream.readline()
self.lineno += 1
if self.posix:
self.state = ' '
if self.token or (self.posix and quoted):
break # emit current token
else:
continue
elif self.posix and nextchar in self.quotes:
self.state = nextchar
elif self.posix and nextchar in self.escape:
escapedstate = 'a'
self.state = nextchar
elif self.state == 'c':
if nextchar in self.punctuation_chars:
self.token += nextchar
else:
if nextchar not in self.whitespace:
self._pushback_chars.append(nextchar)
self.state = ' '
break
elif nextchar in self.wordchars or nextchar in self.quotes or self.whitespace_split:
self.token += nextchar
# Modified by argcomplete: Record last wordbreak position
if nextchar in self.wordbreaks:
self.last_wordbreak_pos = len(self.token) - 1
else:
if self.punctuation_chars:
self._pushback_chars.append(nextchar)
else:
self.pushback.appendleft(nextchar)
if self.debug >= 2:
print("shlex: I see punctuation in word state")
self.state = ' '
if self.token or (self.posix and quoted):
break # emit current token
else:
continue
result: Optional[str] = self.token
self.token = ''
if self.posix and not quoted and result == '':
result = None
if self.debug > 1:
if result:
print("shlex: raw token=" + repr(result))
else:
print("shlex: raw token=EOF")
# Modified by argcomplete: Record last wordbreak position
if self.state == ' ':
self.last_wordbreak_pos = None
return result
def sourcehook(self, newfile):
"Hook called on a filename to be sourced."
if newfile[0] == '"':
newfile = newfile[1:-1]
# This implements cpp-like semantics for relative-path inclusion.
# Modified by argcomplete: 2/3 compatibility
if isinstance(self.infile, str) and not os.path.isabs(newfile):
newfile = os.path.join(os.path.dirname(self.infile), newfile)
return (newfile, open(newfile, "r"))
def error_leader(self, infile=None, lineno=None):
"Emit a C-compiler-like, Emacs-friendly error-message leader."
if infile is None:
infile = self.infile
if lineno is None:
lineno = self.lineno
return "\"%s\", line %d: " % (infile, lineno)
def __iter__(self):
return self
def __next__(self):
token = self.get_token()
if token == self.eof:
raise StopIteration
return token
# Modified by argcomplete: 2/3 compatibility
next = __next__

View File

@@ -0,0 +1,193 @@
# Copyright 2012-2023, Andrey Kislyuk and argcomplete contributors. Licensed under the terms of the
# `Apache License, Version 2.0 <http://www.apache.org/licenses/LICENSE-2.0>`_. Distribution of the LICENSE and NOTICE
# files with source copies of this package and derivative works is **REQUIRED** as specified by the Apache License.
# See https://github.com/kislyuk/argcomplete for more info.
from shlex import quote
bashcode = r"""
# Run something, muting output or redirecting it to the debug stream
# depending on the value of _ARC_DEBUG.
# If ARGCOMPLETE_USE_TEMPFILES is set, use tempfiles for IPC.
__python_argcomplete_run() {
if [[ -z "${ARGCOMPLETE_USE_TEMPFILES-}" ]]; then
__python_argcomplete_run_inner "$@"
return
fi
local tmpfile="$(mktemp)"
_ARGCOMPLETE_STDOUT_FILENAME="$tmpfile" __python_argcomplete_run_inner "$@"
local code=$?
cat "$tmpfile"
rm "$tmpfile"
return $code
}
__python_argcomplete_run_inner() {
if [[ -z "${_ARC_DEBUG-}" ]]; then
"$@" 8>&1 9>&2 1>/dev/null 2>&1
else
"$@" 8>&1 9>&2 1>&9 2>&1
fi
}
_python_argcomplete%(function_suffix)s() {
local IFS=$'\013'
local script="%(argcomplete_script)s"
if [[ -n "${ZSH_VERSION-}" ]]; then
local completions
completions=($(IFS="$IFS" \
COMP_LINE="$BUFFER" \
COMP_POINT="$CURSOR" \
_ARGCOMPLETE=1 \
_ARGCOMPLETE_SHELL="zsh" \
_ARGCOMPLETE_SUPPRESS_SPACE=1 \
__python_argcomplete_run ${script:-${words[1]}}))
local nosort=()
local nospace=()
if is-at-least 5.8; then
nosort=(-o nosort)
fi
if [[ "${completions-}" =~ ([^\\]): && "${match[1]}" =~ [=/:] ]]; then
nospace=(-S '')
fi
_describe "${words[1]}" completions "${nosort[@]}" "${nospace[@]}"
else
local SUPPRESS_SPACE=0
if compopt +o nospace 2> /dev/null; then
SUPPRESS_SPACE=1
fi
COMPREPLY=($(IFS="$IFS" \
COMP_LINE="$COMP_LINE" \
COMP_POINT="$COMP_POINT" \
COMP_TYPE="$COMP_TYPE" \
_ARGCOMPLETE_COMP_WORDBREAKS="$COMP_WORDBREAKS" \
_ARGCOMPLETE=1 \
_ARGCOMPLETE_SHELL="bash" \
_ARGCOMPLETE_SUPPRESS_SPACE=$SUPPRESS_SPACE \
__python_argcomplete_run ${script:-$1}))
if [[ $? != 0 ]]; then
unset COMPREPLY
elif [[ $SUPPRESS_SPACE == 1 ]] && [[ "${COMPREPLY-}" =~ [=/:]$ ]]; then
compopt -o nospace
fi
fi
}
if [[ -z "${ZSH_VERSION-}" ]]; then
complete %(complete_opts)s -F _python_argcomplete%(function_suffix)s %(executables)s
else
autoload is-at-least
compdef _python_argcomplete%(function_suffix)s %(executables)s
fi
"""
tcshcode = """\
complete "%(executable)s" 'p@*@`python-argcomplete-tcsh "%(argcomplete_script)s"`@' ;
"""
fishcode = r"""
function __fish_%(function_name)s_complete
set -x _ARGCOMPLETE 1
set -x _ARGCOMPLETE_DFS \t
set -x _ARGCOMPLETE_IFS \n
set -x _ARGCOMPLETE_SUPPRESS_SPACE 1
set -x _ARGCOMPLETE_SHELL fish
set -x COMP_LINE (commandline -p)
set -x COMP_POINT (string length (commandline -cp))
set -x COMP_TYPE
if set -q _ARC_DEBUG
%(argcomplete_script)s 8>&1 9>&2 1>&9 2>&1
else
%(argcomplete_script)s 8>&1 9>&2 1>/dev/null 2>&1
end
end
complete %(completion_arg)s %(executable)s -f -a '(__fish_%(function_name)s_complete)'
"""
powershell_code = r"""
Register-ArgumentCompleter -Native -CommandName %(executable)s -ScriptBlock {
param($commandName, $wordToComplete, $cursorPosition)
$completion_file = New-TemporaryFile
$env:ARGCOMPLETE_USE_TEMPFILES = 1
$env:_ARGCOMPLETE_STDOUT_FILENAME = $completion_file
$env:COMP_LINE = $wordToComplete
$env:COMP_POINT = $cursorPosition
$env:_ARGCOMPLETE = 1
$env:_ARGCOMPLETE_SUPPRESS_SPACE = 0
$env:_ARGCOMPLETE_IFS = "`n"
$env:_ARGCOMPLETE_SHELL = "powershell"
%(argcomplete_script)s 2>&1 | Out-Null
Get-Content $completion_file | ForEach-Object {
[System.Management.Automation.CompletionResult]::new($_, $_, "ParameterValue", $_)
}
Remove-Item $completion_file, Env:\_ARGCOMPLETE_STDOUT_FILENAME, Env:\ARGCOMPLETE_USE_TEMPFILES, Env:\COMP_LINE, Env:\COMP_POINT, Env:\_ARGCOMPLETE, Env:\_ARGCOMPLETE_SUPPRESS_SPACE, Env:\_ARGCOMPLETE_IFS, Env:\_ARGCOMPLETE_SHELL
}
""" # noqa: E501
shell_codes = {"bash": bashcode, "tcsh": tcshcode, "fish": fishcode, "powershell": powershell_code}
def shellcode(executables, use_defaults=True, shell="bash", complete_arguments=None, argcomplete_script=None):
"""
Provide the shell code required to register a python executable for use with the argcomplete module.
:param list(str) executables: Executables to be completed (when invoked exactly with this name)
:param bool use_defaults: Whether to fallback to readline's default completion when no matches are generated
(affects bash only)
:param str shell: Name of the shell to output code for
:param complete_arguments: Arguments to call complete with (affects bash only)
:type complete_arguments: list(str) or None
:param argcomplete_script: Script to call complete with, if not the executable to complete.
If supplied, will be used to complete *all* passed executables.
:type argcomplete_script: str or None
"""
if complete_arguments is None:
complete_options = "-o nospace -o default -o bashdefault" if use_defaults else "-o nospace -o bashdefault"
else:
complete_options = " ".join(complete_arguments)
if shell == "bash" or shell == "zsh":
quoted_executables = [quote(i) for i in executables]
executables_list = " ".join(quoted_executables)
script = argcomplete_script
if script:
function_suffix = "_" + script
else:
script = ""
function_suffix = ""
code = bashcode % dict(
complete_opts=complete_options,
executables=executables_list,
argcomplete_script=script,
function_suffix=function_suffix,
)
elif shell == "fish":
code = ""
for executable in executables:
script = argcomplete_script or executable
completion_arg = "--path" if "/" in executable else "--command" # use path for absolute paths
function_name = executable.replace("/", "_") # / not allowed in function name
code += fishcode % dict(
executable=executable,
argcomplete_script=script,
completion_arg=completion_arg,
function_name=function_name,
)
elif shell == "powershell":
code = ""
for executable in executables:
script = argcomplete_script or executable
code += powershell_code % dict(executable=executable, argcomplete_script=script)
else:
code = ""
for executable in executables:
script = argcomplete_script
# If no script was specified, default to the executable being completed.
if not script:
script = executable
code += shell_codes.get(shell, "") % dict(executable=executable, argcomplete_script=script)
return code