Skip to content
Back to top

# Source code for xonsh.completer

    """A (tab-)completer for xonsh."""

    import collections.abc as cabc
    import os
    import sys
    import typing as tp

    from xonsh.built_ins import XSH
    from xonsh.completers.tools import (
        RICH_COMPLETION_DEFAULTS,
        Completion,
        RichCompletion,
        apply_lprefix,
        get_filter_function,
        is_contextual_completer,
        is_exclusive_completer,
    )
    from xonsh.events import events
    from xonsh.parsers.completion_context import CompletionContext, CompletionContextParser
    from xonsh.tools import print_above_prompt, print_exception

    events.doc(
        "on_completer_filter",
        """
    on_completer_filter(completer: str, command: str, context: CompletionContext) -> bool | None

    Fires once for every registered completer just before it is invoked,
    allowing handlers to veto the call. Return ``False`` to skip the
    completer; return ``True``, ``None`` (or nothing) to let it run. If any
    handler returns ``False``, the completer is skipped.

    Parameters:

    * ``completer``: the registered name of the completer about to run
      (e.g. ``'bash'``, ``'path'``).
    * ``command``: the basename of the command being completed
      (e.g. ``'kubectl'`` even when the user typed ``/usr/bin/kubectl``).
      Empty string when no command has been typed yet, or when the
      completion happens outside the command-line context.
    * ``context``: the full :class:`CompletionContext` for the current
      completion request.

    Use this to short-circuit completers such as ``bash`` on commands
    where the user does not need bash-supplied completions:

    .. code-block:: xonsh

        @events.on_completer_filter
        def _only_bash_for_some(completer, command, context, **_):
            return command in {'kubectl', 'docker'} if completer == 'bash' else True

    Handlers run on the completion thread; exceptions raised inside a
    handler are logged and treated as if no value was returned (i.e. the
    completer is still allowed to run).
    """,
    )

    class Completer:
        """This provides a list of optional completions for the xonsh shell."""

        def __init__(self):
            self.context_parser = CompletionContextParser()

        def parse(
            self, text: str, cursor_index: "None|int" = None, ctx=None
        ) -> "CompletionContext":
            """Parse the given text

            Parameters
            ----------
            text
                multi-line text
            cursor_index
                position of the cursor. If not given, then it is considered to be at the end.
            ctx
                Execution context
            """
            cursor_index = len(text) if cursor_index is None else cursor_index
            return self.context_parser.parse(text, cursor_index, ctx)

        def complete_line(self, text: str):
            """Handy wrapper to build command-completion-context when cursor is at the end.

            Notes
            -----
            suffix is not supported; text after last space is parsed as prefix.
            """
            ctx = self.parse(text)

            if not ctx:
                raise RuntimeError("CompletionContext is None")

            if ctx.python is not None:
                prefix = ctx.python.prefix
            elif ctx.command is not None:
                prefix = ctx.command.prefix
            else:
                raise RuntimeError("CompletionContext is empty")

            line = text
            begidx = text.rfind(prefix)
            endidx = begidx + len(prefix)

            return self.complete(
                prefix,
                line,
                begidx,
                endidx,
                cursor_index=len(line),
                multiline_text=line,
                completion_context=ctx,
            )

        def complete(
            self,
            prefix,
            line,
            begidx,
            endidx,
            ctx=None,
            multiline_text=None,
            cursor_index=None,
            completion_context=None,
        ):
            """Complete the string, given a possible execution context.

            Parameters
            ----------
            prefix : str
                The string to match
            line : str
                The line that prefix appears on.
            begidx : int
                The index in line that prefix starts on.
            endidx : int
                The index in line that prefix ends on.
            ctx : dict, optional
                Names in the current execution context.
            multiline_text : str
                The complete multiline text. Needed to get completion context.
            cursor_index : int
                The current cursor's index in the multiline text.
                May be ``len(multiline_text)`` for cursor at the end.
                Needed to get completion context.

            Returns
            -------
            rtn : list of str
                Possible completions of prefix, sorted alphabetically.
            lprefix : int
                Length of the prefix to be replaced in the completion.
            """

            if (
                (multiline_text is not None)
                and (cursor_index is not None)
                and (completion_context is None)
            ):
                completion_context: CompletionContext | None = self.parse(
                    multiline_text,
                    cursor_index,
                    ctx,
                )

            ctx = ctx or {}
            return self.complete_from_context(
                completion_context,
                (prefix, line, begidx, endidx, ctx),
            )

        @staticmethod
        def _format_completion(
            completion,
            completion_context,
            completing_contextual_command: bool,
            lprefix: int,
            custom_lprefix: bool,
        ) -> tuple[Completion, int]:
            if (
                completing_contextual_command
                and completion_context.command.is_after_closing_quote
            ):
                """
                The cursor is appending to a closed string literal, i.e. cursor at the end of ``ls "/usr/"``.
                1. The closing quote will be appended to all completions.
                    I.e the completion ``/usr/bin`` will turn into ``/usr/bin"``
                    To prevent this behavior, a completer can return a ``RichCompletion`` with ``append_closing_quote=False``.
                2. If not specified, lprefix will cover the closing prefix.
                    I.e for ``ls "/usr/"``, the default lprefix will be 6 to include the closing quote.
                    To prevent this behavior, a completer can return a different lprefix or specify it inside ``RichCompletion``.
                """
                closing_quote = completion_context.command.closing_quote
                if not custom_lprefix:
                    lprefix += len(closing_quote)
                if closing_quote:
                    if isinstance(completion, RichCompletion):
                        if completion.append_closing_quote:
                            completion = completion.replace(
                                value=completion.value + closing_quote
                            )
                    else:
                        completion = completion + closing_quote

            completion = list(apply_lprefix([completion], lprefix))[0]

            if (
                isinstance(completion, RichCompletion)
                and completion.append_space
                and not completion.value.endswith(" ")
            ):
                # append spaces AFTER appending closing quote
                completion = completion.replace(value=completion.value + " ")

            return completion, lprefix

        # Renames applied to ``RichCompletion`` attribute names in trace
        # output only; the Python attributes themselves are unchanged.
        _TRACE_ATTR_ALIASES = {
            "append_closing_quote": "close_quote",
        }

        @staticmethod
        def _format_trace_item(comp, lprefix, source: str, exclusive: bool) -> str:
            """Render one completion as a single ``$XONSH_COMPLETER_TRACE`` line.

            Format::

                "value": src=<completer>, pvd=<sub-source>, type=<exclusive|non-exclusive>, <non-default RichCompletion attrs>

            ``pvd`` (provider) is placed right after ``src`` (source) when
            set, so the two origin fields stay visually adjacent. Other
            non-default ``RichCompletion`` fields (``prefix_len``,
            ``display``, ``description``, ``append_space``, ``close_quote``
            — short for ``append_closing_quote`` — and ``style``) are
            appended afterwards. Plain ``str`` completions fall back to
            showing the pipeline's ``lprefix``.
            """
            parts = [f"src={source}"]
            if isinstance(comp, RichCompletion) and comp.provider is not None:
                parts.append(f"pvd={comp.provider!r}")
            parts.append(f"type={'exclusive' if exclusive else 'non-exclusive'}")
            if isinstance(comp, RichCompletion):
                for name, default in RICH_COMPLETION_DEFAULTS:
                    if name == "provider":
                        continue  # already placed right after ``src``
                    val = getattr(comp, name)
                    if val != default:
                        label = Completer._TRACE_ATTR_ALIASES.get(name, name)
                        parts.append(f"{label}={val!r}")
            else:
                parts.append(f"prefix_len={lprefix}")
            return f"{str(comp)!r}: {', '.join(parts)}"

        @staticmethod
        def generate_completions(
            completion_context, old_completer_args, trace: bool
        ) -> tp.Iterator[tuple[Completion, int]]:
            filter_func = get_filter_function()

            if (
                completion_context is not None
                and completion_context.command is not None
                and completion_context.command.args
            ):
                command = os.path.basename(completion_context.command.args[0].value)
            else:
                command = ""

            for name, func in XSH.completers.items():
                veto_by = None
                for handler, rv in events.on_completer_filter.fire_iter(
                    completer=name, command=command, context=completion_context
                ):
                    if rv is False:
                        veto_by = handler
                if veto_by is not None:
                    if trace:
                        hmod = getattr(veto_by, "__module__", "unknown")
                        hname = getattr(veto_by, "__name__", repr(veto_by))
                        print_above_prompt(
                            f"TRACE COMPLETIONS: Skipped '{name}' by on_completer_filter "
                            f"handler '{hmod}.{hname}'."
                        )
                    continue
                try:
                    if is_contextual_completer(func):
                        if completion_context is None:
                            continue
                        out = func(completion_context)
                    else:
                        if old_completer_args is None:
                            continue
                        out = func(*old_completer_args)
                except StopIteration:
                    # completer requested to stop collecting completions
                    break
                except Exception as e:
                    name = func.__name__ if hasattr(func, "__name__") else str(func)
                    print_exception(
                        f"Completer {name} raises exception when gets "
                        f"old_args={old_completer_args[:-1]} / completion_context={completion_context!r}:\n"
                        f"{type(e)} - {e}"
                    )
                    continue

                completing_contextual_command = (
                    is_contextual_completer(func)
                    and completion_context is not None
                    and completion_context.command is not None
                )

                # -- set comp-defaults --

                # the default is that the completer function filters out as necessary
                # we can change that once fuzzy/substring matches are added
                is_filtered = True
                custom_lprefix = False
                prefix = ""
                if completing_contextual_command:
                    prefix = completion_context.command.prefix
                elif old_completer_args is not None:
                    prefix = old_completer_args[0]
                lprefix = len(prefix)

                if isinstance(out, cabc.Sequence):
                    # update comp-defaults from
                    res, lprefix_filtered = out
                    if isinstance(lprefix_filtered, bool):
                        is_filtered = lprefix_filtered
                    else:
                        lprefix = lprefix_filtered
                        custom_lprefix = True
                else:
                    res = out

                # Completer was invoked (didn't raise, wasn't skipped by the
                # contextual gate). Process its output into ``items``; note
                # that ``res is None`` is valid — it just yields an empty
                # ``items``, and trace will report ``Got 0 results``.
                items: list = []
                exclusive = is_exclusive_completer(func)
                try:
                    if res is not None:
                        for comp in res:
                            if (not is_filtered) and (not filter_func(comp, prefix)):
                                continue
                            # Skip empty or whitespace-only completions as if the
                            # completer had not returned them. Without this an
                            # exclusive completer that yields only blanks (e.g. a
                            # subprocess-based completer returning spurious empty
                            # lines) would short-circuit the pipeline and hide
                            # the fallback ``python``/``path`` completers. See #5810.
                            if not str(comp).strip():
                                continue
                            comp = Completer._format_completion(
                                comp,
                                completion_context,
                                completing_contextual_command,
                                lprefix or 0,
                                custom_lprefix,
                            )
                            items.append(comp)
                            yield comp
                finally:
                    # Trace lives in ``finally`` so it still runs when the
                    # consumer breaks mid-completer (``GeneratorExit`` at the
                    # ``yield``) — otherwise the items that triggered
                    # ``$COMPLETION_QUERY_LIMIT`` would never be printed.
                    if trace:
                        exclusivity = "exclusive" if exclusive else "non-exclusive"
                        if not items:
                            # Completer was invoked but produced nothing usable —
                            # still report it so the user can see which completers
                            # ran. Trailing ``.`` since no list follows.
                            print(
                                f"TRACE COMPLETIONS: Got 0"
                                f" from {exclusivity} '{name}'"
                                f" for {prefix!r}."
                            )
                        else:
                            print(
                                f"TRACE COMPLETIONS: Got {len(items)}"
                                f" from {exclusivity} '{name}'"
                                f" for {prefix!r}:"
                            )
                            for item_comp, item_lprefix in items:
                                print(
                                    Completer._format_trace_item(
                                        item_comp, item_lprefix, name, exclusive
                                    )
                                )

                if not items:  # empty completion
                    continue

                if exclusive:
                    # we got completions for an exclusive completer
                    break

        def complete_from_context(self, completion_context, old_completer_args=None):
            trace = XSH.env.get("XONSH_COMPLETER_TRACE")
            if trace:
                print("\nTRACE COMPLETIONS: Getting completions with context:")
                sys.displayhook(completion_context)
            lprefix = 0

            # using dict to keep order py3.6+
            completions = {}

            query_limit = XSH.env.get("COMPLETION_QUERY_LIMIT")

            # Whether there is any typed content at all. Bare Tab on a
            # completely empty line always yields a huge candidate list, so
            # the truncation notice would be pure noise there. For anything
            # else (including ``ls <Tab>`` — empty arg prefix but a real
            # command line) we want the notice to surface.
            has_line_content = False
            if old_completer_args and len(old_completer_args) >= 2:
                has_line_content = bool(old_completer_args[1])
            elif completion_context is not None:
                if completion_context.command is not None:
                    cmd = completion_context.command
                    has_line_content = bool(cmd.args or cmd.prefix)
                elif completion_context.python is not None:
                    has_line_content = bool(completion_context.python.prefix)

            gen = self.generate_completions(
                completion_context,
                old_completer_args,
                trace,
            )
            limit_reached = False
            try:
                for comp in gen:
                    completion, lprefix = comp
                    completions[completion] = None
                    if query_limit and len(completions) >= query_limit:
                        limit_reached = True
                        if has_line_content:
                            print_above_prompt(
                                f"List truncated by $COMPLETION_QUERY_LIMIT = {query_limit}"
                            )
                        break
            finally:
                # Force the generator's ``finally`` (the per-completer trace
                # block) to run before any post-loop output, so the items of
                # the completer that was interrupted by the query-limit
                # ``break`` actually get printed.
                gen.close()

            if trace and limit_reached:
                print("TRACE COMPLETIONS: Stopped after $COMPLETION_QUERY_LIMIT reached.")

            # Deduplicate completions that differ only by a trailing space.
            # For example, ``_cd`` (from Python name completions) and ``_cd ``
            # (from command completions with append_space=True) should appear
            # only once.  We keep the spaced variant because it carries the
            # richer completion metadata.
            spaced = {str(c) for c in completions if str(c).endswith(" ")}
            if spaced:
                completions = {
                    c: None
                    for c in completions
                    if str(c).endswith(" ") or (str(c) + " ") not in spaced
                }

            if completion_context:
                if completion_context.command is not None:
                    prefix = completion_context.command.prefix
                elif completion_context.python is not None:
                    prefix = completion_context.python.prefix
                else:
                    raise RuntimeError("Completion context is empty")

                if prefix.startswith("$"):
                    prefix = prefix[1:]

                lower_prefix = prefix.lower()

                def sortkey(s):
                    """Sort completions by match quality tier, then by underscore prefix, match position, and name.

                    Tiers:
                      0 - case-sensitive prefix match
                      1 - case-insensitive prefix match
                      2 - case-sensitive substring match
                      3 - case-insensitive substring match
                      4 - no match (sorted last)

                    Within each tier, completions whose last component starts
                    with '_' are sorted last (handles both bare names like
                    ``_codecs`` and dotted names like ``json._default_decoder``),
                    then by position of the match, then alphabetically.
                    """
                    text = str(s)
                    ltext = text.lower()
                    if text.startswith(prefix):
                        tier = 0
                    elif ltext.startswith(lower_prefix):
                        tier = 1
                    elif prefix in text:
                        tier = 2
                    elif lower_prefix in ltext:
                        tier = 3
                    else:
                        tier = 4
                    last_part = text.rsplit(".", 1)[-1]
                    has_leading_underscore = last_part.startswith("_")
                    pos = ltext.find(lower_prefix) if lower_prefix else 0
                    if pos < 0:
                        pos = 0
                    return (tier, has_leading_underscore, pos, ltext)
            else:
                # Fallback sort.
                sortkey = lambda s: s.lstrip(''''"''').lower()

            # the last completer's lprefix is returned. other lprefix values are inside the RichCompletions.
            return tuple(sorted(completions, key=sortkey)), lprefix