__Settings.load_config_file() - Code Metrics - buildtimetrend/python-lib - Measure and Improve Code Quality continuously with Scrutinizer

__Settings.load_config_file() F
last analyzed 2016-05-18 18:14 UTC

↳ Parent: __Settings

Complexity

Conditions

Size

Total Lines

Duplication

Lines	0
Ratio	0 %

Metric	Value
cc	10
dl	0
loc	29
rs	3.1304

How to fix Complexity

# vim: set expandtab sw=4 ts=4:
# pylint: disable=invalid-name,too-few-public-methods
"""
Manage Buildtime Trend settings.

Copyright (C) 2014-2016 Dieter Adriaenssens <[email protected]>

This file is part of buildtimetrend/python-lib
<https://github.com/buildtimetrend/python-lib/>

This program is free software: you can redistribute it and/or modify
it under the terms of the GNU Affero General Public License as published by
the Free Software Foundation, either version 3 of the License, or
any later version.

This program is distributed in the hope that it will be useful,
but WITHOUT ANY WARRANTY; without even the implied warranty of
MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
GNU Affero General Public License for more details.

You should have received a copy of the GNU Affero General Public License
along with this program. If not, see <http://www.gnu.org/licenses/>.
"""

from __future__ import print_function
import os
import getopt
from collections import OrderedDict
import yaml
import keen
from configobj import ConfigObj
from validate import Validator
import buildtimetrend
from buildtimetrend.collection import Collection
from buildtimetrend.tools import check_file
from buildtimetrend.tools import is_dict
from buildtimetrend import set_loglevel
from buildtimetrend import logger


class Settings(object):

    """
    Settings class is a singleton.

    Inspired by
    http://python-3-patterns-idioms-test.readthedocs.org/en/latest/Singleton.html
    """

    class __Settings(object):

        """Settings class contains settings and config options."""

        def __init__(self):
            """Initialise class."""
            self.settings = Collection()
            self.set_defaults()

        def set_defaults(self):
            """Set default values."""
            # set loglevel
            self.add_setting("loglevel", "WARNING")

            # set default project name
            self.set_project_name(buildtimetrend.NAME)

            # set modes
            self.set_mode("native", False)
            self.set_mode("keen", True)

            # set default paths
            self.add_setting('dashboard_configfile',
                             'dashboard/config.js')

            # set multi build import settings
            self.add_setting(
                'multi_import',
                {
                    'max_builds': 100,
                    'delay': 3
                }
            )

            # set level detail of build job data storage
            self.add_setting("data_detail", "full")
            self.add_setting("repo_data_detail", {})
            self.add_setting(
                "task_queue",
                {
                    "backend": "",
                    "broker_url": ""
                },
            )

        def set_project_name(self, name):
            """
            Set project name.

            Parameters :
            - name : project name
            """
            self.add_setting("project_name", name)

        def get_project_name(self):
            """Get project name."""
            return self.get_setting("project_name")

        def set_client(self, name, version):
            """
            Set client name and version.

            Parameters :
            - name : client name (fe. service, python-client)
            - version : client version
            """
            self.add_setting("client", name)
            self.add_setting("client_version", version)

        def add_setting(self, name, value):
            """
            Add a setting.

            Parameters :
            - name : Setting name
            - value : Setting value
            """
            self.settings.add_item(name, value)

        def get_setting(self, name):
            """
            Get a setting value.

            Parameters :
            - name : Setting name
            """
            return self.settings.get_item(name)

        def get_value_or_setting(self, name, value=None):
            """
            Get a setting if value is not defined.

            Parameters :
            - name : Setting name
            - value : value
            """
            if value is None:
                return self.get_setting(name)
            else:
                return value

        def load_settings(self, argv=None, config_file="config.yml"):
            """
            Load config settings.

            Settings are retrieved from :
            - configfile
            - environment variables
            - command line arguments
            """
            self.load_config_file(config_file)
            self.load_env_vars()
            return self.process_argv(argv)

        def load_config_file(self, config_file):
            """
            Load settings from a config file.

            Parameters :
            - config_file : name of the config file
            """
            if not check_file(config_file):
                return False

            with open(config_file, 'r') as file_stream:
                config = yaml.load(file_stream)
                if "buildtimetrend" in config and \
                        is_dict(config["buildtimetrend"]):
                    self.settings.add_items(config["buildtimetrend"])

                set_loglevel(self.get_setting("loglevel"))

                # set Keen.io settings
                if "keen" in config:
                    if "project_id" in config["keen"]:
                        keen.project_id = config["keen"]["project_id"]
                    if "write_key" in config["keen"]:
                        keen.write_key = config["keen"]["write_key"]
                    if "read_key" in config["keen"]:
                        keen.read_key = config["keen"]["read_key"]
                    if "master_key" in config["keen"]:
                        keen.master_key = config["keen"]["master_key"]
                return True

        def load_config_ini_file(self, config_file="config.ini"):
            """
            Load settings from a config file, using objconfig.

            Parameters :
            - config_file : name of the config file
            """
            spec_filename = "buildtimetrend/config_spec.ini"

            configspec = ConfigObj(
                spec_filename,
                interpolation=False,
                list_values=False,
                _inspec=True
            )

            config = ConfigObj(config_file, configspec=configspec)

            if not config.validate(Validator()):
                logger.warning(
                    "Error validating config file %s",
                    config_file
                )

            return config

        def get_project_info(self):
            """Get project info as a dictonary."""
            return {
                "lib_version": buildtimetrend.VERSION,
                "schema_version": buildtimetrend.SCHEMA_VERSION,
                "client": str(self.get_setting("client")),
                "client_version": str(self.get_setting("client_version")),
                "project_name": str(self.get_project_name())
            }

        def process_argv(self, argv):
            """
            Process command line arguments.

            Returns a list with arguments (non-options) or
            None if options are invalid
            """
            if argv is None:
                return None

            usage_string = '{0!s} -h --log=<log_level> --build=<buildID>' \
                ' --job=<jobID> --branch=<branchname> --repo=<repo_slug>' \
                ' --ci=<ci_platform> --result=<build_result>' \
                ' --mode=<storage_mode>'
            usage_string = usage_string.format(argv[0])

            options = {
                "--build": "build",
                "--job": "job",
                "--branch": "branch",
                "--ci": "ci_platform",
                "--result": "result"
            }

            try:
                opts, args = getopt.getopt(
                    argv[1:], "h", [
                        "log=",
                        "build=", "job=", "branch=", "repo=",
                        "ci=", "result=", "mode=", "help"]
                )
            except getopt.GetoptError:
                print(usage_string)
                return None

            # check options
            for opt, arg in opts:
                if opt in ('-h', "--help"):
                    print(usage_string)
                    return None
                elif opt == "--log":
                    set_loglevel(arg)
                elif opt in options:
                    self.add_setting(options[opt], arg)
                elif opt == "--repo":
                    self.set_project_name(arg)
                elif opt == "--mode":
                    self.set_mode(arg)

            return args

        def set_mode(self, mode, value=True):
            """
            Set operating mode.

            Parameters:
            - mode : keen, native
            - value : enable (=True, default) or disable (=False) mode
            """
            if mode == "native":
                self.add_setting("mode_native", bool(value))
            elif mode == "keen":
                self.add_setting("mode_keen", bool(value))

        def load_env_vars(self):
            """
            Load environment variables.

            Assign the environment variable values to
            the corresponding setting.
            """
            # assign environment variable values to setting value
            if self.env_var_to_settings("BTT_LOGLEVEL", "loglevel"):
                set_loglevel(self.get_setting("loglevel"))

            self.env_var_to_settings("TRAVIS_ACCOUNT_TOKEN",
                                     "travis_account_token")
            self.env_var_to_settings("BUILD_TREND_CONFIGFILE",
                                     "dashboard_configfile")
            self.env_var_to_settings("BTT_DATA_DETAIL", "data_detail")

            # load task queue environment variables
            self.load_env_vars_task_queue()
            # load multi build import environment variables
            self.load_env_vars_multi_import()

        def load_env_vars_task_queue(self):
            """
            Load task queue environment variables.

            The environment variable that matches first is loaded,
            other variables are ignored.
            """
            # prepare list of env vars, in order of priority
            queue_env_vars = OrderedDict()
            queue_env_vars["BTT_AMQP_URL"] = "amqp"
            queue_env_vars["BTT_REDIS_URL"] = "redis"
            queue_env_vars["RABBITMQ_BIGWIG_URL"] = "amqp"
            queue_env_vars["CLOUDAMQP_URL"] = "amqp"
            queue_env_vars["REDISGREEN_URL"] = "redis"

            # loop over list of env vars, loading first match
            for env_var in queue_env_vars.keys():
                if env_var in os.environ:
                    self.add_setting(
                        "task_queue",
                        {
                            "backend": queue_env_vars[env_var],
                            "broker_url": os.environ[env_var]
                        }
                    )
                    # exit loop on first match
                    break

        def load_env_vars_multi_import(self):
            """Load multi build import environment variables."""
            multi_import = {}

            if "BTT_MULTI_MAX_BUILDS" in os.environ:
                multi_import["max_builds"] = \
                    int(os.environ["BTT_MULTI_MAX_BUILDS"])
            if "BTT_MULTI_DELAY" in os.environ:
                multi_import["delay"] = int(os.environ["BTT_MULTI_DELAY"])

            # check if collection is empty
            if multi_import:
                self.add_setting("multi_import", multi_import)

        def env_var_to_settings(self, env_var_name, settings_name):
            """
            Store environment variable value as a setting.

            Parameters:
            - env_var_name : Name of the environment variable
            - settings_name : Name of the corresponding settings value
            """
            if env_var_name in os.environ:
                self.add_setting(settings_name, os.environ[env_var_name])
                logger.debug(
                    "Setting %s was set to %s",
                    settings_name, os.environ[env_var_name])
                return True
            else:
                logger.debug(
                    "Setting %s was not set,"
                    " environment variable %s doesn't exist",
                    settings_name, env_var_name)
                return False

    instance = None

    def __new__(cls):  # __new__ always a classmethod
        """Create a singleton."""
        if not Settings.instance:
            Settings.instance = Settings.__Settings()
        return Settings.instance

    def __getattr__(self, name):
        """Redirect access to get singleton properties."""
        return getattr(self.instance, name)

    def __setattr__(self, name, value):
        """Redirect access to set singleton properties."""
        return setattr(self.instance, name, value)


1			# vim: set expandtab sw=4 ts=4:
2			# pylint: disable=invalid-name,too-few-public-methods
3			"""
4			Manage Buildtime Trend settings.
5
6			Copyright (C) 2014-2016 Dieter Adriaenssens <[email protected]>
7
8			This file is part of buildtimetrend/python-lib
9			<https://github.com/buildtimetrend/python-lib/>
10
11			This program is free software: you can redistribute it and/or modify
12			it under the terms of the GNU Affero General Public License as published by
13			the Free Software Foundation, either version 3 of the License, or
14			any later version.
15
16			This program is distributed in the hope that it will be useful,
17			but WITHOUT ANY WARRANTY; without even the implied warranty of
18			MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
19			GNU Affero General Public License for more details.
20
21			You should have received a copy of the GNU Affero General Public License
22			along with this program. If not, see <http://www.gnu.org/licenses/>.
23			"""
24
25			from __future__ import print_function
26			import os
27			import getopt
28			from collections import OrderedDict
29			import yaml
30			import keen
31			from configobj import ConfigObj
32			from validate import Validator
33			import buildtimetrend
34			from buildtimetrend.collection import Collection
35			from buildtimetrend.tools import check_file
36			from buildtimetrend.tools import is_dict
37			from buildtimetrend import set_loglevel
38			from buildtimetrend import logger
39
40
41			class Settings(object):
42
43			"""
44			Settings class is a singleton.
45
46			Inspired by
47			http://python-3-patterns-idioms-test.readthedocs.org/en/latest/Singleton.html
48			"""
49
50			class __Settings(object):
51
52			"""Settings class contains settings and config options."""
53
54			def __init__(self):
55			"""Initialise class."""
56			self.settings = Collection()
57			self.set_defaults()
58
59			def set_defaults(self):
60			"""Set default values."""
61			# set loglevel
62			self.add_setting("loglevel", "WARNING")
63
64			# set default project name
65			self.set_project_name(buildtimetrend.NAME)
66
67			# set modes
68			self.set_mode("native", False)
69			self.set_mode("keen", True)
70
71			# set default paths
72			self.add_setting('dashboard_configfile',
73			'dashboard/config.js')
74
75			# set multi build import settings
76			self.add_setting(
77			'multi_import',
78			{
79			'max_builds': 100,
80			'delay': 3
81			}
82			)
83
84			# set level detail of build job data storage
85			self.add_setting("data_detail", "full")
86			self.add_setting("repo_data_detail", {})
87			self.add_setting(
88			"task_queue",
89			{
90			"backend": "",
91			"broker_url": ""
92			},
93			)
94
95			def set_project_name(self, name):
96			"""
97			Set project name.
98
99			Parameters :
100			- name : project name
101			"""
102			self.add_setting("project_name", name)
103
104			def get_project_name(self):
105			"""Get project name."""
106			return self.get_setting("project_name")
107
108			def set_client(self, name, version):
109			"""
110			Set client name and version.
111
112			Parameters :
113			- name : client name (fe. service, python-client)
114			- version : client version
115			"""
116			self.add_setting("client", name)
117			self.add_setting("client_version", version)
118
119			def add_setting(self, name, value):
120			"""
121			Add a setting.
122
123			Parameters :
124			- name : Setting name
125			- value : Setting value
126			"""
127			self.settings.add_item(name, value)
128
129			def get_setting(self, name):
130			"""
131			Get a setting value.
132
133			Parameters :
134			- name : Setting name
135			"""
136			return self.settings.get_item(name)
137
138			def get_value_or_setting(self, name, value=None):
139			"""
140			Get a setting if value is not defined.
141
142			Parameters :
143			- name : Setting name
144			- value : value
145			"""
146			if value is None:
147			return self.get_setting(name)
148			else:
149			return value
150
151			def load_settings(self, argv=None, config_file="config.yml"):
152			"""
153			Load config settings.
154
155			Settings are retrieved from :
156			- configfile
157			- environment variables
158			- command line arguments
159			"""
160			self.load_config_file(config_file)
161			self.load_env_vars()
162			return self.process_argv(argv)
163
164			def load_config_file(self, config_file):
165			"""
166			Load settings from a config file.
167
168			Parameters :
169			- config_file : name of the config file
170			"""
171			if not check_file(config_file):
172			return False
173
174			with open(config_file, 'r') as file_stream:
175			config = yaml.load(file_stream)
176			if "buildtimetrend" in config and \
177			is_dict(config["buildtimetrend"]):
178			self.settings.add_items(config["buildtimetrend"])
179
180			set_loglevel(self.get_setting("loglevel"))
181
182			# set Keen.io settings
183			if "keen" in config:
184			if "project_id" in config["keen"]:
185			keen.project_id = config["keen"]["project_id"]
186			if "write_key" in config["keen"]:
187			keen.write_key = config["keen"]["write_key"]
188			if "read_key" in config["keen"]:
189			keen.read_key = config["keen"]["read_key"]
190			if "master_key" in config["keen"]:
191			keen.master_key = config["keen"]["master_key"]
192			return True
193
194			def load_config_ini_file(self, config_file="config.ini"):
195			"""
196			Load settings from a config file, using objconfig.
197
198			Parameters :
199			- config_file : name of the config file
200			"""
201			spec_filename = "buildtimetrend/config_spec.ini"
202
203			configspec = ConfigObj(
204			spec_filename,
205			interpolation=False,
206			list_values=False,
207			_inspec=True
208			)
209
210			config = ConfigObj(config_file, configspec=configspec)
211
212			if not config.validate(Validator()):
213			logger.warning(
214			"Error validating config file %s",
215			config_file
216			)
217
218			return config
219
220			def get_project_info(self):
221			"""Get project info as a dictonary."""
222			return {
223			"lib_version": buildtimetrend.VERSION,
224			"schema_version": buildtimetrend.SCHEMA_VERSION,
225			"client": str(self.get_setting("client")),
226			"client_version": str(self.get_setting("client_version")),
227			"project_name": str(self.get_project_name())
228			}
229
230			def process_argv(self, argv):
231			"""
232			Process command line arguments.
233
234			Returns a list with arguments (non-options) or
235			None if options are invalid
236			"""
237			if argv is None:
238			return None
239
240			usage_string = '{0!s} -h --log=<log_level> --build=<buildID>' \
241			' --job=<jobID> --branch=<branchname> --repo=<repo_slug>' \
242			' --ci=<ci_platform> --result=<build_result>' \
243			' --mode=<storage_mode>'
244			usage_string = usage_string.format(argv[0])
245
246			options = {
247			"--build": "build",
248			"--job": "job",
249			"--branch": "branch",
250			"--ci": "ci_platform",
251			"--result": "result"
252			}
253
254			try:
255			opts, args = getopt.getopt(
256			argv[1:], "h", [
257			"log=",
258			"build=", "job=", "branch=", "repo=",
259			"ci=", "result=", "mode=", "help"]
260			)
261			except getopt.GetoptError:
262			print(usage_string)
263			return None
264
265			# check options
266			for opt, arg in opts:
267			if opt in ('-h', "--help"):
268			print(usage_string)
269			return None
270			elif opt == "--log":
271			set_loglevel(arg)
272			elif opt in options:
273			self.add_setting(options[opt], arg)
274			elif opt == "--repo":
275			self.set_project_name(arg)
276			elif opt == "--mode":
277			self.set_mode(arg)
278
279			return args
280
281			def set_mode(self, mode, value=True):
282			"""
283			Set operating mode.
284
285			Parameters:
286			- mode : keen, native
287			- value : enable (=True, default) or disable (=False) mode
288			"""
289			if mode == "native":
290			self.add_setting("mode_native", bool(value))
291			elif mode == "keen":
292			self.add_setting("mode_keen", bool(value))
293
294			def load_env_vars(self):
295			"""
296			Load environment variables.
297
298			Assign the environment variable values to
299			the corresponding setting.
300			"""
301			# assign environment variable values to setting value
302			if self.env_var_to_settings("BTT_LOGLEVEL", "loglevel"):
303			set_loglevel(self.get_setting("loglevel"))
304
305			self.env_var_to_settings("TRAVIS_ACCOUNT_TOKEN",
306			"travis_account_token")
307			self.env_var_to_settings("BUILD_TREND_CONFIGFILE",
308			"dashboard_configfile")
309			self.env_var_to_settings("BTT_DATA_DETAIL", "data_detail")
310
311			# load task queue environment variables
312			self.load_env_vars_task_queue()
313			# load multi build import environment variables
314			self.load_env_vars_multi_import()
315
316			def load_env_vars_task_queue(self):
317			"""
318			Load task queue environment variables.
319
320			The environment variable that matches first is loaded,
321			other variables are ignored.
322			"""
323			# prepare list of env vars, in order of priority
324			queue_env_vars = OrderedDict()
325			queue_env_vars["BTT_AMQP_URL"] = "amqp"
326			queue_env_vars["BTT_REDIS_URL"] = "redis"
327			queue_env_vars["RABBITMQ_BIGWIG_URL"] = "amqp"
328			queue_env_vars["CLOUDAMQP_URL"] = "amqp"
329			queue_env_vars["REDISGREEN_URL"] = "redis"
330
331			# loop over list of env vars, loading first match
332			for env_var in queue_env_vars.keys():
333			if env_var in os.environ:
334			self.add_setting(
335			"task_queue",
336			{
337			"backend": queue_env_vars[env_var],
338			"broker_url": os.environ[env_var]
339			}
340			)
341			# exit loop on first match
342			break
343
344			def load_env_vars_multi_import(self):
345			"""Load multi build import environment variables."""
346			multi_import = {}
347
348			if "BTT_MULTI_MAX_BUILDS" in os.environ:
349			multi_import["max_builds"] = \
350			int(os.environ["BTT_MULTI_MAX_BUILDS"])
351			if "BTT_MULTI_DELAY" in os.environ:
352			multi_import["delay"] = int(os.environ["BTT_MULTI_DELAY"])
353
354			# check if collection is empty
355			if multi_import:
356			self.add_setting("multi_import", multi_import)
357
358			def env_var_to_settings(self, env_var_name, settings_name):
359			"""
360			Store environment variable value as a setting.
361
362			Parameters:
363			- env_var_name : Name of the environment variable
364			- settings_name : Name of the corresponding settings value
365			"""
366			if env_var_name in os.environ:
367			self.add_setting(settings_name, os.environ[env_var_name])
368			logger.debug(
369			"Setting %s was set to %s",
370			settings_name, os.environ[env_var_name])
371			return True
372			else:
373			logger.debug(
374			"Setting %s was not set,"
375			" environment variable %s doesn't exist",
376			settings_name, env_var_name)
377			return False
378
379			instance = None
380
381			def __new__(cls): # __new__ always a classmethod
382			"""Create a singleton."""
383			if not Settings.instance:
384			Settings.instance = Settings.__Settings()
385			return Settings.instance
386
387			def __getattr__(self, name):
388			"""Redirect access to get singleton properties."""
389			return getattr(self.instance, name)
390
391			def __setattr__(self, name, value):
392			"""Redirect access to set singleton properties."""
393			return setattr(self.instance, name, value)
394

buildtimetrend / python-lib

__Settings.load_config_file() F last analyzed 2016-05-18 18:14 UTC

Complexity

Size

Duplication

How to fix Complexity

Complexity

Duplication Side-by-Side

Filter issues like

__Settings.load_config_file() F
last analyzed 2016-05-18 18:14 UTC