forked from clan/clan-core
Merge pull request 'clan: add dynamic completions' (#1522) from a-kenji-clan/cli/init-dynamic-completions into main
This commit is contained in:
commit
9f19a8e605
@ -152,6 +152,7 @@ For more detailed information, visit: https://docs.clan.lol/getting-started
|
|||||||
),
|
),
|
||||||
formatter_class=argparse.RawTextHelpFormatter,
|
formatter_class=argparse.RawTextHelpFormatter,
|
||||||
)
|
)
|
||||||
|
|
||||||
flakes.register_parser(parser_flake)
|
flakes.register_parser(parser_flake)
|
||||||
|
|
||||||
parser_config = subparsers.add_parser(
|
parser_config = subparsers.add_parser(
|
||||||
|
@ -2,6 +2,7 @@ import argparse
|
|||||||
import json
|
import json
|
||||||
import logging
|
import logging
|
||||||
|
|
||||||
|
from ..completions import add_dynamic_completer, complete_machines
|
||||||
from ..errors import ClanError
|
from ..errors import ClanError
|
||||||
from ..machines.machines import Machine
|
from ..machines.machines import Machine
|
||||||
|
|
||||||
@ -40,8 +41,10 @@ def create_command(args: argparse.Namespace) -> None:
|
|||||||
|
|
||||||
|
|
||||||
def register_create_parser(parser: argparse.ArgumentParser) -> None:
|
def register_create_parser(parser: argparse.ArgumentParser) -> None:
|
||||||
parser.add_argument(
|
machines_parser = parser.add_argument(
|
||||||
"machine", type=str, help="machine in the flake to create backups of"
|
"machine", type=str, help="machine in the flake to create backups of"
|
||||||
)
|
)
|
||||||
|
add_dynamic_completer(machines_parser, complete_machines)
|
||||||
|
|
||||||
parser.add_argument("--provider", type=str, help="backup provider to use")
|
parser.add_argument("--provider", type=str, help="backup provider to use")
|
||||||
parser.set_defaults(func=create_command)
|
parser.set_defaults(func=create_command)
|
||||||
|
@ -3,6 +3,7 @@ import json
|
|||||||
import subprocess
|
import subprocess
|
||||||
from dataclasses import dataclass
|
from dataclasses import dataclass
|
||||||
|
|
||||||
|
from ..completions import add_dynamic_completer, complete_machines
|
||||||
from ..errors import ClanError
|
from ..errors import ClanError
|
||||||
from ..machines.machines import Machine
|
from ..machines.machines import Machine
|
||||||
|
|
||||||
@ -57,8 +58,9 @@ def list_command(args: argparse.Namespace) -> None:
|
|||||||
|
|
||||||
|
|
||||||
def register_list_parser(parser: argparse.ArgumentParser) -> None:
|
def register_list_parser(parser: argparse.ArgumentParser) -> None:
|
||||||
parser.add_argument(
|
machines_parser = parser.add_argument(
|
||||||
"machine", type=str, help="machine in the flake to show backups of"
|
"machine", type=str, help="machine in the flake to show backups of"
|
||||||
)
|
)
|
||||||
|
add_dynamic_completer(machines_parser, complete_machines)
|
||||||
parser.add_argument("--provider", type=str, help="backup provider to filter by")
|
parser.add_argument("--provider", type=str, help="backup provider to filter by")
|
||||||
parser.set_defaults(func=list_command)
|
parser.set_defaults(func=list_command)
|
||||||
|
81
pkgs/clan-cli/clan_cli/completions.py
Normal file
81
pkgs/clan-cli/clan_cli/completions.py
Normal file
@ -0,0 +1,81 @@
|
|||||||
|
import argparse
|
||||||
|
import json
|
||||||
|
import subprocess
|
||||||
|
import threading
|
||||||
|
from collections.abc import Callable, Iterable
|
||||||
|
from types import ModuleType
|
||||||
|
from typing import Any
|
||||||
|
|
||||||
|
"""
|
||||||
|
This module provides dynamic completions.
|
||||||
|
The completions should feel fast.
|
||||||
|
We target a maximum of 1second on our average machine.
|
||||||
|
"""
|
||||||
|
|
||||||
|
|
||||||
|
argcomplete: ModuleType | None = None
|
||||||
|
try:
|
||||||
|
import argcomplete # type: ignore[no-redef]
|
||||||
|
except ImportError:
|
||||||
|
pass
|
||||||
|
|
||||||
|
|
||||||
|
def clan_dir(flake: str | None) -> str | None:
|
||||||
|
from .dirs import get_clan_flake_toplevel_or_env
|
||||||
|
|
||||||
|
path_result = get_clan_flake_toplevel_or_env()
|
||||||
|
return str(path_result) if path_result is not None else None
|
||||||
|
|
||||||
|
|
||||||
|
def complete_machines(
|
||||||
|
prefix: str, parsed_args: argparse.Namespace, **kwargs: Any
|
||||||
|
) -> Iterable[str]:
|
||||||
|
"""
|
||||||
|
Provides completion functionality for machine names configured in the clan.
|
||||||
|
"""
|
||||||
|
machines: list[str] = []
|
||||||
|
|
||||||
|
def run_cmd() -> None:
|
||||||
|
try:
|
||||||
|
# In my tests this was consistently faster than:
|
||||||
|
# nix eval .#nixosConfigurations --apply builtins.attrNames
|
||||||
|
cmd = ["nix", "flake", "show", "--system", "no-eval", "--json"]
|
||||||
|
if (clan_dir_result := clan_dir(None)) is not None:
|
||||||
|
cmd.append(clan_dir_result)
|
||||||
|
result = subprocess.run(
|
||||||
|
cmd,
|
||||||
|
check=True,
|
||||||
|
capture_output=True,
|
||||||
|
text=True,
|
||||||
|
)
|
||||||
|
|
||||||
|
data = json.loads(result.stdout)
|
||||||
|
try:
|
||||||
|
machines.extend(data.get("nixosConfigurations").keys())
|
||||||
|
except KeyError:
|
||||||
|
pass
|
||||||
|
except subprocess.CalledProcessError:
|
||||||
|
pass
|
||||||
|
|
||||||
|
thread = threading.Thread(target=run_cmd)
|
||||||
|
thread.start()
|
||||||
|
thread.join(timeout=3)
|
||||||
|
|
||||||
|
if thread.is_alive():
|
||||||
|
return iter([])
|
||||||
|
|
||||||
|
machines_dict = {name: "machine" for name in machines}
|
||||||
|
return machines_dict
|
||||||
|
|
||||||
|
|
||||||
|
def add_dynamic_completer(
|
||||||
|
action: argparse.Action,
|
||||||
|
completer: Callable[..., Iterable[str]],
|
||||||
|
) -> None:
|
||||||
|
"""
|
||||||
|
Add a completion function to an argparse action, this will only be added,
|
||||||
|
if the argcomplete module is loaded.
|
||||||
|
"""
|
||||||
|
if argcomplete:
|
||||||
|
# mypy doesn't check this correctly, so we ignore it
|
||||||
|
action.completer = completer # type: ignore[attr-defined]
|
@ -2,6 +2,7 @@ import argparse
|
|||||||
import importlib
|
import importlib
|
||||||
import logging
|
import logging
|
||||||
|
|
||||||
|
from ..completions import add_dynamic_completer, complete_machines
|
||||||
from ..machines.machines import Machine
|
from ..machines.machines import Machine
|
||||||
|
|
||||||
log = logging.getLogger(__name__)
|
log = logging.getLogger(__name__)
|
||||||
@ -54,10 +55,12 @@ def check_command(args: argparse.Namespace) -> None:
|
|||||||
|
|
||||||
|
|
||||||
def register_check_parser(parser: argparse.ArgumentParser) -> None:
|
def register_check_parser(parser: argparse.ArgumentParser) -> None:
|
||||||
parser.add_argument(
|
machines_parser = parser.add_argument(
|
||||||
"machine",
|
"machine",
|
||||||
help="The machine to check secrets for",
|
help="The machine to check secrets for",
|
||||||
)
|
)
|
||||||
|
add_dynamic_completer(machines_parser, complete_machines)
|
||||||
|
|
||||||
parser.add_argument(
|
parser.add_argument(
|
||||||
"--service",
|
"--service",
|
||||||
help="the service to check",
|
help="the service to check",
|
||||||
|
@ -9,6 +9,7 @@ from tempfile import TemporaryDirectory
|
|||||||
|
|
||||||
from clan_cli.cmd import run
|
from clan_cli.cmd import run
|
||||||
|
|
||||||
|
from ..completions import add_dynamic_completer, complete_machines
|
||||||
from ..errors import ClanError
|
from ..errors import ClanError
|
||||||
from ..git import commit_files
|
from ..git import commit_files
|
||||||
from ..machines.inventory import get_all_machines, get_selected_machines
|
from ..machines.inventory import get_all_machines, get_selected_machines
|
||||||
@ -216,13 +217,15 @@ def generate_command(args: argparse.Namespace) -> None:
|
|||||||
|
|
||||||
|
|
||||||
def register_generate_parser(parser: argparse.ArgumentParser) -> None:
|
def register_generate_parser(parser: argparse.ArgumentParser) -> None:
|
||||||
parser.add_argument(
|
machines_parser = parser.add_argument(
|
||||||
"machines",
|
"machines",
|
||||||
type=str,
|
type=str,
|
||||||
help="machine to generate facts for. if empty, generate facts for all machines",
|
help="machine to generate facts for. if empty, generate facts for all machines",
|
||||||
nargs="*",
|
nargs="*",
|
||||||
default=[],
|
default=[],
|
||||||
)
|
)
|
||||||
|
add_dynamic_completer(machines_parser, complete_machines)
|
||||||
|
|
||||||
parser.add_argument(
|
parser.add_argument(
|
||||||
"--service",
|
"--service",
|
||||||
type=str,
|
type=str,
|
||||||
|
@ -3,6 +3,7 @@ import importlib
|
|||||||
import json
|
import json
|
||||||
import logging
|
import logging
|
||||||
|
|
||||||
|
from ..completions import add_dynamic_completer, complete_machines
|
||||||
from ..machines.machines import Machine
|
from ..machines.machines import Machine
|
||||||
|
|
||||||
log = logging.getLogger(__name__)
|
log = logging.getLogger(__name__)
|
||||||
@ -37,8 +38,10 @@ def get_command(args: argparse.Namespace) -> None:
|
|||||||
|
|
||||||
|
|
||||||
def register_list_parser(parser: argparse.ArgumentParser) -> None:
|
def register_list_parser(parser: argparse.ArgumentParser) -> None:
|
||||||
parser.add_argument(
|
machines_parser = parser.add_argument(
|
||||||
"machine",
|
"machine",
|
||||||
help="The machine to print facts for",
|
help="The machine to print facts for",
|
||||||
)
|
)
|
||||||
|
add_dynamic_completer(machines_parser, complete_machines)
|
||||||
|
|
||||||
parser.set_defaults(func=get_command)
|
parser.set_defaults(func=get_command)
|
||||||
|
@ -5,6 +5,7 @@ from pathlib import Path
|
|||||||
from tempfile import TemporaryDirectory
|
from tempfile import TemporaryDirectory
|
||||||
|
|
||||||
from ..cmd import Log, run
|
from ..cmd import Log, run
|
||||||
|
from ..completions import add_dynamic_completer, complete_machines
|
||||||
from ..machines.machines import Machine
|
from ..machines.machines import Machine
|
||||||
from ..nix import nix_shell
|
from ..nix import nix_shell
|
||||||
|
|
||||||
@ -46,8 +47,10 @@ def upload_command(args: argparse.Namespace) -> None:
|
|||||||
|
|
||||||
|
|
||||||
def register_upload_parser(parser: argparse.ArgumentParser) -> None:
|
def register_upload_parser(parser: argparse.ArgumentParser) -> None:
|
||||||
parser.add_argument(
|
machines_parser = parser.add_argument(
|
||||||
"machine",
|
"machine",
|
||||||
help="The machine to upload secrets to",
|
help="The machine to upload secrets to",
|
||||||
)
|
)
|
||||||
|
add_dynamic_completer(machines_parser, complete_machines)
|
||||||
|
|
||||||
parser.set_defaults(func=upload_command)
|
parser.set_defaults(func=upload_command)
|
||||||
|
@ -12,6 +12,7 @@ from tempfile import TemporaryDirectory
|
|||||||
from typing import Any
|
from typing import Any
|
||||||
|
|
||||||
from .cmd import Log, run
|
from .cmd import Log, run
|
||||||
|
from .completions import add_dynamic_completer, complete_machines
|
||||||
from .errors import ClanError
|
from .errors import ClanError
|
||||||
from .facts.secret_modules import SecretStoreBase
|
from .facts.secret_modules import SecretStoreBase
|
||||||
from .machines.machines import Machine
|
from .machines.machines import Machine
|
||||||
@ -173,11 +174,13 @@ def flash_command(args: argparse.Namespace) -> None:
|
|||||||
|
|
||||||
|
|
||||||
def register_parser(parser: argparse.ArgumentParser) -> None:
|
def register_parser(parser: argparse.ArgumentParser) -> None:
|
||||||
parser.add_argument(
|
machines_parser = parser.add_argument(
|
||||||
"machine",
|
"machine",
|
||||||
type=str,
|
type=str,
|
||||||
help="machine to install",
|
help="machine to install",
|
||||||
)
|
)
|
||||||
|
add_dynamic_completer(machines_parser, complete_machines)
|
||||||
|
|
||||||
parser.add_argument(
|
parser.add_argument(
|
||||||
"--disk",
|
"--disk",
|
||||||
type=str,
|
type=str,
|
||||||
|
@ -1,6 +1,7 @@
|
|||||||
import argparse
|
import argparse
|
||||||
import shutil
|
import shutil
|
||||||
|
|
||||||
|
from ..completions import add_dynamic_completer, complete_machines
|
||||||
from ..dirs import specific_machine_dir
|
from ..dirs import specific_machine_dir
|
||||||
from ..errors import ClanError
|
from ..errors import ClanError
|
||||||
|
|
||||||
@ -14,5 +15,7 @@ def delete_command(args: argparse.Namespace) -> None:
|
|||||||
|
|
||||||
|
|
||||||
def register_delete_parser(parser: argparse.ArgumentParser) -> None:
|
def register_delete_parser(parser: argparse.ArgumentParser) -> None:
|
||||||
parser.add_argument("host", type=str)
|
machines_parser = parser.add_argument("host", type=str)
|
||||||
|
add_dynamic_completer(machines_parser, complete_machines)
|
||||||
|
|
||||||
parser.set_defaults(func=delete_command)
|
parser.set_defaults(func=delete_command)
|
||||||
|
@ -8,6 +8,7 @@ from pathlib import Path
|
|||||||
from tempfile import TemporaryDirectory
|
from tempfile import TemporaryDirectory
|
||||||
|
|
||||||
from ..cmd import Log, run
|
from ..cmd import Log, run
|
||||||
|
from ..completions import add_dynamic_completer, complete_machines
|
||||||
from ..facts.generate import generate_facts
|
from ..facts.generate import generate_facts
|
||||||
from ..machines.machines import Machine
|
from ..machines.machines import Machine
|
||||||
from ..nix import nix_shell
|
from ..nix import nix_shell
|
||||||
@ -188,11 +189,14 @@ def register_install_parser(parser: argparse.ArgumentParser) -> None:
|
|||||||
help="do not ask for confirmation",
|
help="do not ask for confirmation",
|
||||||
default=False,
|
default=False,
|
||||||
)
|
)
|
||||||
parser.add_argument(
|
|
||||||
|
machines_parser = parser.add_argument(
|
||||||
"machine",
|
"machine",
|
||||||
type=str,
|
type=str,
|
||||||
help="machine to install",
|
help="machine to install",
|
||||||
)
|
)
|
||||||
|
add_dynamic_completer(machines_parser, complete_machines)
|
||||||
|
|
||||||
parser.add_argument(
|
parser.add_argument(
|
||||||
"target_host",
|
"target_host",
|
||||||
type=str,
|
type=str,
|
||||||
|
@ -6,6 +6,7 @@ import shlex
|
|||||||
import subprocess
|
import subprocess
|
||||||
import sys
|
import sys
|
||||||
|
|
||||||
|
from ..completions import add_dynamic_completer, complete_machines
|
||||||
from ..errors import ClanError
|
from ..errors import ClanError
|
||||||
from ..facts.generate import generate_facts
|
from ..facts.generate import generate_facts
|
||||||
from ..facts.upload import upload_secrets
|
from ..facts.upload import upload_secrets
|
||||||
@ -180,7 +181,7 @@ def update(args: argparse.Namespace) -> None:
|
|||||||
|
|
||||||
|
|
||||||
def register_update_parser(parser: argparse.ArgumentParser) -> None:
|
def register_update_parser(parser: argparse.ArgumentParser) -> None:
|
||||||
parser.add_argument(
|
machines_parser = parser.add_argument(
|
||||||
"machines",
|
"machines",
|
||||||
type=str,
|
type=str,
|
||||||
nargs="*",
|
nargs="*",
|
||||||
@ -188,6 +189,9 @@ def register_update_parser(parser: argparse.ArgumentParser) -> None:
|
|||||||
metavar="MACHINE",
|
metavar="MACHINE",
|
||||||
help="machine to update. If no machine is specified, all machines will be updated.",
|
help="machine to update. If no machine is specified, all machines will be updated.",
|
||||||
)
|
)
|
||||||
|
|
||||||
|
add_dynamic_completer(machines_parser, complete_machines)
|
||||||
|
|
||||||
parser.add_argument(
|
parser.add_argument(
|
||||||
"--target-host",
|
"--target-host",
|
||||||
type=str,
|
type=str,
|
||||||
|
Loading…
Reference in New Issue
Block a user