File: state.py

package info (click to toggle)
zabbix-cli 3.5.3-1
  • links: PTS, VCS
  • area: main
  • in suites: forky, sid
  • size: 2,860 kB
  • sloc: python: 18,557; makefile: 3
file content (273 lines) | stat: -rw-r--r-- 9,495 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
"""Defines the global state object for the application.

The global state object is a singleton that holds the current state of the
application. It is used to store the current configuration, Zabbix client,
and other stateful objects.
"""

# This module re-treads all the sins of Harbor CLI's state module.
#
# The lazy-loading, properties with setters/getters, and the singleton pattern
# are all code smells in their own ways (lazy-loading maybe less so in a CLI context).
# The sad thing is that this works well, and it's hard to find a good way to configure
# the application once, save that state, then maintain it for the duration
# of the application's lifetime. The only thing we have to be careful about it
# is to ensure we configure the State object before we use it, and perform
# all imports from other modules inside functions, so that we don't create
# circular imports.
#
# For some reason, I am totally blanking on how to do this in a way that doesn't
# suck when using Typer. With Click, it is easy, because we could just use the
# `@pass_obj` decorator to pass the State object to every command, but Typer
# doesn't have anything like that.
#
# We have a bunch of state we need to access in order to determine things like
# authentication, logging, default values, etc. So having them all gathered
# in this one place is convenient.
from __future__ import annotations

import logging
from typing import TYPE_CHECKING
from typing import Any
from typing import Optional

from typing_extensions import Self

# This module should not import from other local modules because it's widely
# used throughout the application, and we don't want to create circular imports.
# Runtime imports from other modules should be done inside functions,
# while annotations imports should done in `if TYPE_CHECKING:` blocks.
if TYPE_CHECKING:
    from prompt_toolkit.history import History
    from rich.console import Console

    from zabbix_cli.config.model import Config
    from zabbix_cli.pyzabbix.client import ZabbixAPI

logger = logging.getLogger(__name__)


class State:
    """Application state singleton.

    Object that encapsulates the current state of the application.
    Holds the current configuration, Zabbix client, and other stateful objects.
    """

    _instance = None

    def __new__(cls, *args: Any, **kwargs: Any) -> Self:
        """Instantiate the singleton object or return the existing instance."""
        if cls._instance is None:
            cls._instance = super().__new__(cls, *args, **kwargs)
        return cls._instance

    repl: bool = False
    """REPL is active."""

    bulk: bool = False
    """Running in bulk mode."""

    _client: Optional[ZabbixAPI] = None
    """Zabbix API client object."""

    _config: Optional[Config] = None
    """Current Config object (may have overrides)."""

    _config_loaded: bool = False
    """Config has been loaded from file."""

    _config_repl_original: Optional[Config] = None
    """Config object when the REPL was first launched."""

    _console: Optional[Console] = None
    """Stdout Rich console object."""

    _err_console: Optional[Console] = None
    """Stderr Rich console object."""

    token: Optional[str] = None
    """Active Zabbix API auth token."""

    _history: Optional[History] = None

    @property
    def client(self) -> ZabbixAPI:
        """Zabbix API client object.

        Fails if the client is not configured.
        """
        from zabbix_cli.exceptions import ZabbixCLIError

        if self._client is None:
            raise ZabbixCLIError("Not connected to the Zabbix API.")
        return self._client

    @client.setter
    def client(self, client: ZabbixAPI) -> None:
        self._client = client

    @property
    def is_client_loaded(self) -> bool:
        """Zabbix client is configured and ready to use."""
        return self._client is not None

    @property
    def config(self) -> Config:
        """Current Config object."""
        if self._config is None:
            from zabbix_cli.config.model import Config
            from zabbix_cli.logs import configure_logging

            # HACK: configure logging with sample config
            # in order to log the debug message to the default log file
            config = Config.sample_config()
            configure_logging(config.logging)
            logger.debug(
                "Using sample config file as fallback.",
                stacklevel=2,  # See who called this
            )
            return config
        return self._config

    @config.setter
    def config(self, config: Config) -> None:
        """Set the config object and update active configurations.

        Triggers an update of configuration of logging, consoles, etc.
        """
        from zabbix_cli.logs import configure_logging
        from zabbix_cli.output.console import configure_console

        self._config = config
        configure_logging(config.logging)
        configure_console(config)
        # HACK: don't consider using sample config as "loaded".
        # Signals that main callback should create a new config file.
        self._config_loaded = config.sample is False

    @property
    def is_config_loaded(self) -> bool:
        """Config has been loaded from file."""
        return self._config_loaded

    @property
    def console(self) -> Console:
        """Rich console object."""
        # fmt: off
        if not self._console:
            from .output.console import console
            self._console = console
        return self._console
        # fmt: on

    @property
    def err_console(self) -> Console:
        """Rich console object."""
        # fmt: off
        if not self._err_console:
            from .output.console import err_console
            self._err_console = err_console
        return self._err_console
        # fmt: on

    @property
    def history(self) -> History:
        """Prompt history.

        Lazily instantiates the history object if it doesn't exist.
        """
        from prompt_toolkit.history import FileHistory
        from prompt_toolkit.history import InMemoryHistory

        if not self._history:
            if self.config.app.history and self.config.app.history_file:
                try:
                    self._history = FileHistory(str(self.config.app.history_file))
                except Exception as e:
                    from zabbix_cli.output.console import error

                    error(f"Failed to instantiate CLI history: {e}")
                    self._history = InMemoryHistory()
            else:
                self._history = InMemoryHistory()
        return self._history

    @property
    def ready(self) -> bool:
        """State is configured and ready to use."""
        return self.is_client_loaded and self.is_config_loaded

    def revert_config_overrides(self) -> None:
        """Revert config overrides from CLI args applied in REPL.

        Ensures that overrides only apply to a single command invocation,
        and are reset afterwards.

        In REPL mode, we have to ensure overrides don't persist between commands:

        ```
        > show_trigger_events 123 # renders table
        > -o json show_trigger_events 123 # renders json (override applied to config)
        > show_trigger_events 123 # renders table (override reverted)
        ```

        The override is reset after the command is executed.
        """
        if not self.repl or not self.is_config_loaded:
            return
        if not self._config_repl_original:
            self._config_repl_original = self.config.model_copy(deep=True)
        else:
            self.config = self._config_repl_original.model_copy(deep=True)

    def login(self) -> None:
        """Log in to the Zabbix API.

        Uses the authentication info from the config to log into the Zabbix API.

        Also sets the JSON rendering mode on the TableRenderable base class
        used to render application output.
        """
        from zabbix_cli import auth
        from zabbix_cli.models import TableRenderable

        self.client = auth.login(self.config)
        TableRenderable.legacy_json_format = self.config.app.legacy_json_format

    def logout(self) -> None:
        """Log out and clear auth token file if configured."""
        from zabbix_cli import auth

        auth.logout(self.client, self.config)

    def logout_on_exit(self) -> None:
        """Log out on exit if not keeping the session alive via auth token file."""
        # If we are NOT keeping the API session alive between CLI invocations
        # we need to remember to log out once we are done in order to end the
        # session properly.
        # https://www.zabbix.com/documentation/current/en/manual/api/reference/user/login
        if (
            # State did not finish configuration before termination
            not self.ready
            # OR We want to keep the session alive
            or self.config.app.use_session_file
            # OR we are using an API token
            or self.client.use_api_token
        ):
            return
        try:
            self.logout()
        except Exception as e:
            from zabbix_cli.exceptions import handle_exception

            # Outside of main loop, handle the exception
            handle_exception(e)


def get_state() -> State:
    """Returns the global state object.

    Instantiates a new state object with defaults if it doesn't exist.
    """
    return State()