__Settings.load_config_file() - Code Metrics - Inspection of "add more config specs" - buildtimetrend/python-lib - Measure and Improve Code Quality continuously with Scrutinizer

Completed

Push — master ( 889440...e965d2 )

by Dieter

created 2016-05-15 19:56 UTC

__Settings.load_config_file() F

↳ 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
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", {})

        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 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			import buildtimetrend
32			from buildtimetrend.collection import Collection
33			from buildtimetrend.tools import check_file
34			from buildtimetrend.tools import is_dict
35			from buildtimetrend import set_loglevel
36			from buildtimetrend import logger
37
38
39			class Settings(object):
40
41			"""
42			Settings class is a singleton.
43
44			Inspired by
45			http://python-3-patterns-idioms-test.readthedocs.org/en/latest/Singleton.html
46			"""
47
48			class __Settings(object):
49
50			"""Settings class contains settings and config options."""
51
52			def __init__(self):
53			"""Initialise class."""
54			self.settings = Collection()
55			self.set_defaults()
56
57			def set_defaults(self):
58			"""Set default values."""
59			# set loglevel
60			self.add_setting("loglevel", "WARNING")
61
62			# set default project name
63			self.set_project_name(buildtimetrend.NAME)
64
65			# set modes
66			self.set_mode("native", False)
67			self.set_mode("keen", True)
68
69			# set default paths
70			self.add_setting('dashboard_configfile',
71			'dashboard/config.js')
72
73			# set multi build import settings
74			self.add_setting(
75			'multi_import',
76			{
77			'max_builds': 100,
78			'delay': 3
79			}
80			)
81
82			# set level detail of build job data storage
83			self.add_setting("data_detail", "full")
84			self.add_setting("repo_data_detail", {})
85
86			def set_project_name(self, name):
87			"""
88			Set project name.
89
90			Parameters :
91			- name : project name
92			"""
93			self.add_setting("project_name", name)
94
95			def get_project_name(self):
96			"""Get project name."""
97			return self.get_setting("project_name")
98
99			def set_client(self, name, version):
100			"""
101			Set client name and version.
102
103			Parameters :
104			- name : client name (fe. service, python-client)
105			- version : client version
106			"""
107			self.add_setting("client", name)
108			self.add_setting("client_version", version)
109
110			def add_setting(self, name, value):
111			"""
112			Add a setting.
113
114			Parameters :
115			- name : Setting name
116			- value : Setting value
117			"""
118			self.settings.add_item(name, value)
119
120			def get_setting(self, name):
121			"""
122			Get a setting value.
123
124			Parameters :
125			- name : Setting name
126			"""
127			return self.settings.get_item(name)
128
129			def get_value_or_setting(self, name, value=None):
130			"""
131			Get a setting if value is not defined.
132
133			Parameters :
134			- name : Setting name
135			- value : value
136			"""
137			if value is None:
138			return self.get_setting(name)
139			else:
140			return value
141
142			def load_settings(self, argv=None, config_file="config.yml"):
143			"""
144			Load config settings.
145
146			Settings are retrieved from :
147			- configfile
148			- environment variables
149			- command line arguments
150			"""
151			self.load_config_file(config_file)
152			self.load_env_vars()
153			return self.process_argv(argv)
154
155			def load_config_file(self, config_file):
156			"""
157			Load settings from a config file.
158
159			Parameters :
160			- config_file : name of the config file
161			"""
162			if not check_file(config_file):
163			return False
164
165			with open(config_file, 'r') as file_stream:
166			config = yaml.load(file_stream)
167			if "buildtimetrend" in config and \
168			is_dict(config["buildtimetrend"]):
169			self.settings.add_items(config["buildtimetrend"])
170
171			set_loglevel(self.get_setting("loglevel"))
172
173			# set Keen.io settings
174			if "keen" in config:
175			if "project_id" in config["keen"]:
176			keen.project_id = config["keen"]["project_id"]
177			if "write_key" in config["keen"]:
178			keen.write_key = config["keen"]["write_key"]
179			if "read_key" in config["keen"]:
180			keen.read_key = config["keen"]["read_key"]
181			if "master_key" in config["keen"]:
182			keen.master_key = config["keen"]["master_key"]
183			return True
184
185			def get_project_info(self):
186			"""Get project info as a dictonary."""
187			return {
188			"lib_version": buildtimetrend.VERSION,
189			"schema_version": buildtimetrend.SCHEMA_VERSION,
190			"client": str(self.get_setting("client")),
191			"client_version": str(self.get_setting("client_version")),
192			"project_name": str(self.get_project_name())
193			}
194
195			def process_argv(self, argv):
196			"""
197			Process command line arguments.
198
199			Returns a list with arguments (non-options) or
200			None if options are invalid
201			"""
202			if argv is None:
203			return None
204
205			usage_string = '{0!s} -h --log=<log_level> --build=<buildID>' \
206			' --job=<jobID> --branch=<branchname> --repo=<repo_slug>' \
207			' --ci=<ci_platform> --result=<build_result>' \
208			' --mode=<storage_mode>'
209			usage_string = usage_string.format(argv[0])
210
211			options = {
212			"--build": "build",
213			"--job": "job",
214			"--branch": "branch",
215			"--ci": "ci_platform",
216			"--result": "result"
217			}
218
219			try:
220			opts, args = getopt.getopt(
221			argv[1:], "h", [
222			"log=",
223			"build=", "job=", "branch=", "repo=",
224			"ci=", "result=", "mode=", "help"]
225			)
226			except getopt.GetoptError:
227			print(usage_string)
228			return None
229
230			# check options
231			for opt, arg in opts:
232			if opt in ('-h', "--help"):
233			print(usage_string)
234			return None
235			elif opt == "--log":
236			set_loglevel(arg)
237			elif opt in options:
238			self.add_setting(options[opt], arg)
239			elif opt == "--repo":
240			self.set_project_name(arg)
241			elif opt == "--mode":
242			self.set_mode(arg)
243
244			return args
245
246			def set_mode(self, mode, value=True):
247			"""
248			Set operating mode.
249
250			Parameters:
251			- mode : keen, native
252			- value : enable (=True, default) or disable (=False) mode
253			"""
254			if mode == "native":
255			self.add_setting("mode_native", bool(value))
256			elif mode == "keen":
257			self.add_setting("mode_keen", bool(value))
258
259			def load_env_vars(self):
260			"""
261			Load environment variables.
262
263			Assign the environment variable values to
264			the corresponding setting.
265			"""
266			# assign environment variable values to setting value
267			if self.env_var_to_settings("BTT_LOGLEVEL", "loglevel"):
268			set_loglevel(self.get_setting("loglevel"))
269
270			self.env_var_to_settings("TRAVIS_ACCOUNT_TOKEN",
271			"travis_account_token")
272			self.env_var_to_settings("BUILD_TREND_CONFIGFILE",
273			"dashboard_configfile")
274			self.env_var_to_settings("BTT_DATA_DETAIL", "data_detail")
275
276			# load task queue environment variables
277			self.load_env_vars_task_queue()
278			# load multi build import environment variables
279			self.load_env_vars_multi_import()
280
281			def load_env_vars_task_queue(self):
282			"""
283			Load task queue environment variables.
284
285			The environment variable that matches first is loaded,
286			other variables are ignored.
287			"""
288			# prepare list of env vars, in order of priority
289			queue_env_vars = OrderedDict()
290			queue_env_vars["BTT_AMQP_URL"] = "amqp"
291			queue_env_vars["BTT_REDIS_URL"] = "redis"
292			queue_env_vars["RABBITMQ_BIGWIG_URL"] = "amqp"
293			queue_env_vars["CLOUDAMQP_URL"] = "amqp"
294			queue_env_vars["REDISGREEN_URL"] = "redis"
295
296			# loop over list of env vars, loading first match
297			for env_var in queue_env_vars.keys():
298			if env_var in os.environ:
299			self.add_setting(
300			"task_queue",
301			{
302			"backend": queue_env_vars[env_var],
303			"broker_url": os.environ[env_var]
304			}
305			)
306			# exit loop on first match
307			break
308
309			def load_env_vars_multi_import(self):
310			"""Load multi build import environment variables."""
311			multi_import = {}
312
313			if "BTT_MULTI_MAX_BUILDS" in os.environ:
314			multi_import["max_builds"] = \
315			int(os.environ["BTT_MULTI_MAX_BUILDS"])
316			if "BTT_MULTI_DELAY" in os.environ:
317			multi_import["delay"] = int(os.environ["BTT_MULTI_DELAY"])
318
319			# check if collection is empty
320			if multi_import:
321			self.add_setting("multi_import", multi_import)
322
323			def env_var_to_settings(self, env_var_name, settings_name):
324			"""
325			Store environment variable value as a setting.
326
327			Parameters:
328			- env_var_name : Name of the environment variable
329			- settings_name : Name of the corresponding settings value
330			"""
331			if env_var_name in os.environ:
332			self.add_setting(settings_name, os.environ[env_var_name])
333			logger.debug(
334			"Setting %s was set to %s",
335			settings_name, os.environ[env_var_name])
336			return True
337			else:
338			logger.debug(
339			"Setting %s was not set,"
340			" environment variable %s doesn't exist",
341			settings_name, env_var_name)
342			return False
343
344			instance = None
345
346			def __new__(cls): # __new__ always a classmethod
347			"""Create a singleton."""
348			if not Settings.instance:
349			Settings.instance = Settings.__Settings()
350			return Settings.instance
351
352			def __getattr__(self, name):
353			"""Redirect access to get singleton properties."""
354			return getattr(self.instance, name)
355
356			def __setattr__(self, name, value):
357			"""Redirect access to set singleton properties."""
358			return setattr(self.instance, name, value)
359

buildtimetrend / python-lib

Push — master ( 889440...e965d2 )

__Settings.load_config_file() F

Complexity

Size

Duplication

How to fix Complexity

Complexity

Duplication Side-by-Side

Filter issues like