File: cli.py

package info (click to toggle)
zabbix-cli 3.5.2-2
  • links: PTS, VCS
  • area: main
  • in suites: forky, sid
  • size: 2,892 kB
  • sloc: python: 18,211; makefile: 3
file content (398 lines) | stat: -rw-r--r-- 13,055 bytes parent folder | download | duplicates (2)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
"""Commands that interact with the application itself."""

from __future__ import annotations

from enum import Enum
from pathlib import Path
from typing import TYPE_CHECKING
from typing import Optional

import typer
from click import Command

from zabbix_cli.app import app
from zabbix_cli.commands.common.args import OPTION_LIMIT
from zabbix_cli.config.constants import SecretMode
from zabbix_cli.dirs import CONFIG_DIR
from zabbix_cli.dirs import DATA_DIR
from zabbix_cli.dirs import EXPORT_DIR
from zabbix_cli.dirs import LOGS_DIR
from zabbix_cli.dirs import SITE_CONFIG_DIR
from zabbix_cli.exceptions import ConfigExistsError
from zabbix_cli.exceptions import ZabbixCLIError
from zabbix_cli.output.console import exit_err
from zabbix_cli.output.console import info
from zabbix_cli.output.console import print_path
from zabbix_cli.output.console import print_toml
from zabbix_cli.output.console import success
from zabbix_cli.output.formatting.path import path_link
from zabbix_cli.output.render import render_result
from zabbix_cli.utils.fs import open_directory

if TYPE_CHECKING:
    from zabbix_cli.config.model import Config


HELP_PANEL = "CLI"


class DirectoryType(Enum):
    """Directory types."""

    CONFIG = "config"
    DATA = "data"
    LOGS = "logs"
    SITE_CONFIG = "siteconfig"
    EXPORTS = "exports"

    def as_path(self) -> Path:
        DIR_MAP = {
            DirectoryType.CONFIG: CONFIG_DIR,
            DirectoryType.DATA: DATA_DIR,
            DirectoryType.LOGS: LOGS_DIR,
            DirectoryType.SITE_CONFIG: SITE_CONFIG_DIR,
            DirectoryType.EXPORTS: EXPORT_DIR,
        }
        d = DIR_MAP.get(self)
        if d is None:
            raise ZabbixCLIError(f"No default path available for {self!r}.")
        return d


def get_directory(directory_type: DirectoryType, config: Optional[Config]) -> Path:
    if config:
        if directory_type == DirectoryType.CONFIG and config.config_path:
            return config.config_path.parent
        elif directory_type == DirectoryType.EXPORTS:
            return config.app.export_directory
    return directory_type.as_path()


@app.command("debug", hidden=True, rich_help_panel=HELP_PANEL)
def debug_cmd(
    ctx: typer.Context,
    with_auth: bool = typer.Option(
        False, "--auth", help="Include auth token in the result."
    ),
) -> None:
    """Print debug info."""
    from zabbix_cli.commands.results.cli import DebugInfo

    render_result(DebugInfo.from_debug_data(app.state, with_auth=with_auth))


@app.command("help", rich_help_panel=HELP_PANEL)
def help(
    ctx: typer.Context,
    command: Optional[Command] = typer.Argument(None, help="Command name"),
) -> None:
    """Show help for a commmand"""
    # TODO: patch get_help() to make it return a string instead of magically
    # printing to stdout, which we have no control over.
    if not command:
        ctx.find_root().get_help()
        return

    # HACK: Set the info name to the resolved command name, otherwise
    # when we call get_help, it will use the name of the help command
    # instead of the resolved command name.
    # Maybe we can use make_context for this?
    ctx.info_name = command.name
    command.get_help(ctx)


@app.command("init", rich_help_panel=HELP_PANEL)
def init(
    ctx: typer.Context,
    config_file: Optional[Path] = typer.Option(
        None, "--config-file", "-c", help="Location of the config file."
    ),
    overwrite: bool = typer.Option(
        False, "--overwrite", help="Overwrite existing config"
    ),
    url: Optional[str] = typer.Option(
        None, "--url", "-u", help="Zabbix API URL to use."
    ),
) -> None:
    """Create and initialize config file."""
    from zabbix_cli.config.utils import init_config

    try:
        init_config(config_file=config_file, overwrite=overwrite, url=url)
    except ConfigExistsError as e:
        raise ZabbixCLIError(f"{e}. Use [option]--overwrite[/] to overwrite it") from e


@app.command(name="login", rich_help_panel=HELP_PANEL)
def login(
    ctx: typer.Context,
    username: Optional[str] = typer.Option(
        None,
        "--username",
        "-u",
        help="Username to log in with.",
        show_default=False,
    ),
    password: Optional[str] = typer.Option(
        None,
        "--password",
        "-p",
        help="Password to log in with.",
        show_default=False,
    ),
    token: Optional[str] = typer.Option(
        None,
        "--token",
        "-t",
        help="API token to log in with.",
        show_default=False,
    ),
) -> None:
    """Trigger a new login prompt in the REPL.

    Ends the current session before triggering a new login prompt.

    Triggers a login prompt if [option]--username[/] & [option]--password[/] or [option]--token[/]
    is not provided.


    Creates a new auth token file if enabled in the config.
    """
    from pydantic import SecretStr

    from zabbix_cli.auth import Authenticator

    if not app.state.repl:
        raise ZabbixCLIError("This command is only available in the REPL.")

    # Copy config to avoid invalid/incomplete state if login fails
    config = app.state.config.model_copy(deep=True)

    # Set auth info in the config to use as defaults in prompts
    config.api.username = username or ""
    config.api.auth_token = SecretStr(token or "")
    config.api.password = SecretStr(password or "")

    # Try to end current session
    # We might not have an active session, which means any attempt to
    # terminate it will result in an error. Catch and log it.
    try:
        app.state.logout()
    except ZabbixCLIError as e:
        app.logger.warning(f"Failed to terminate session in login command: {e}")

    if username and password:
        client, info = Authenticator.login_with_username_password(
            config, username, password
        )
        success(f"Logged in to {config.api.url} as {info.credentials.username}.")
    elif token:
        client, info = Authenticator.login_with_token(config, token=token)
        success(f"Logged in to {config.api.url} with token.")
    else:
        client, info = Authenticator.login_with_prompt(config)
        success(f"Logged in to {config.api.url} as {info.credentials.username}.")

    app.state.client = client
    app.state.config = config


@app.command("migrate_config", rich_help_panel=HELP_PANEL)
def migrate_config(
    ctx: typer.Context,
    source: Optional[Path] = typer.Option(
        None, "--source", "-s", help="Location of the config file to migrate."
    ),
    destination: Optional[Path] = typer.Option(
        None,
        "--destination",
        "-d",
        help="Path of the new config file to create. Uses the default config path if not specified.",
    ),
    overwrite: bool = typer.Option(
        False, "--overwrite", help="Overwrite destination config file if it exists."
    ),
    legacy_json: bool = typer.Option(
        False,
        "--legacy-json-format",
        help="Use legacy JSON format mode in the new config file.",
    ),
) -> None:
    """Migrate a legacy .conf config to a new .toml config.

    The new config file will be created in the default location if no destination is specified.
    The new config enables the new JSON format by default.
    """
    from zabbix_cli.config.constants import DEFAULT_CONFIG_FILE
    from zabbix_cli.config.model import Config

    if source:
        conf = Config.from_file(source)
    else:
        if not app.state.is_config_loaded:
            # this should never happen!
            exit_err(
                "Application was unable to load a config. Use [option]--source[/] to specify one."
            )
        conf = app.state.config

    if not conf.app.is_legacy:
        exit_err(
            "Unable to detect legacy config file. Use [option]--source[/] to specify one."
        )

    if not destination:
        destination = DEFAULT_CONFIG_FILE
    if not destination.suffix == ".toml":
        destination = destination.with_suffix(".toml")

    if destination.exists() and not overwrite:
        exit_err(
            f"File {destination} already exists. Use [option]--overwrite[/] to overwrite it."
        )

    # Set the legacy JSON format flag in the new file
    # By default, we move users over to the new format.
    conf.app.legacy_json_format = legacy_json

    conf.dump_to_file(destination)
    success(f"Config migrated to {destination}")


@app.command("open", rich_help_panel=HELP_PANEL)
def open_config_dir(
    ctx: typer.Context,
    directory_type: DirectoryType = typer.Argument(
        help="The type of directory to open.",
        case_sensitive=False,
        show_default=False,
    ),
    force: bool = typer.Option(
        False,
        "--force",
        help="LINUX: Try to open with [option]--command[/] even if no window manager is detected.",
    ),
    path: bool = typer.Option(
        False,
        "--path",
        help="Show path instead of opening directory.",
    ),
    open_command: Optional[str] = typer.Option(
        None,
        "--command",
        help="Specify command to use to use for opening.",
    ),
) -> None:
    """Open an app directory in the system's file manager.

    Use --force to attempt to open when no DISPLAY env var is set.
    """
    # Try to load the config, but don't fail if it's not available
    try:
        config = app.state.config
    except ZabbixCLIError:
        config = None

    directory = get_directory(directory_type, config)
    if path:
        print_path(directory)
    else:
        open_directory(directory, command=open_command, force=force)


@app.command("sample_config", rich_help_panel=HELP_PANEL)
def sample_config(ctx: typer.Context) -> None:
    """Print a sample configuration file."""
    # Load the entire history, then limit afterwards
    from zabbix_cli.config.model import Config

    conf = Config.sample_config()
    print_toml(conf.as_toml())


@app.command(
    "show_zabbixcli_config", rich_help_panel=HELP_PANEL, hidden=True, deprecated=True
)
@app.command("show_config", rich_help_panel=HELP_PANEL)
def show_config(
    ctx: typer.Context,
    secrets: SecretMode = typer.Option(
        SecretMode.MASK, "--secrets", help="Display mode for secrets."
    ),
) -> None:
    """Show the current application configuration."""
    config = app.state.config
    print_toml(config.as_toml(secrets=secrets))
    if config.config_path:
        info(f"Config file: {config.config_path.absolute()}")


@app.command("show_dirs", rich_help_panel=HELP_PANEL)
def show_directories(ctx: typer.Context) -> None:
    """Show the default directories used by the application."""
    from zabbix_cli.commands.results.cli import DirectoriesResult

    result = DirectoriesResult.from_directory_types(list(DirectoryType))
    render_result(result)


@app.command("show_history", rich_help_panel=HELP_PANEL)
def show_history(
    ctx: typer.Context,
    limit: int = OPTION_LIMIT,
    # TODO: Add --session option to limit to current session
    # In order to add that, we need to store the history len at the start of the session
) -> None:
    """Show the command history."""
    # Load the entire history, then limit afterwards
    from zabbix_cli.commands.results.cli import HistoryResult

    history = list(app.state.history.get_strings())
    history = history[-limit:]
    render_result(HistoryResult(commands=history))


@app.command("update_config", rich_help_panel=HELP_PANEL)
def update_config(
    ctx: typer.Context,
    config_file: Optional[Path] = typer.Option(
        None, "--config-file", "-c", help="Location of the config file to update."
    ),
    secrets: SecretMode = typer.Option(
        SecretMode.PLAIN, "--secrets", help="Visibility mode for secrets."
    ),
    force: bool = typer.Option(False, "--force", help="Skip confirmation prompt."),
) -> None:
    """Update the config file with the current application state.

    Adds missing fields and updates deprecated fields to their new values.
    """
    from zabbix_cli.output.prompts import bool_prompt

    config_file = config_file or app.state.config.config_path
    if not config_file:
        exit_err("No config file specified and no config loaded.")
    if not force:
        if not bool_prompt("Update config file?", default=False):
            exit_err("Update cancelled.")

    config = app.state.config
    config.dump_to_file(config_file, secrets=secrets)
    success(f"Config saved to {path_link(config_file)}")


@app.command("update", rich_help_panel=HELP_PANEL, hidden=True)
def update_application(ctx: typer.Context) -> None:
    """Update the application to the latest version.

    Primarily intended for use with PyInstaller builds, but can also be
    used for updating other installations (except Homebrew).
    """
    from zabbix_cli.__about__ import __version__
    from zabbix_cli.update import update

    info = update()
    if info and info.version:
        success(f"Application updated from {__version__} to {info.version}")
    else:
        success("Application updated.")