glances.outputs.glances_restful_api   F
last analyzed

Complexity

Total Complexity 114

Size/Duplication

Total Lines 978
Duplicated Lines 0 %

Importance

Changes 0
Metric Value
eloc 442
dl 0
loc 978
rs 2
c 0
b 0
f 0
wmc 114

46 Methods

Rating   Name   Duplication   Size   Complexity  
A GlancesUvicornServer.run_in_thread() 0 18 4
A GlancesUvicornServer.install_signal_handlers() 0 2 1
A GlancesJSONResponse.render() 0 2 1
A GlancesRestfulApi.load_config() 0 13 4
A GlancesRestfulApi.__update_servers_list() 0 5 3
A GlancesRestfulApi.authentication() 0 10 3
A GlancesRestfulApi.__update_stats() 0 5 2
B GlancesRestfulApi.__init__() 0 74 5
A GlancesRestfulApi._index() 0 15 1
A GlancesRestfulApi._api_status() 0 10 1
A GlancesRestfulApi.end() 0 5 3
B GlancesRestfulApi._router() 0 93 5
A GlancesRestfulApi.start() 0 17 2
A GlancesRestfulApi._start_uvicorn() 0 14 5
A GlancesRestfulApi._api_plugins() 0 35 2
A GlancesRestfulApi._api_help() 0 11 2
A GlancesRestfulApi._api_servers_list() 0 10 2
A GlancesRestfulApi._events_clear_warning() 0 9 1
A GlancesRestfulApi._browser() 0 9 1
A GlancesRestfulApi._events_clear_all() 0 9 1
A GlancesRestfulApi._api_args() 0 16 2
A GlancesRestfulApi._api_value() 0 23 3
A GlancesRestfulApi._api_item_views() 0 23 2
A GlancesRestfulApi._api_args_item() 0 20 3
A GlancesRestfulApi._api_all_views() 0 15 2
A GlancesRestfulApi._api_all_limits() 0 15 2
A GlancesRestfulApi._api_item_unit() 0 17 3
A GlancesRestfulApi._api_config_section_item() 0 27 4
A GlancesRestfulApi._api_set_extended_processes() 0 16 2
A GlancesRestfulApi._api_item() 0 24 2
A GlancesRestfulApi._api_config() 0 14 3
A GlancesRestfulApi._api_top() 0 26 3
A GlancesRestfulApi._api_limits() 0 17 2
A GlancesRestfulApi._api_views() 0 20 3
A GlancesRestfulApi._api_get_extended_processes() 0 14 2
A GlancesRestfulApi._api_get_processes() 0 14 2
A GlancesRestfulApi._api_config_section() 0 19 3
A GlancesRestfulApi._api_disable_extended_processes() 0 11 1
A GlancesRestfulApi._api_key() 0 24 2
A GlancesRestfulApi._api_history() 0 21 2
A GlancesRestfulApi._check_if_plugin_available() 0 6 2
A GlancesRestfulApi._api_key_views() 0 23 2
A GlancesRestfulApi._api_item_history() 0 21 3
A GlancesRestfulApi._api_item_description() 0 19 3
A GlancesRestfulApi._api() 0 20 2
A GlancesRestfulApi._api_all() 0 27 5

How to fix   Complexity   

Complexity

Complex classes like glances.outputs.glances_restful_api often do a lot of different things. To break such a class down, we need to identify a cohesive component within that class. A common approach to find such a component is to look for fields/methods that share the same prefixes, or suffixes.

Once you have determined the fields that belong together, you can apply the Extract Class refactoring. If the component makes sense as a sub-class, Extract Subclass is also a candidate, and is often faster.

1
#
2
# This file is part of Glances.
3
#
4
# SPDX-FileCopyrightText: 2024 Nicolas Hennion <[email protected]>
5
#
6
# SPDX-License-Identifier: LGPL-3.0-only
7
#
8
9
"""RestFul API interface class."""
10
11
import os
12
import socket
13
import sys
14
import tempfile
15
import webbrowser
16
from typing import Annotated, Any, Union
17
from urllib.parse import urljoin
18
19
from glances import __apiversion__, __version__
20
from glances.events_list import glances_events
21
from glances.globals import json_dumps
22
from glances.logger import logger
23
from glances.password import GlancesPassword
24
from glances.processes import glances_processes
25
from glances.servers_list import GlancesServersList
26
from glances.servers_list_dynamic import GlancesAutoDiscoverClient
27
from glances.stats import GlancesStats
28
from glances.timer import Timer
29
30
# FastAPI import
31
try:
32
    from fastapi import APIRouter, Depends, FastAPI, HTTPException, Request, status
33
    from fastapi.middleware.cors import CORSMiddleware
34
    from fastapi.middleware.gzip import GZipMiddleware
35
    from fastapi.responses import HTMLResponse, JSONResponse
36
    from fastapi.security import HTTPBasic, HTTPBasicCredentials
37
    from fastapi.staticfiles import StaticFiles
38
    from fastapi.templating import Jinja2Templates
39
except ImportError as e:
40
    logger.critical(f'FastAPI import error: {e}')
41
    logger.critical('Glances cannot start in web server mode.')
42
    sys.exit(2)
43
44
try:
45
    import uvicorn
46
except ImportError:
47
    logger.critical('Uvicorn import error. Glances cannot start in web server mode.')
48
    sys.exit(2)
49
import builtins
50
import contextlib
51
import threading
52
import time
53
54
security = HTTPBasic()
55
56
57
class GlancesJSONResponse(JSONResponse):
58
    """
59
    Glances impl of fastapi's JSONResponse to use internal JSON Serialization features
60
61
    Ref: https://fastapi.tiangolo.com/advanced/custom-response/
62
    """
63
64
    def render(self, content: Any) -> bytes:
65
        return json_dumps(content)
66
67
68
class GlancesUvicornServer(uvicorn.Server):
69
    def install_signal_handlers(self):
70
        pass
71
72
    @contextlib.contextmanager
73
    def run_in_thread(self, timeout=3):
74
        thread = threading.Thread(target=self.run)
75
        thread.start()
76
        try:
77
            chrono = Timer(timeout)
78
            while not self.started and not chrono.finished():
79
                time.sleep(0.5)
80
            # Timeout reached
81
            # Something go wrong...
82
            # The Uvicorn server should be stopped
83
            if not self.started:
84
                self.should_exit = True
85
                thread.join()
86
            yield
87
        finally:
88
            self.should_exit = True
89
            thread.join()
90
91
92
class GlancesRestfulApi:
93
    """This class manages the Restful API server."""
94
95
    API_VERSION = __apiversion__
96
97
    def __init__(self, config=None, args=None):
98
        # Init config
99
        self.config = config
100
101
        # Init args
102
        self.args = args
103
104
        # Init stats
105
        # Will be updated within route
106
        self.stats = None
107
108
        # Init servers list (only for the browser mode)
109
        if self.args.browser:
110
            self.servers_list = GlancesServersList(config=config, args=args)
111
        else:
112
            self.servers_list = None
113
114
        # cached_time is the minimum time interval between stats updates
115
        # i.e. HTTP/RESTful calls will not retrieve updated info until the time
116
        # since last update is passed (will retrieve old cached info instead)
117
        self.timer = Timer(0)
118
119
        # Load configuration file
120
        self.load_config(config)
121
122
        # Set the bind URL
123
        self.bind_url = urljoin(f'http://{self.args.bind_address}:{self.args.port}/', self.url_prefix)
124
125
        # FastAPI Init
126
        if self.args.password:
127
            self._app = FastAPI(default_response_class=GlancesJSONResponse, dependencies=[Depends(self.authentication)])
128
            self._password = GlancesPassword(username=args.username, config=config)
129
130
        else:
131
            self._app = FastAPI(default_response_class=GlancesJSONResponse)
132
            self._password = None
133
134
        # Set path for WebUI
135
        webui_root_path = config.get_value(
136
            'outputs', 'webui_root_path', default=os.path.dirname(os.path.realpath(__file__))
137
        )
138
        if webui_root_path == '':
139
            webui_root_path = os.path.dirname(os.path.realpath(__file__))
140
        self.STATIC_PATH = os.path.join(webui_root_path, 'static/public')
141
        self.TEMPLATE_PATH = os.path.join(webui_root_path, 'static/templates')
142
        self._templates = Jinja2Templates(directory=self.TEMPLATE_PATH)
143
144
        # FastAPI Enable GZIP compression
145
        # https://fastapi.tiangolo.com/advanced/middleware/
146
        # Should be done before other middlewares to avoid
147
        # LocalProtocolError("Too much data for declared Content-Length")
148
        self._app.add_middleware(GZipMiddleware, minimum_size=1000)
149
150
        # FastAPI Enable CORS
151
        # https://fastapi.tiangolo.com/tutorial/cors/
152
        self._app.add_middleware(
153
            CORSMiddleware,
154
            # Related to https://github.com/nicolargo/glances/issues/2812
155
            allow_origins=config.get_list_value('outputs', 'cors_origins', default=["*"]),
156
            allow_credentials=config.get_bool_value('outputs', 'cors_credentials', default=True),
157
            allow_methods=config.get_list_value('outputs', 'cors_methods', default=["*"]),
158
            allow_headers=config.get_list_value('outputs', 'cors_headers', default=["*"]),
159
        )
160
161
        # FastAPI Define routes
162
        self._app.include_router(self._router())
163
164
        # Enable auto discovering of the service
165
        self.autodiscover_client = None
166
        if not self.args.disable_autodiscover:
167
            logger.info('Autodiscover is enabled with service name {}'.format(socket.gethostname().split('.', 1)[0]))
168
            self.autodiscover_client = GlancesAutoDiscoverClient(socket.gethostname().split('.', 1)[0], self.args)
169
        else:
170
            logger.info("Glances autodiscover announce is disabled")
171
172
    def load_config(self, config):
173
        """Load the outputs section of the configuration file."""
174
        # Limit the number of processes to display in the WebUI
175
        self.url_prefix = ''
176
        if config is not None and config.has_section('outputs'):
177
            # Max process to display in the WebUI
178
            n = config.get_value('outputs', 'max_processes_display', default=None)
179
            logger.debug(f'Number of processes to display in the WebUI: {n}')
180
            # URL prefix
181
            self.url_prefix = config.get_value('outputs', 'url_prefix', default='')
182
            if self.url_prefix != '':
183
                self.url_prefix = self.url_prefix.rstrip('/')
184
            logger.debug(f'URL prefix: {self.url_prefix}')
185
186
    def __update_stats(self):
187
        # Never update more than 1 time per cached_time
188
        if self.timer.finished():
189
            self.stats.update()
190
            self.timer = Timer(self.args.cached_time)
191
192
    def __update_servers_list(self):
193
        # Never update more than 1 time per cached_time
194
        if self.timer.finished() and self.servers_list is not None:
195
            self.servers_list.update_servers_stats()
196
            self.timer = Timer(self.args.cached_time)
197
198
    def authentication(self, creds: Annotated[HTTPBasicCredentials, Depends(security)]):
199
        """Check if a username/password combination is valid."""
200
        if creds.username == self.args.username:
201
            # check_password
202
            if self._password.check_password(self.args.password, self._password.get_hash(creds.password)):
203
                return creds.username
204
205
        # If the username/password combination is invalid, return an HTTP 401
206
        raise HTTPException(
207
            status.HTTP_401_UNAUTHORIZED, "Incorrect username or password", {"WWW-Authenticate": "Basic"}
208
        )
209
210
    def _router(self) -> APIRouter:
211
        """Define a custom router for Glances path."""
212
        base_path = f'/api/{self.API_VERSION}'
213
        plugin_path = f"{base_path}/{{plugin}}"
214
215
        # Create the main router
216
        router = APIRouter(prefix=self.url_prefix)
217
218
        # REST API route definition
219
        # ==========================
220
221
        # HEAD
222
        router.add_api_route(f'{base_path}/status', self._api_status, methods=['HEAD'])
223
224
        # POST
225
        router.add_api_route(f'{base_path}/events/clear/warning', self._events_clear_warning, methods=['POST'])
226
        router.add_api_route(f'{base_path}/events/clear/all', self._events_clear_all, methods=['POST'])
227
        router.add_api_route(
228
            f'{base_path}/processes/extended/disable', self._api_disable_extended_processes, methods=['POST']
229
        )
230
        router.add_api_route(
231
            f'{base_path}/processes/extended/{{pid}}', self._api_set_extended_processes, methods=['POST']
232
        )
233
234
        # GET
235
        router.add_api_route(f'{base_path}/status', self._api_status, methods=['GET'])
236
        route_mapping = {
237
            f'{base_path}/config': self._api_config,
238
            f'{base_path}/config/{{section}}': self._api_config_section,
239
            f'{base_path}/config/{{section}}/{{item}}': self._api_config_section_item,
240
            f'{base_path}/args': self._api_args,
241
            f'{base_path}/args/{{item}}': self._api_args_item,
242
            f'{base_path}/help': self._api_help,
243
            f'{base_path}/all': self._api_all,
244
            f'{base_path}/all/limits': self._api_all_limits,
245
            f'{base_path}/all/views': self._api_all_views,
246
            f'{base_path}/pluginslist': self._api_plugins,
247
            f'{base_path}/serverslist': self._api_servers_list,
248
            f'{base_path}/processes/extended': self._api_get_extended_processes,
249
            f'{base_path}/processes/{{pid}}': self._api_get_processes,
250
            f'{plugin_path}': self._api,
251
            f'{plugin_path}/history': self._api_history,
252
            f'{plugin_path}/history/{{nb}}': self._api_history,
253
            f'{plugin_path}/top/{{nb}}': self._api_top,
254
            f'{plugin_path}/limits': self._api_limits,
255
            f'{plugin_path}/views': self._api_views,
256
            f'{plugin_path}/{{item}}': self._api_item,
257
            f'{plugin_path}/{{item}}/views': self._api_item_views,
258
            f'{plugin_path}/{{item}}/history': self._api_item_history,
259
            f'{plugin_path}/{{item}}/history/{{nb}}': self._api_item_history,
260
            f'{plugin_path}/{{item}}/description': self._api_item_description,
261
            f'{plugin_path}/{{item}}/unit': self._api_item_unit,
262
            f'{plugin_path}/{{item}}/value/{{value:path}}': self._api_value,
263
            f'{plugin_path}/{{item}}/{{key}}': self._api_key,
264
            f'{plugin_path}/{{item}}/{{key}}/views': self._api_key_views,
265
        }
266
        for path, endpoint in route_mapping.items():
267
            router.add_api_route(path, endpoint)
268
269
        # Browser WEBUI
270
        if hasattr(self.args, 'browser') and self.args.browser:
271
            # Template for the root browser.html file
272
            router.add_api_route('/browser', self._browser, response_class=HTMLResponse)
273
274
            # Statics files
275
            self._app.mount(self.url_prefix + '/static', StaticFiles(directory=self.STATIC_PATH), name="static")
276
            logger.debug(f"The Browser WebUI is enable and got statics files in {self.STATIC_PATH}")
277
278
            bindmsg = f'Glances Browser Web User Interface started on {self.bind_url}browser'
279
            logger.info(bindmsg)
280
            print(bindmsg)
281
282
        # WEBUI
283
        if not self.args.disable_webui:
284
            # Template for the root index.html file
285
            router.add_api_route('/', self._index, response_class=HTMLResponse)
286
287
            # Statics files
288
            self._app.mount(self.url_prefix + '/static', StaticFiles(directory=self.STATIC_PATH), name="static")
289
            logger.debug(f"The WebUI is enable and got statics files in {self.STATIC_PATH}")
290
291
            bindmsg = f'Glances Web User Interface started on {self.bind_url}'
292
            logger.info(bindmsg)
293
            print(bindmsg)
294
        else:
295
            logger.info('The WebUI is disable (--disable-webui)')
296
297
        # Restful API
298
        bindmsg = f'Glances RESTful API Server started on {self.bind_url}api/{self.API_VERSION}'
299
        logger.info(bindmsg)
300
        print(bindmsg)
301
302
        return router
303
304
    def start(self, stats: GlancesStats) -> None:
305
        """Start the bottle."""
306
        # Init stats
307
        self.stats = stats
308
309
        # Init plugin list
310
        self.plugins_list = self.stats.getPluginsList()
311
312
        if self.args.open_web_browser:
313
            # Implementation of the issue #946
314
            # Try to open the Glances Web UI in the default Web browser if:
315
            # 1) --open-web-browser option is used
316
            # 2) Glances standalone mode is running on Windows OS
317
            webbrowser.open(self.bind_url, new=2, autoraise=1)
318
319
        # Start Uvicorn server
320
        self._start_uvicorn()
321
322
    def _start_uvicorn(self):
323
        # Run the Uvicorn Web server
324
        uvicorn_config = uvicorn.Config(
325
            self._app, host=self.args.bind_address, port=self.args.port, access_log=self.args.debug
326
        )
327
        try:
328
            self.uvicorn_server = GlancesUvicornServer(config=uvicorn_config)
329
        except Exception as e:
330
            logger.critical(f'Error: Can not ran Glances Web server ({e})')
331
            self.uvicorn_server = None
332
        else:
333
            with self.uvicorn_server.run_in_thread():
334
                while not self.uvicorn_server.should_exit:
335
                    time.sleep(1)
336
337
    def end(self):
338
        """End the Web server"""
339
        if not self.args.disable_autodiscover and self.autodiscover_client:
340
            self.autodiscover_client.close()
341
        logger.info("Close the Web server")
342
343
    def _index(self, request: Request):
344
        """Return main index.html (/) file.
345
346
        Parameters are available through the request object.
347
        Example: http://localhost:61208/?refresh=5
348
349
        Note: This function is only called the first time the page is loaded.
350
        """
351
        refresh_time = request.query_params.get('refresh', default=max(1, int(self.args.time)))
352
353
        # Update the stat
354
        self.__update_stats()
355
356
        # Display
357
        return self._templates.TemplateResponse("index.html", {"request": request, "refresh_time": refresh_time})
358
359
    def _browser(self, request: Request):
360
        """Return main browser.html (/browser) file.
361
362
        Note: This function is only called the first time the page is loaded.
363
        """
364
        refresh_time = request.query_params.get('refresh', default=max(1, int(self.args.time)))
365
366
        # Display
367
        return self._templates.TemplateResponse("browser.html", {"request": request, "refresh_time": refresh_time})
368
369
    def _api_status(self):
370
        """Glances API RESTful implementation.
371
372
        Return a 200 status code.
373
        This entry point should be used to check the API health.
374
375
        See related issue:  Web server health check endpoint #1988
376
        """
377
378
        return GlancesJSONResponse({'version': __version__})
379
380
    def _events_clear_warning(self):
381
        """Glances API RESTful implementation.
382
383
        Return a 200 status code.
384
385
        It's a post message to clean warning events
386
        """
387
        glances_events.clean()
388
        return GlancesJSONResponse({})
389
390
    def _events_clear_all(self):
391
        """Glances API RESTful implementation.
392
393
        Return a 200 status code.
394
395
        It's a post message to clean all events
396
        """
397
        glances_events.clean(critical=True)
398
        return GlancesJSONResponse({})
399
400
    def _api_help(self):
401
        """Glances API RESTful implementation.
402
403
        Return the help data or 404 error.
404
        """
405
        try:
406
            plist = self.stats.get_plugin("help").get_view_data()
407
        except Exception as e:
408
            raise HTTPException(status.HTTP_404_NOT_FOUND, f"Cannot get help view data ({str(e)})")
409
410
        return GlancesJSONResponse(plist)
411
412
    def _api_plugins(self):
413
        """Glances API RESTFul implementation.
414
415
        @api {get} /api/%s/pluginslist Get plugins list
416
        @apiVersion 2.0
417
        @apiName pluginslist
418
        @apiGroup plugin
419
420
        @apiSuccess {String[]} Plugins list.
421
422
        @apiSuccessExample Success-Response:
423
            HTTP/1.1 200 OK
424
            [
425
               "load",
426
               "help",
427
               "ip",
428
               "memswap",
429
               "processlist",
430
               ...
431
            ]
432
433
         @apiError Cannot get plugin list.
434
435
         @apiErrorExample Error-Response:
436
            HTTP/1.1 404 Not Found
437
        """
438
        # Update the stat
439
        self.__update_stats()
440
441
        try:
442
            plist = self.plugins_list
443
        except Exception as e:
444
            raise HTTPException(status.HTTP_404_NOT_FOUND, f"Cannot get plugin list ({str(e)})")
445
446
        return GlancesJSONResponse(plist)
447
448
    def _api_servers_list(self):
449
        """Glances API RESTful implementation.
450
451
        Return the JSON representation of the servers list (for browser mode)
452
        HTTP/200 if OK
453
        """
454
        # Update the servers list (and the stats for all the servers)
455
        self.__update_servers_list()
456
457
        return GlancesJSONResponse(self.servers_list.get_servers_list() if self.servers_list else [])
458
459
    # Comment this solve an issue on Home Assistant See #3238
460
    def _api_all(self):
461
        """Glances API RESTful implementation.
462
463
        Return the JSON representation of all the plugins
464
        HTTP/200 if OK
465
        HTTP/400 if plugin is not found
466
        HTTP/404 if others error
467
        """
468
        if self.args.debug:
469
            fname = os.path.join(tempfile.gettempdir(), 'glances-debug.json')
470
            try:
471
                with builtins.open(fname) as f:
472
                    return f.read()
473
            except OSError:
474
                logger.debug(f"Debug file ({fname}) not found")
475
476
        # Update the stat
477
        self.__update_stats()
478
479
        try:
480
            # Get the RAW value of the stat ID
481
            # TODO in #3211: use getAllExportsAsDict instead but break UI for uptime, processlist, others ?
482
            statval = self.stats.getAllAsDict()
483
        except Exception as e:
484
            raise HTTPException(status.HTTP_404_NOT_FOUND, f"Cannot get stats ({str(e)})")
485
486
        return GlancesJSONResponse(statval)
487
488
    def _api_all_limits(self):
489
        """Glances API RESTful implementation.
490
491
        Return the JSON representation of all the plugins limits
492
        HTTP/200 if OK
493
        HTTP/400 if plugin is not found
494
        HTTP/404 if others error
495
        """
496
        try:
497
            # Get the RAW value of the stat limits
498
            limits = self.stats.getAllLimitsAsDict()
499
        except Exception as e:
500
            raise HTTPException(status.HTTP_404_NOT_FOUND, f"Cannot get limits ({str(e)})")
501
502
        return GlancesJSONResponse(limits)
503
504
    def _api_all_views(self):
505
        """Glances API RESTful implementation.
506
507
        Return the JSON representation of all the plugins views
508
        HTTP/200 if OK
509
        HTTP/400 if plugin is not found
510
        HTTP/404 if others error
511
        """
512
        try:
513
            # Get the RAW value of the stat view
514
            limits = self.stats.getAllViewsAsDict()
515
        except Exception as e:
516
            raise HTTPException(status.HTTP_404_NOT_FOUND, f"Cannot get views ({str(e)})")
517
518
        return GlancesJSONResponse(limits)
519
520
    def _api(self, plugin: str):
521
        """Glances API RESTful implementation.
522
523
        Return the JSON representation of a given plugin
524
        HTTP/200 if OK
525
        HTTP/400 if plugin is not found
526
        HTTP/404 if others error
527
        """
528
        self._check_if_plugin_available(plugin)
529
530
        # Update the stat
531
        self.__update_stats()
532
533
        try:
534
            # Get the RAW value of the stat ID
535
            statval = self.stats.get_plugin(plugin).get_raw()
536
        except Exception as e:
537
            raise HTTPException(status.HTTP_404_NOT_FOUND, f"Cannot get plugin {plugin} ({str(e)})")
538
539
        return GlancesJSONResponse(statval)
540
541
    def _check_if_plugin_available(self, plugin: str) -> None:
542
        if plugin in self.plugins_list:
543
            return
544
545
        raise HTTPException(
546
            status.HTTP_400_BAD_REQUEST, f"Unknown plugin {plugin} (available plugins: {self.plugins_list})"
547
        )
548
549
    def _api_top(self, plugin: str, nb: int = 0):
550
        """Glances API RESTful implementation.
551
552
        Return the JSON representation of a given plugin limited to the top nb items.
553
        It is used to reduce the payload of the HTTP response (example: processlist).
554
555
        HTTP/200 if OK
556
        HTTP/400 if plugin is not found
557
        HTTP/404 if others error
558
        """
559
        self._check_if_plugin_available(plugin)
560
561
        # Update the stat
562
        self.__update_stats()
563
564
        try:
565
            # Get the RAW value of the stat ID
566
            # TODO in #3211: use get_export instead but break API
567
            statval = self.stats.get_plugin(plugin).get_raw()
568
        except Exception as e:
569
            raise HTTPException(status.HTTP_404_NOT_FOUND, f"Cannot get plugin {plugin} ({str(e)})")
570
571
        if isinstance(statval, list):
572
            statval = statval[:nb]
573
574
        return GlancesJSONResponse(statval)
575
576
    def _api_history(self, plugin: str, nb: int = 0):
577
        """Glances API RESTful implementation.
578
579
        Return the JSON representation of a given plugin history
580
        Limit to the last nb items (all if nb=0)
581
        HTTP/200 if OK
582
        HTTP/400 if plugin is not found
583
        HTTP/404 if others error
584
        """
585
        self._check_if_plugin_available(plugin)
586
587
        # Update the stat
588
        self.__update_stats()
589
590
        try:
591
            # Get the RAW value of the stat ID
592
            statval = self.stats.get_plugin(plugin).get_raw_history(nb=int(nb))
593
        except Exception as e:
594
            raise HTTPException(status.HTTP_404_NOT_FOUND, f"Cannot get plugin history {plugin} ({str(e)})")
595
596
        return statval
597
598
    def _api_limits(self, plugin: str):
599
        """Glances API RESTful implementation.
600
601
        Return the JSON limits of a given plugin
602
        HTTP/200 if OK
603
        HTTP/400 if plugin is not found
604
        HTTP/404 if others error
605
        """
606
        self._check_if_plugin_available(plugin)
607
608
        try:
609
            # Get the RAW value of the stat limits
610
            ret = self.stats.get_plugin(plugin).get_limits()
611
        except Exception as e:
612
            raise HTTPException(status.HTTP_404_NOT_FOUND, f"Cannot get limits for plugin {plugin} ({str(e)})")
613
614
        return GlancesJSONResponse(ret)
615
616
    def _api_views(self, plugin: str):
617
        """Glances API RESTful implementation.
618
619
        Return the JSON views of a given plugin
620
        HTTP/200 if OK
621
        HTTP/400 if plugin is not found
622
        HTTP/404 if others error
623
        """
624
        if plugin not in self.plugins_list:
625
            raise HTTPException(
626
                status.HTTP_400_BAD_REQUEST, f"Unknown plugin {plugin} (available plugins: {self.plugins_list})"
627
            )
628
629
        try:
630
            # Get the RAW value of the stat views
631
            ret = self.stats.get_plugin(plugin).get_views()
632
        except Exception as e:
633
            raise HTTPException(status.HTTP_404_NOT_FOUND, f"Cannot get views for plugin {plugin} ({str(e)})")
634
635
        return GlancesJSONResponse(ret)
636
637
    def _api_item(self, plugin: str, item: str):
638
        """Glances API RESTful implementation.
639
640
        Return the JSON representation of the couple plugin/item
641
        HTTP/200 if OK
642
        HTTP/400 if plugin is not found
643
        HTTP/404 if others error
644
        """
645
        self._check_if_plugin_available(plugin)
646
647
        # Update the stat
648
        self.__update_stats()
649
650
        try:
651
            # Get the RAW value of the stat views
652
            # TODO in #3211: use a non existing (to be created) get_export_item instead but break API
653
            ret = self.stats.get_plugin(plugin).get_raw_stats_item(item)
654
        except Exception as e:
655
            raise HTTPException(
656
                status.HTTP_404_NOT_FOUND,
657
                f"Cannot get item {item} in plugin {plugin} ({str(e)})",
658
            )
659
660
        return GlancesJSONResponse(ret)
661
662
    def _api_key(self, plugin: str, item: str, key: str):
663
        """Glances API RESTful implementation.
664
665
        Return the JSON representation of  plugin/item/key
666
        HTTP/200 if OK
667
        HTTP/400 if plugin is not found
668
        HTTP/404 if others error
669
        """
670
        self._check_if_plugin_available(plugin)
671
672
        # Update the stat
673
        self.__update_stats()
674
675
        try:
676
            # Get the RAW value of the stat views
677
            # TODO in #3211: use a non existing (to be created) get_export_key instead but break API
678
            ret = self.stats.get_plugin(plugin).get_raw_stats_key(item, key)
679
        except Exception as e:
680
            raise HTTPException(
681
                status.HTTP_404_NOT_FOUND,
682
                f"Cannot get item {item} for key {key} in plugin {plugin} ({str(e)})",
683
            )
684
685
        return GlancesJSONResponse(ret)
686
687
    def _api_item_views(self, plugin: str, item: str):
688
        """Glances API RESTful implementation.
689
690
        Return the JSON view representation of the couple plugin/item
691
        HTTP/200 if OK
692
        HTTP/400 if plugin is not found
693
        HTTP/404 if others error
694
        """
695
        self._check_if_plugin_available(plugin)
696
697
        # Update the stat
698
        self.__update_stats()
699
700
        try:
701
            # Get the RAW value of the stat views
702
            ret = self.stats.get_plugin(plugin).get_views().get(item)
703
        except Exception as e:
704
            raise HTTPException(
705
                status.HTTP_404_NOT_FOUND,
706
                f"Cannot get item {item} in plugin view {plugin} ({str(e)})",
707
            )
708
709
        return GlancesJSONResponse(ret)
710
711
    def _api_key_views(self, plugin: str, item: str, key: str):
712
        """Glances API RESTful implementation.
713
714
        Return the JSON view representation of plugin/item/key
715
        HTTP/200 if OK
716
        HTTP/400 if plugin is not found
717
        HTTP/404 if others error
718
        """
719
        self._check_if_plugin_available(plugin)
720
721
        # Update the stat
722
        self.__update_stats()
723
724
        try:
725
            # Get the RAW value of the stat views
726
            ret = self.stats.get_plugin(plugin).get_views().get(key).get(item)
727
        except Exception as e:
728
            raise HTTPException(
729
                status.HTTP_404_NOT_FOUND,
730
                f"Cannot get item {item} for key {key} in plugin view {plugin} ({str(e)})",
731
            )
732
733
        return GlancesJSONResponse(ret)
734
735
    def _api_item_history(self, plugin: str, item: str, nb: int = 0):
736
        """Glances API RESTful implementation.
737
738
        Return the JSON representation of the couple plugin/history of item
739
        HTTP/200 if OK
740
        HTTP/400 if plugin is not found
741
        HTTP/404 if others error
742
743
        """
744
        self._check_if_plugin_available(plugin)
745
746
        # Update the stat
747
        self.__update_stats()
748
749
        try:
750
            # Get the RAW value of the stat history
751
            ret = self.stats.get_plugin(plugin).get_raw_history(item, nb=nb)
752
        except Exception as e:
753
            raise HTTPException(status.HTTP_404_NOT_FOUND, f"Cannot get history for plugin {plugin} ({str(e)})")
754
        else:
755
            return GlancesJSONResponse(ret)
756
757
    def _api_item_description(self, plugin: str, item: str):
758
        """Glances API RESTful implementation.
759
760
        Return the JSON representation of the couple plugin/item description
761
        HTTP/200 if OK
762
        HTTP/400 if plugin is not found
763
        HTTP/404 if others error
764
        """
765
        self._check_if_plugin_available(plugin)
766
767
        try:
768
            # Get the description
769
            ret = self.stats.get_plugin(plugin).get_item_info(item, 'description')
770
        except Exception as e:
771
            raise HTTPException(
772
                status.HTTP_404_NOT_FOUND, f"Cannot get {item} description for plugin {plugin} ({str(e)})"
773
            )
774
        else:
775
            return GlancesJSONResponse(ret)
776
777
    def _api_item_unit(self, plugin: str, item: str):
778
        """Glances API RESTful implementation.
779
780
        Return the JSON representation of the couple plugin/item unit
781
        HTTP/200 if OK
782
        HTTP/400 if plugin is not found
783
        HTTP/404 if others error
784
        """
785
        self._check_if_plugin_available(plugin)
786
787
        try:
788
            # Get the unit
789
            ret = self.stats.get_plugin(plugin).get_item_info(item, 'unit')
790
        except Exception as e:
791
            raise HTTPException(status.HTTP_404_NOT_FOUND, f"Cannot get {item} unit for plugin {plugin} ({str(e)})")
792
        else:
793
            return GlancesJSONResponse(ret)
794
795
    def _api_value(self, plugin: str, item: str, value: Union[str, int, float]):
796
        """Glances API RESTful implementation.
797
798
        Return the process stats (dict) for the given item=value
799
        HTTP/200 if OK
800
        HTTP/400 if plugin is not found
801
        HTTP/404 if others error
802
        """
803
        self._check_if_plugin_available(plugin)
804
805
        # Update the stat
806
        self.__update_stats()
807
808
        try:
809
            # Get the RAW value
810
            # TODO in #3211: use a non existing (to be created) get_export_item_value instead but break API
811
            ret = self.stats.get_plugin(plugin).get_raw_stats_value(item, value)
812
        except Exception as e:
813
            raise HTTPException(
814
                status.HTTP_404_NOT_FOUND, f"Cannot get {item} = {value} for plugin {plugin} ({str(e)})"
815
            )
816
        else:
817
            return GlancesJSONResponse(ret)
818
819
    def _api_config(self):
820
        """Glances API RESTful implementation.
821
822
        Return the JSON representation of the Glances configuration file
823
        HTTP/200 if OK
824
        HTTP/404 if others error
825
        """
826
        try:
827
            # Get the RAW value of the config' dict
828
            args_json = self.config.as_dict()
829
        except Exception as e:
830
            raise HTTPException(status.HTTP_404_NOT_FOUND, f"Cannot get config ({str(e)})")
831
        else:
832
            return GlancesJSONResponse(args_json)
833
834
    def _api_config_section(self, section: str):
835
        """Glances API RESTful implementation.
836
837
        Return the JSON representation of the Glances configuration section
838
        HTTP/200 if OK
839
        HTTP/400 if item is not found
840
        HTTP/404 if others error
841
        """
842
        config_dict = self.config.as_dict()
843
        if section not in config_dict:
844
            raise HTTPException(status.HTTP_400_BAD_REQUEST, f"Unknown configuration item {section}")
845
846
        try:
847
            # Get the RAW value of the config' dict
848
            ret_section = config_dict[section]
849
        except Exception as e:
850
            raise HTTPException(status.HTTP_404_NOT_FOUND, f"Cannot get config section {section} ({str(e)})")
851
852
        return GlancesJSONResponse(ret_section)
853
854
    def _api_config_section_item(self, section: str, item: str):
855
        """Glances API RESTful implementation.
856
857
        Return the JSON representation of the Glances configuration section/item
858
        HTTP/200 if OK
859
        HTTP/400 if item is not found
860
        HTTP/404 if others error
861
        """
862
        config_dict = self.config.as_dict()
863
        if section not in config_dict:
864
            raise HTTPException(status.HTTP_400_BAD_REQUEST, f"Unknown configuration item {section}")
865
866
        try:
867
            # Get the RAW value of the config' dict section
868
            ret_section = config_dict[section]
869
        except Exception as e:
870
            raise HTTPException(status.HTTP_404_NOT_FOUND, f"Cannot get config section {section} ({str(e)})")
871
872
        try:
873
            # Get the RAW value of the config' dict item
874
            ret_item = ret_section[item]
875
        except Exception as e:
876
            raise HTTPException(
877
                status.HTTP_404_NOT_FOUND, f"Cannot get item {item} in config section {section} ({str(e)})"
878
            )
879
880
        return GlancesJSONResponse(ret_item)
881
882
    def _api_args(self):
883
        """Glances API RESTful implementation.
884
885
        Return the JSON representation of the Glances command line arguments
886
        HTTP/200 if OK
887
        HTTP/404 if others error
888
        """
889
        try:
890
            # Get the RAW value of the args' dict
891
            # Use vars to convert namespace to dict
892
            # Source: https://docs.python.org/%s/library/functions.html#vars
893
            args_json = vars(self.args)
894
        except Exception as e:
895
            raise HTTPException(status.HTTP_404_NOT_FOUND, f"Cannot get args ({str(e)})")
896
897
        return GlancesJSONResponse(args_json)
898
899
    def _api_args_item(self, item: str):
900
        """Glances API RESTful implementation.
901
902
        Return the JSON representation of the Glances command line arguments item
903
        HTTP/200 if OK
904
        HTTP/400 if item is not found
905
        HTTP/404 if others error
906
        """
907
        if item not in self.args:
908
            raise HTTPException(status.HTTP_400_BAD_REQUEST, f"Unknown argument item {item}")
909
910
        try:
911
            # Get the RAW value of the args' dict
912
            # Use vars to convert namespace to dict
913
            # Source: https://docs.python.org/%s/library/functions.html#vars
914
            args_json = vars(self.args)[item]
915
        except Exception as e:
916
            raise HTTPException(status.HTTP_404_NOT_FOUND, f"Cannot get args item ({str(e)})")
917
918
        return GlancesJSONResponse(args_json)
919
920
    def _api_set_extended_processes(self, pid: str):
921
        """Glances API RESTful implementation.
922
923
        Set the extended process stats for the given PID
924
        HTTP/200 if OK
925
        HTTP/400 if PID is not found
926
        HTTP/404 if others error
927
        """
928
        process_stats = glances_processes.get_stats(int(pid))
929
930
        if not process_stats:
931
            raise HTTPException(status.HTTP_404_NOT_FOUND, f"Unknown PID process {pid}")
932
933
        glances_processes.extended_process = process_stats
934
935
        return GlancesJSONResponse(True)
936
937
    def _api_disable_extended_processes(self):
938
        """Glances API RESTful implementation.
939
940
        Disable extended process stats
941
        HTTP/200 if OK
942
        HTTP/400 if PID is not found
943
        HTTP/404 if others error
944
        """
945
        glances_processes.extended_process = None
946
947
        return GlancesJSONResponse(True)
948
949
    def _api_get_extended_processes(self):
950
        """Glances API RESTful implementation.
951
952
        Get the extended process stats (if set before)
953
        HTTP/200 if OK
954
        HTTP/400 if PID is not found
955
        HTTP/404 if others error
956
        """
957
        process_stats = glances_processes.get_extended_stats()
958
959
        if not process_stats:
960
            process_stats = {}
961
962
        return GlancesJSONResponse(process_stats)
963
964
    def _api_get_processes(self, pid: str):
965
        """Glances API RESTful implementation.
966
967
        Get the process stats for the given PID
968
        HTTP/200 if OK
969
        HTTP/400 if PID is not found
970
        HTTP/404 if others error
971
        """
972
        process_stats = glances_processes.get_stats(int(pid))
973
974
        if not process_stats:
975
            raise HTTPException(status.HTTP_404_NOT_FOUND, f"Unknown PID process {pid}")
976
977
        return GlancesJSONResponse(process_stats)
978