Source code for aiomonitor.termui.completion
from __future__ import annotations
import asyncio
import signal
from typing import TYPE_CHECKING, Iterable, List
import click
from click.parser import split_arg_string
from click.shell_completion import CompletionItem, _resolve_context, _resolve_incomplete
from prompt_toolkit.completion import CompleteEvent, Completer, Completion
from prompt_toolkit.document import Document
from ..context import current_monitor
if TYPE_CHECKING:
from ..monitor import Monitor
[docs]
class ClickCompleter(Completer):
def __init__(self, root_command: click.BaseCommand) -> None:
self._root_command = root_command
[docs]
def get_completions(
self,
document: Document,
complete_event: CompleteEvent,
) -> Iterable[Completion]:
args = split_arg_string(document.current_line)
incomplete = document.get_word_under_cursor()
if incomplete and args and args[-1] == incomplete:
args.pop()
click_ctx = _resolve_context(self._root_command, {}, "", args)
cmd_or_param, incomplete = _resolve_incomplete(click_ctx, args, incomplete)
completions: List[CompletionItem] = cmd_or_param.shell_complete(
click_ctx, incomplete
)
start_position = document.find_backwards(incomplete) or 0
for completion in completions:
yield Completion(
completion.value,
start_position=start_position,
)
[docs]
def complete_task_id(
ctx: click.Context,
param: click.Parameter,
incomplete: str,
) -> Iterable[str]:
# ctx here is created in the completer and does not have ctx.obj set as monitor.
# We take the monitor instance from the global context variable instead.
try:
self: Monitor = current_monitor.get()
except LookupError:
return []
return [
task_id
for task_id in map(
str, sorted(map(id, asyncio.all_tasks(loop=self._monitored_loop)))
)
if task_id.startswith(incomplete)
][:10]
[docs]
def complete_trace_id(
ctx: click.Context,
param: click.Parameter,
incomplete: str,
) -> Iterable[str]:
try:
self: Monitor = current_monitor.get()
except LookupError:
return []
return [
trace_id
for trace_id in map(str, sorted(self._terminated_tasks.keys()))
if trace_id.startswith(incomplete)
][:10]
[docs]
def complete_signal_names(
ctx: click.Context,
param: click.Parameter,
incomplete: str,
) -> Iterable[str]:
return [sig.name for sig in signal.Signals if sig.name.startswith(incomplete)]