glances.outputs.glances_restful_api   F
last analyzed

Complexity

Total Complexity 114

Size/Duplication

Total Lines 985
Duplicated Lines 0 %

Importance

Changes 0
Metric Value
eloc 450
dl 0
loc 985
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._api_args() 0 16 2
A GlancesRestfulApi._api_value() 0 23 3
A GlancesRestfulApi._index() 0 15 1
A GlancesRestfulApi._api_item_views() 0 23 2
A GlancesRestfulApi._api_status() 0 10 1
A GlancesRestfulApi._api_args_item() 0 20 3
A GlancesRestfulApi._api_all_views() 0 16 2
A GlancesRestfulApi._api_all_limits() 0 16 2
A GlancesRestfulApi.end() 0 5 3
B GlancesRestfulApi._router() 0 93 5
A GlancesRestfulApi._api_item_unit() 0 17 3
A GlancesRestfulApi._api_config_section_item() 0 27 4
A GlancesRestfulApi.start() 0 17 2
A GlancesRestfulApi._api_set_extended_processes() 0 16 2
A GlancesRestfulApi._api_item() 0 24 2
A GlancesRestfulApi._start_uvicorn() 0 14 5
A GlancesRestfulApi._api_config() 0 14 3
A GlancesRestfulApi._api_plugins() 0 35 2
A GlancesRestfulApi._api_top() 0 27 3
A GlancesRestfulApi._api_limits() 0 18 2
A GlancesRestfulApi._api_views() 0 21 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_help() 0 11 2
A GlancesRestfulApi._api_disable_extended_processes() 0 11 1
A GlancesRestfulApi._api_servers_list() 0 10 2
A GlancesRestfulApi._api_key() 0 24 2
A GlancesRestfulApi._api_history() 0 22 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._events_clear_warning() 0 9 1
A GlancesRestfulApi._api_item_description() 0 19 3
A GlancesRestfulApi._browser() 0 9 1
A GlancesRestfulApi._events_clear_all() 0 9 1
A GlancesRestfulApi._api() 0 21 2
A GlancesRestfulApi._api_all() 0 28 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
"""RestFull 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, weak_lru_cache
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
    @weak_lru_cache(maxsize=1, ttl=1)
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
    @weak_lru_cache(maxsize=1, ttl=1)
489
    def _api_all_limits(self):
490
        """Glances API RESTful implementation.
491
492
        Return the JSON representation of all the plugins limits
493
        HTTP/200 if OK
494
        HTTP/400 if plugin is not found
495
        HTTP/404 if others error
496
        """
497
        try:
498
            # Get the RAW value of the stat limits
499
            limits = self.stats.getAllLimitsAsDict()
500
        except Exception as e:
501
            raise HTTPException(status.HTTP_404_NOT_FOUND, f"Cannot get limits ({str(e)})")
502
503
        return GlancesJSONResponse(limits)
504
505
    @weak_lru_cache(maxsize=1, ttl=1)
506
    def _api_all_views(self):
507
        """Glances API RESTful implementation.
508
509
        Return the JSON representation of all the plugins views
510
        HTTP/200 if OK
511
        HTTP/400 if plugin is not found
512
        HTTP/404 if others error
513
        """
514
        try:
515
            # Get the RAW value of the stat view
516
            limits = self.stats.getAllViewsAsDict()
517
        except Exception as e:
518
            raise HTTPException(status.HTTP_404_NOT_FOUND, f"Cannot get views ({str(e)})")
519
520
        return GlancesJSONResponse(limits)
521
522
    @weak_lru_cache(maxsize=1, ttl=1)
523
    def _api(self, plugin: str):
524
        """Glances API RESTful implementation.
525
526
        Return the JSON representation of a given plugin
527
        HTTP/200 if OK
528
        HTTP/400 if plugin is not found
529
        HTTP/404 if others error
530
        """
531
        self._check_if_plugin_available(plugin)
532
533
        # Update the stat
534
        self.__update_stats()
535
536
        try:
537
            # Get the RAW value of the stat ID
538
            statval = self.stats.get_plugin(plugin).get_raw()
539
        except Exception as e:
540
            raise HTTPException(status.HTTP_404_NOT_FOUND, f"Cannot get plugin {plugin} ({str(e)})")
541
542
        return GlancesJSONResponse(statval)
543
544
    def _check_if_plugin_available(self, plugin: str) -> None:
545
        if plugin in self.plugins_list:
546
            return
547
548
        raise HTTPException(
549
            status.HTTP_400_BAD_REQUEST, f"Unknown plugin {plugin} (available plugins: {self.plugins_list})"
550
        )
551
552
    @weak_lru_cache(maxsize=1, ttl=1)
553
    def _api_top(self, plugin: str, nb: int = 0):
554
        """Glances API RESTful implementation.
555
556
        Return the JSON representation of a given plugin limited to the top nb items.
557
        It is used to reduce the payload of the HTTP response (example: processlist).
558
559
        HTTP/200 if OK
560
        HTTP/400 if plugin is not found
561
        HTTP/404 if others error
562
        """
563
        self._check_if_plugin_available(plugin)
564
565
        # Update the stat
566
        self.__update_stats()
567
568
        try:
569
            # Get the RAW value of the stat ID
570
            # TODO in #3211: use get_export instead but break API
571
            statval = self.stats.get_plugin(plugin).get_raw()
572
        except Exception as e:
573
            raise HTTPException(status.HTTP_404_NOT_FOUND, f"Cannot get plugin {plugin} ({str(e)})")
574
575
        if isinstance(statval, list):
576
            statval = statval[:nb]
577
578
        return GlancesJSONResponse(statval)
579
580
    @weak_lru_cache(maxsize=1, ttl=1)
581
    def _api_history(self, plugin: str, nb: int = 0):
582
        """Glances API RESTful implementation.
583
584
        Return the JSON representation of a given plugin history
585
        Limit to the last nb items (all if nb=0)
586
        HTTP/200 if OK
587
        HTTP/400 if plugin is not found
588
        HTTP/404 if others error
589
        """
590
        self._check_if_plugin_available(plugin)
591
592
        # Update the stat
593
        self.__update_stats()
594
595
        try:
596
            # Get the RAW value of the stat ID
597
            statval = self.stats.get_plugin(plugin).get_raw_history(nb=int(nb))
598
        except Exception as e:
599
            raise HTTPException(status.HTTP_404_NOT_FOUND, f"Cannot get plugin history {plugin} ({str(e)})")
600
601
        return statval
602
603
    @weak_lru_cache(maxsize=1, ttl=1)
604
    def _api_limits(self, plugin: str):
605
        """Glances API RESTful implementation.
606
607
        Return the JSON limits of a given plugin
608
        HTTP/200 if OK
609
        HTTP/400 if plugin is not found
610
        HTTP/404 if others error
611
        """
612
        self._check_if_plugin_available(plugin)
613
614
        try:
615
            # Get the RAW value of the stat limits
616
            ret = self.stats.get_plugin(plugin).get_limits()
617
        except Exception as e:
618
            raise HTTPException(status.HTTP_404_NOT_FOUND, f"Cannot get limits for plugin {plugin} ({str(e)})")
619
620
        return GlancesJSONResponse(ret)
621
622
    @weak_lru_cache(maxsize=1, ttl=1)
623
    def _api_views(self, plugin: str):
624
        """Glances API RESTful implementation.
625
626
        Return the JSON views of a given plugin
627
        HTTP/200 if OK
628
        HTTP/400 if plugin is not found
629
        HTTP/404 if others error
630
        """
631
        if plugin not in self.plugins_list:
632
            raise HTTPException(
633
                status.HTTP_400_BAD_REQUEST, f"Unknown plugin {plugin} (available plugins: {self.plugins_list})"
634
            )
635
636
        try:
637
            # Get the RAW value of the stat views
638
            ret = self.stats.get_plugin(plugin).get_views()
639
        except Exception as e:
640
            raise HTTPException(status.HTTP_404_NOT_FOUND, f"Cannot get views for plugin {plugin} ({str(e)})")
641
642
        return GlancesJSONResponse(ret)
643
644
    def _api_item(self, plugin: str, item: str):
645
        """Glances API RESTful implementation.
646
647
        Return the JSON representation of the couple plugin/item
648
        HTTP/200 if OK
649
        HTTP/400 if plugin is not found
650
        HTTP/404 if others error
651
        """
652
        self._check_if_plugin_available(plugin)
653
654
        # Update the stat
655
        self.__update_stats()
656
657
        try:
658
            # Get the RAW value of the stat views
659
            # TODO in #3211: use a non existing (to be created) get_export_item instead but break API
660
            ret = self.stats.get_plugin(plugin).get_raw_stats_item(item)
661
        except Exception as e:
662
            raise HTTPException(
663
                status.HTTP_404_NOT_FOUND,
664
                f"Cannot get item {item} in plugin {plugin} ({str(e)})",
665
            )
666
667
        return GlancesJSONResponse(ret)
668
669
    def _api_key(self, plugin: str, item: str, key: str):
670
        """Glances API RESTful implementation.
671
672
        Return the JSON representation of  plugin/item/key
673
        HTTP/200 if OK
674
        HTTP/400 if plugin is not found
675
        HTTP/404 if others error
676
        """
677
        self._check_if_plugin_available(plugin)
678
679
        # Update the stat
680
        self.__update_stats()
681
682
        try:
683
            # Get the RAW value of the stat views
684
            # TODO in #3211: use a non existing (to be created) get_export_key instead but break API
685
            ret = self.stats.get_plugin(plugin).get_raw_stats_key(item, key)
686
        except Exception as e:
687
            raise HTTPException(
688
                status.HTTP_404_NOT_FOUND,
689
                f"Cannot get item {item} for key {key} in plugin {plugin} ({str(e)})",
690
            )
691
692
        return GlancesJSONResponse(ret)
693
694
    def _api_item_views(self, plugin: str, item: str):
695
        """Glances API RESTful implementation.
696
697
        Return the JSON view representation of the couple plugin/item
698
        HTTP/200 if OK
699
        HTTP/400 if plugin is not found
700
        HTTP/404 if others error
701
        """
702
        self._check_if_plugin_available(plugin)
703
704
        # Update the stat
705
        self.__update_stats()
706
707
        try:
708
            # Get the RAW value of the stat views
709
            ret = self.stats.get_plugin(plugin).get_views().get(item)
710
        except Exception as e:
711
            raise HTTPException(
712
                status.HTTP_404_NOT_FOUND,
713
                f"Cannot get item {item} in plugin view {plugin} ({str(e)})",
714
            )
715
716
        return GlancesJSONResponse(ret)
717
718
    def _api_key_views(self, plugin: str, item: str, key: str):
719
        """Glances API RESTful implementation.
720
721
        Return the JSON view representation of plugin/item/key
722
        HTTP/200 if OK
723
        HTTP/400 if plugin is not found
724
        HTTP/404 if others error
725
        """
726
        self._check_if_plugin_available(plugin)
727
728
        # Update the stat
729
        self.__update_stats()
730
731
        try:
732
            # Get the RAW value of the stat views
733
            ret = self.stats.get_plugin(plugin).get_views().get(key).get(item)
734
        except Exception as e:
735
            raise HTTPException(
736
                status.HTTP_404_NOT_FOUND,
737
                f"Cannot get item {item} for key {key} in plugin view {plugin} ({str(e)})",
738
            )
739
740
        return GlancesJSONResponse(ret)
741
742
    def _api_item_history(self, plugin: str, item: str, nb: int = 0):
743
        """Glances API RESTful implementation.
744
745
        Return the JSON representation of the couple plugin/history of item
746
        HTTP/200 if OK
747
        HTTP/400 if plugin is not found
748
        HTTP/404 if others error
749
750
        """
751
        self._check_if_plugin_available(plugin)
752
753
        # Update the stat
754
        self.__update_stats()
755
756
        try:
757
            # Get the RAW value of the stat history
758
            ret = self.stats.get_plugin(plugin).get_raw_history(item, nb=nb)
759
        except Exception as e:
760
            raise HTTPException(status.HTTP_404_NOT_FOUND, f"Cannot get history for plugin {plugin} ({str(e)})")
761
        else:
762
            return GlancesJSONResponse(ret)
763
764
    def _api_item_description(self, plugin: str, item: str):
765
        """Glances API RESTful implementation.
766
767
        Return the JSON representation of the couple plugin/item description
768
        HTTP/200 if OK
769
        HTTP/400 if plugin is not found
770
        HTTP/404 if others error
771
        """
772
        self._check_if_plugin_available(plugin)
773
774
        try:
775
            # Get the description
776
            ret = self.stats.get_plugin(plugin).get_item_info(item, 'description')
777
        except Exception as e:
778
            raise HTTPException(
779
                status.HTTP_404_NOT_FOUND, f"Cannot get {item} description for plugin {plugin} ({str(e)})"
780
            )
781
        else:
782
            return GlancesJSONResponse(ret)
783
784
    def _api_item_unit(self, plugin: str, item: str):
785
        """Glances API RESTful implementation.
786
787
        Return the JSON representation of the couple plugin/item unit
788
        HTTP/200 if OK
789
        HTTP/400 if plugin is not found
790
        HTTP/404 if others error
791
        """
792
        self._check_if_plugin_available(plugin)
793
794
        try:
795
            # Get the unit
796
            ret = self.stats.get_plugin(plugin).get_item_info(item, 'unit')
797
        except Exception as e:
798
            raise HTTPException(status.HTTP_404_NOT_FOUND, f"Cannot get {item} unit for plugin {plugin} ({str(e)})")
799
        else:
800
            return GlancesJSONResponse(ret)
801
802
    def _api_value(self, plugin: str, item: str, value: Union[str, int, float]):
803
        """Glances API RESTful implementation.
804
805
        Return the process stats (dict) for the given item=value
806
        HTTP/200 if OK
807
        HTTP/400 if plugin is not found
808
        HTTP/404 if others error
809
        """
810
        self._check_if_plugin_available(plugin)
811
812
        # Update the stat
813
        self.__update_stats()
814
815
        try:
816
            # Get the RAW value
817
            # TODO in #3211: use a non existing (to be created) get_export_item_value instead but break API
818
            ret = self.stats.get_plugin(plugin).get_raw_stats_value(item, value)
819
        except Exception as e:
820
            raise HTTPException(
821
                status.HTTP_404_NOT_FOUND, f"Cannot get {item} = {value} for plugin {plugin} ({str(e)})"
822
            )
823
        else:
824
            return GlancesJSONResponse(ret)
825
826
    def _api_config(self):
827
        """Glances API RESTful implementation.
828
829
        Return the JSON representation of the Glances configuration file
830
        HTTP/200 if OK
831
        HTTP/404 if others error
832
        """
833
        try:
834
            # Get the RAW value of the config' dict
835
            args_json = self.config.as_dict()
836
        except Exception as e:
837
            raise HTTPException(status.HTTP_404_NOT_FOUND, f"Cannot get config ({str(e)})")
838
        else:
839
            return GlancesJSONResponse(args_json)
840
841
    def _api_config_section(self, section: str):
842
        """Glances API RESTful implementation.
843
844
        Return the JSON representation of the Glances configuration section
845
        HTTP/200 if OK
846
        HTTP/400 if item is not found
847
        HTTP/404 if others error
848
        """
849
        config_dict = self.config.as_dict()
850
        if section not in config_dict:
851
            raise HTTPException(status.HTTP_400_BAD_REQUEST, f"Unknown configuration item {section}")
852
853
        try:
854
            # Get the RAW value of the config' dict
855
            ret_section = config_dict[section]
856
        except Exception as e:
857
            raise HTTPException(status.HTTP_404_NOT_FOUND, f"Cannot get config section {section} ({str(e)})")
858
859
        return GlancesJSONResponse(ret_section)
860
861
    def _api_config_section_item(self, section: str, item: str):
862
        """Glances API RESTful implementation.
863
864
        Return the JSON representation of the Glances configuration section/item
865
        HTTP/200 if OK
866
        HTTP/400 if item is not found
867
        HTTP/404 if others error
868
        """
869
        config_dict = self.config.as_dict()
870
        if section not in config_dict:
871
            raise HTTPException(status.HTTP_400_BAD_REQUEST, f"Unknown configuration item {section}")
872
873
        try:
874
            # Get the RAW value of the config' dict section
875
            ret_section = config_dict[section]
876
        except Exception as e:
877
            raise HTTPException(status.HTTP_404_NOT_FOUND, f"Cannot get config section {section} ({str(e)})")
878
879
        try:
880
            # Get the RAW value of the config' dict item
881
            ret_item = ret_section[item]
882
        except Exception as e:
883
            raise HTTPException(
884
                status.HTTP_404_NOT_FOUND, f"Cannot get item {item} in config section {section} ({str(e)})"
885
            )
886
887
        return GlancesJSONResponse(ret_item)
888
889
    def _api_args(self):
890
        """Glances API RESTful implementation.
891
892
        Return the JSON representation of the Glances command line arguments
893
        HTTP/200 if OK
894
        HTTP/404 if others error
895
        """
896
        try:
897
            # Get the RAW value of the args' dict
898
            # Use vars to convert namespace to dict
899
            # Source: https://docs.python.org/%s/library/functions.html#vars
900
            args_json = vars(self.args)
901
        except Exception as e:
902
            raise HTTPException(status.HTTP_404_NOT_FOUND, f"Cannot get args ({str(e)})")
903
904
        return GlancesJSONResponse(args_json)
905
906
    def _api_args_item(self, item: str):
907
        """Glances API RESTful implementation.
908
909
        Return the JSON representation of the Glances command line arguments item
910
        HTTP/200 if OK
911
        HTTP/400 if item is not found
912
        HTTP/404 if others error
913
        """
914
        if item not in self.args:
915
            raise HTTPException(status.HTTP_400_BAD_REQUEST, f"Unknown argument item {item}")
916
917
        try:
918
            # Get the RAW value of the args' dict
919
            # Use vars to convert namespace to dict
920
            # Source: https://docs.python.org/%s/library/functions.html#vars
921
            args_json = vars(self.args)[item]
922
        except Exception as e:
923
            raise HTTPException(status.HTTP_404_NOT_FOUND, f"Cannot get args item ({str(e)})")
924
925
        return GlancesJSONResponse(args_json)
926
927
    def _api_set_extended_processes(self, pid: str):
928
        """Glances API RESTful implementation.
929
930
        Set the extended process stats for the given PID
931
        HTTP/200 if OK
932
        HTTP/400 if PID is not found
933
        HTTP/404 if others error
934
        """
935
        process_stats = glances_processes.get_stats(int(pid))
936
937
        if not process_stats:
938
            raise HTTPException(status.HTTP_404_NOT_FOUND, f"Unknown PID process {pid}")
939
940
        glances_processes.extended_process = process_stats
941
942
        return GlancesJSONResponse(True)
943
944
    def _api_disable_extended_processes(self):
945
        """Glances API RESTful implementation.
946
947
        Disable extended process stats
948
        HTTP/200 if OK
949
        HTTP/400 if PID is not found
950
        HTTP/404 if others error
951
        """
952
        glances_processes.extended_process = None
953
954
        return GlancesJSONResponse(True)
955
956
    def _api_get_extended_processes(self):
957
        """Glances API RESTful implementation.
958
959
        Get the extended process stats (if set before)
960
        HTTP/200 if OK
961
        HTTP/400 if PID is not found
962
        HTTP/404 if others error
963
        """
964
        process_stats = glances_processes.get_extended_stats()
965
966
        if not process_stats:
967
            process_stats = {}
968
969
        return GlancesJSONResponse(process_stats)
970
971
    def _api_get_processes(self, pid: str):
972
        """Glances API RESTful implementation.
973
974
        Get the process stats for the given PID
975
        HTTP/200 if OK
976
        HTTP/400 if PID is not found
977
        HTTP/404 if others error
978
        """
979
        process_stats = glances_processes.get_stats(int(pid))
980
981
        if not process_stats:
982
            raise HTTPException(status.HTTP_404_NOT_FOUND, f"Unknown PID process {pid}")
983
984
        return GlancesJSONResponse(process_stats)
985