kytos.core.config.KytosConfig.options_exposed() - Code Metrics - Inspection of "Merge pull request #380 from kytos-ng/feat/starlet..." - kytos-ng/kytos - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Push — master ( e44493...373da3 )

by Vinicius

created 2023-05-03 17:55 UTC

kytos.core.config.KytosConfig.options_exposed() A

↳ Parent: kytos.core.config

Complexity

Conditions

Size

Total Lines	7
Code Lines	6

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
cc	2
eloc	6
nop	1
dl	0
loc	7
rs	10
c	0
b	0
f	0

"""Here you can control the config parameters to run Kytos controller.

Basically you can use a config file (-c option) and use arguments on command
line. If you specify a config file, then and option configured inside this file
will be overridden by the option on command line.
"""

import json
import os
import uuid
import warnings
from argparse import ArgumentParser, RawDescriptionHelpFormatter
from configparser import ConfigParser
from pathlib import Path

from jinja2 import Template

from kytos.core.metadata import __version__

BASE_ENV = os.environ.get('VIRTUAL_ENV', None) or '/'
ETC_KYTOS = 'etc/kytos'
TEMPLATE_FILES = ['templates/kytos.conf.template',
                  'templates/logging.ini.template']
SYSLOG_ARGS = ['/dev/log'] if Path('/dev/log').exists() else []


class KytosConfig():
    """Handle settings of Kytos."""

    def __init__(self):
        """Parse the command line.

        The contructor set defaults parameters that can be used by KytosConfig.
        """
        self.options = {}
        conf_parser = ArgumentParser(add_help=False)

        conf_parser.add_argument("-c", "--conf",
                                 help="Specify a config file",
                                 metavar="FILE")

        parser = ArgumentParser(prog='kytosd',
                                parents=[conf_parser],
                                formatter_class=RawDescriptionHelpFormatter,
                                description=__doc__)

        parser.add_argument('-v', '--version',
                            action='version',
                            version=f"kytosd {__version__}")

        parser.add_argument('-D', '--debug',
                            action='store_true',
                            help="Run in debug mode")

        parser.add_argument('-f', '--foreground',
                            action='store_true',
                            help="Run in foreground (ctrl+c to stop)")

        parser.add_argument('-l', '--listen',
                            action='store',
                            help="IP/Interface to be listened")

        parser.add_argument('-n', '--napps',
                            action='store',
                            help="Specify the napps directory")

        parser.add_argument('-P', '--port',
                            action='store',
                            help="Port to be listened")

        parser.add_argument('-p', '--pidfile',
                            action='store',
                            help="Specify the PID file to save.")

        parser.add_argument('-w', '--workdir',
                            action='store',
                            help="Specify the working directory")

        parser.add_argument('-s', '--protocol_name',
                            action='store',
                            help="Specify the southbound protocol")

        parser.add_argument('-E', '--enable_entities_by_default',
                            action='store_true',
                            help="Enable all new Entities by default.")

        parser.add_argument('-C', '--create-superuser',
                            action='store_true',
                            help="Create a kytos superuser.")

        parser.add_argument('-t', '--thread_pool_max_workers',
                            action='store',
                            help="Maximum number of threads in the pool.")

        parser.add_argument('-d', '--database',
                            action='store',
                            help="Database backend.")

        parser.add_argument('-a', '--apm',
                            action='store',
                            help="APM backend.")

        self.conf_parser, self.parser = conf_parser, parser
        self.parse_args()

    def parse_args(self):
        """Get the command line options and update kytos settings.

        When installed via pip, defaults values are:

        .. code-block:: python

            defaults = {'pidfile': '/var/run/kytos/kytosd.pid',
                        'workdir': '/var/lib/kytos',
                        'napps': '/var/lib/kytos/napps/',
                        'conf': '/etc/kytos/kytos.conf',
                        'logging': '/etc/kytos/logging.ini',
                        'listen': '0.0.0.0',
                        'port': 6653,
                        'foreground': False,
                        'protocol_name': '',
                        'enable_entities_by_default': False,
                        'token_expiration_minutes': 180,
                        'thread_pool_max_workers': 256,
                        'debug': False}

        """
        defaults = {'pidfile': os.path.join(BASE_ENV,
                                            'var/run/kytos/kytosd.pid'),
                    'workdir': os.path.join(BASE_ENV, 'var/lib/kytos'),
                    'napps': os.path.join(BASE_ENV, 'var/lib/kytos/napps/'),
                    'napps_repositories': "['https://napps.kytos.io/repo/']",
                    'installed_napps': os.path.join(BASE_ENV,
                                                    'var/lib/kytos/napps/',
                                                    '.installed'),
                    'conf': os.path.join(BASE_ENV, 'etc/kytos/kytos.conf'),
                    'logging': os.path.join(BASE_ENV, 'etc/kytos/logging.ini'),
                    'logger_decorators':
                        ["kytos.core.logger_decorators.queue_decorator"],
                    'listen': '0.0.0.0',
                    'port': 6653,
                    'api_traceback_on_500': True,
                    'foreground': False,
                    'protocol_name': '',
                    'enable_entities_by_default': False,
                    'napps_pre_installed': [],
                    'authenticate_urls': [],
                    'vlan_pool': {},
                    'token_expiration_minutes': 180,
                    'thread_pool_max_workers': {},
                    'database': '',
                    'apm': '',
                    'connection_timeout': 130,
                    'debug': False}

        options, argv = self.conf_parser.parse_known_args()

        config = ConfigParser()

        config_file = options.conf or defaults.get('conf')

        if not os.path.exists(config_file):
            _render_config_templates(TEMPLATE_FILES, BASE_ENV,
                                     prefix=BASE_ENV,
                                     syslog_args=SYSLOG_ARGS)

        config.read(config_file)

        defaults.update(dict(config.items("daemon")))

        self.parser.set_defaults(**defaults)

        self.options['daemon'] = self._parse_options(argv)

    def _parse_options(self, argv):
        """Create a Namespace using the given argv.

        Args:
            argv(dict): Python Dict used to create the namespace.

        Returns:
            options(Namespace): Namespace with the args given

        """
        options, unknown = self.parser.parse_known_args(argv)
        if unknown:
            warnings.warn(f"Unknown arguments: {unknown}")
        options.napps_repositories = json.loads(options.napps_repositories)
        options.debug = options.debug in ['True', True]
        options.daemon = options.daemon in ['True', True]
        options.port = int(options.port)
        options.api_port = int(options.api_port)
        options.api_traceback_on_500 = options.api_traceback_on_500 in ['True',
                                                                        True]
        options.protocol_name = str(options.protocol_name)
        options.token_expiration_minutes = int(options.
                                               token_expiration_minutes)
        result = options.enable_entities_by_default in ['True', True]
        options.enable_entities_by_default = result

        def _parse_json(value):
            """Parse JSON lists and dicts from the config file."""
            if isinstance(value, str):
                return json.loads(value)
            return value

        options.logger_decorators = _parse_json(options.logger_decorators)
        options.napps_pre_installed = _parse_json(options.napps_pre_installed)
        options.vlan_pool = _parse_json(options.vlan_pool)
        options.authenticate_urls = _parse_json(options.authenticate_urls)
        thread_pool_max_workers = options.thread_pool_max_workers
        options.thread_pool_max_workers = _parse_json(thread_pool_max_workers)

        return options

    @staticmethod
    def options_exposed(daemon_options: dict) -> dict:
        """Options exposed on API."""
        options = dict(daemon_options)
        for key in ["jwt_secret"]:
            options.pop(key, None)
        return options


def _render_config_templates(templates,
                             destination=Path(__file__).parent,
                             **kwargs):
    """Create a config file based on a template file.

    If no destination is passed, the new conf file will be created on the
    directory of the template file.

    Args:
        template (string):    Path of the template file
        destination (string): Directory in which the config file will
                              be placed.
    """
    if str(kwargs['prefix']) != '/':
        kwargs['prefix'] = Path(str(kwargs['prefix']).rstrip('/'))
    kwargs['jwt_secret'] = uuid.uuid4().hex

    # Create the paths used by Kytos.
    directories = [os.path.join(BASE_ENV, ETC_KYTOS)]
    for directory in directories:
        os.makedirs(directory, exist_ok=True)

    tmpl_path = Path(os.path.abspath(os.path.dirname(__file__))).parent

    for tmpl in templates:
        path = os.path.join(tmpl_path, tmpl)
        with open(path, 'r', encoding='utf-8') as src_file:
            content = Template(src_file.read()).render(**kwargs)
            tmpl = tmpl.replace('templates', ETC_KYTOS) \
                       .replace('.template', '')
            dst_path = Path(destination) / tmpl
            with open(dst_path, 'w', encoding="utf8") as dst_file:
                dst_file.write(content)


1			"""Here you can control the config parameters to run Kytos controller.
2
3			Basically you can use a config file (-c option) and use arguments on command
4			line. If you specify a config file, then and option configured inside this file
5			will be overridden by the option on command line.
6			"""
7
8			import json
9			import os
10			import uuid
11			import warnings
12			from argparse import ArgumentParser, RawDescriptionHelpFormatter
13			from configparser import ConfigParser
14			from pathlib import Path
15
16			from jinja2 import Template
17
18			from kytos.core.metadata import __version__
19
20			BASE_ENV = os.environ.get('VIRTUAL_ENV', None) or '/'
21			ETC_KYTOS = 'etc/kytos'
22			TEMPLATE_FILES = ['templates/kytos.conf.template',
23			'templates/logging.ini.template']
24			SYSLOG_ARGS = ['/dev/log'] if Path('/dev/log').exists() else []
25
26
27			class KytosConfig():
28			"""Handle settings of Kytos."""
29
30			def __init__(self):
31			"""Parse the command line.
32
33			The contructor set defaults parameters that can be used by KytosConfig.
34			"""
35			self.options = {}
36			conf_parser = ArgumentParser(add_help=False)
37
38			conf_parser.add_argument("-c", "--conf",
39			help="Specify a config file",
40			metavar="FILE")
41
42			parser = ArgumentParser(prog='kytosd',
43			parents=[conf_parser],
44			formatter_class=RawDescriptionHelpFormatter,
45			description=__doc__)
46
47			parser.add_argument('-v', '--version',
48			action='version',
49			version=f"kytosd {__version__}")
50
51			parser.add_argument('-D', '--debug',
52			action='store_true',
53			help="Run in debug mode")
54
55			parser.add_argument('-f', '--foreground',
56			action='store_true',
57			help="Run in foreground (ctrl+c to stop)")
58
59			parser.add_argument('-l', '--listen',
60			action='store',
61			help="IP/Interface to be listened")
62
63			parser.add_argument('-n', '--napps',
64			action='store',
65			help="Specify the napps directory")
66
67			parser.add_argument('-P', '--port',
68			action='store',
69			help="Port to be listened")
70
71			parser.add_argument('-p', '--pidfile',
72			action='store',
73			help="Specify the PID file to save.")
74
75			parser.add_argument('-w', '--workdir',
76			action='store',
77			help="Specify the working directory")
78
79			parser.add_argument('-s', '--protocol_name',
80			action='store',
81			help="Specify the southbound protocol")
82
83			parser.add_argument('-E', '--enable_entities_by_default',
84			action='store_true',
85			help="Enable all new Entities by default.")
86
87			parser.add_argument('-C', '--create-superuser',
88			action='store_true',
89			help="Create a kytos superuser.")
90
91			parser.add_argument('-t', '--thread_pool_max_workers',
92			action='store',
93			help="Maximum number of threads in the pool.")
94
95			parser.add_argument('-d', '--database',
96			action='store',
97			help="Database backend.")
98
99			parser.add_argument('-a', '--apm',
100			action='store',
101			help="APM backend.")
102
103			self.conf_parser, self.parser = conf_parser, parser
104			self.parse_args()
105
106			def parse_args(self):
107			"""Get the command line options and update kytos settings.
108
109			When installed via pip, defaults values are:
110
111			.. code-block:: python
112
113			defaults = {'pidfile': '/var/run/kytos/kytosd.pid',
114			'workdir': '/var/lib/kytos',
115			'napps': '/var/lib/kytos/napps/',
116			'conf': '/etc/kytos/kytos.conf',
117			'logging': '/etc/kytos/logging.ini',
118			'listen': '0.0.0.0',
119			'port': 6653,
120			'foreground': False,
121			'protocol_name': '',
122			'enable_entities_by_default': False,
123			'token_expiration_minutes': 180,
124			'thread_pool_max_workers': 256,
125			'debug': False}
126
127			"""
128			defaults = {'pidfile': os.path.join(BASE_ENV,
129			'var/run/kytos/kytosd.pid'),
130			'workdir': os.path.join(BASE_ENV, 'var/lib/kytos'),
131			'napps': os.path.join(BASE_ENV, 'var/lib/kytos/napps/'),
132			'napps_repositories': "['https://napps.kytos.io/repo/']",
133			'installed_napps': os.path.join(BASE_ENV,
134			'var/lib/kytos/napps/',
135			'.installed'),
136			'conf': os.path.join(BASE_ENV, 'etc/kytos/kytos.conf'),
137			'logging': os.path.join(BASE_ENV, 'etc/kytos/logging.ini'),
138			'logger_decorators':
139			["kytos.core.logger_decorators.queue_decorator"],
140			'listen': '0.0.0.0',
141			'port': 6653,
142			'api_traceback_on_500': True,
143			'foreground': False,
144			'protocol_name': '',
145			'enable_entities_by_default': False,
146			'napps_pre_installed': [],
147			'authenticate_urls': [],
148			'vlan_pool': {},
149			'token_expiration_minutes': 180,
150			'thread_pool_max_workers': {},
151			'database': '',
152			'apm': '',
153			'connection_timeout': 130,
154			'debug': False}
155
156			options, argv = self.conf_parser.parse_known_args()
157
158			config = ConfigParser()
159
160			config_file = options.conf or defaults.get('conf')
161
162			if not os.path.exists(config_file):
163			_render_config_templates(TEMPLATE_FILES, BASE_ENV,
164			prefix=BASE_ENV,
165			syslog_args=SYSLOG_ARGS)
166
167			config.read(config_file)
168
169			defaults.update(dict(config.items("daemon")))
170
171			self.parser.set_defaults(**defaults)
172
173			self.options['daemon'] = self._parse_options(argv)
174
175			def _parse_options(self, argv):
176			"""Create a Namespace using the given argv.
177
178			Args:
179			argv(dict): Python Dict used to create the namespace.
180
181			Returns:
182			options(Namespace): Namespace with the args given
183
184			"""
185			options, unknown = self.parser.parse_known_args(argv)
186			if unknown:
187			warnings.warn(f"Unknown arguments: {unknown}")
188			options.napps_repositories = json.loads(options.napps_repositories)
189			options.debug = options.debug in ['True', True]
190			options.daemon = options.daemon in ['True', True]
191			options.port = int(options.port)
192			options.api_port = int(options.api_port)
193			options.api_traceback_on_500 = options.api_traceback_on_500 in ['True',
194			True]
195			options.protocol_name = str(options.protocol_name)
196			options.token_expiration_minutes = int(options.
197			token_expiration_minutes)
198			result = options.enable_entities_by_default in ['True', True]
199			options.enable_entities_by_default = result
200
201			def _parse_json(value):
202			"""Parse JSON lists and dicts from the config file."""
203			if isinstance(value, str):
204			return json.loads(value)
205			return value
206
207			options.logger_decorators = _parse_json(options.logger_decorators)
208			options.napps_pre_installed = _parse_json(options.napps_pre_installed)
209			options.vlan_pool = _parse_json(options.vlan_pool)
210			options.authenticate_urls = _parse_json(options.authenticate_urls)
211			thread_pool_max_workers = options.thread_pool_max_workers
212			options.thread_pool_max_workers = _parse_json(thread_pool_max_workers)
213
214			return options
215
216			@staticmethod
217			def options_exposed(daemon_options: dict) -> dict:
218			"""Options exposed on API."""
219			options = dict(daemon_options)
220			for key in ["jwt_secret"]:
221			options.pop(key, None)
222			return options
223
224
225			def _render_config_templates(templates,
226			destination=Path(__file__).parent,
227			**kwargs):
228			"""Create a config file based on a template file.
229
230			If no destination is passed, the new conf file will be created on the
231			directory of the template file.
232
233			Args:
234			template (string): Path of the template file
235			destination (string): Directory in which the config file will
236			be placed.
237			"""
238			if str(kwargs['prefix']) != '/':
239			kwargs['prefix'] = Path(str(kwargs['prefix']).rstrip('/'))
240			kwargs['jwt_secret'] = uuid.uuid4().hex
241
242			# Create the paths used by Kytos.
243			directories = [os.path.join(BASE_ENV, ETC_KYTOS)]
244			for directory in directories:
245			os.makedirs(directory, exist_ok=True)
246
247			tmpl_path = Path(os.path.abspath(os.path.dirname(__file__))).parent
248
249			for tmpl in templates:
250			path = os.path.join(tmpl_path, tmpl)
251			with open(path, 'r', encoding='utf-8') as src_file:
252			content = Template(src_file.read()).render(**kwargs)
253			tmpl = tmpl.replace('templates', ETC_KYTOS) \
254			.replace('.template', '')
255			dst_path = Path(destination) / tmpl
256			with open(dst_path, 'w', encoding="utf8") as dst_file:
257			dst_file.write(content)
258

kytos-ng / kytos

Push — master ( e44493...373da3 )

kytos.core.config.KytosConfig.options_exposed() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like