Issues (48)

glances/outputs/glances_restful_api.py (2 issues)

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