forked from spotify/luigi
-
Notifications
You must be signed in to change notification settings - Fork 0
/
configuration.py
96 lines (77 loc) · 3.25 KB
/
configuration.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
# -*- coding: utf-8 -*-
#
# Copyright 2012-2015 Spotify AB
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
#
import logging
import os
from ConfigParser import ConfigParser, NoOptionError, NoSectionError
class LuigiConfigParser(ConfigParser):
NO_DEFAULT = object()
_instance = None
_config_paths = ['/etc/luigi/client.cfg', 'client.cfg']
if 'LUIGI_CONFIG_PATH' in os.environ:
_config_paths.append(os.environ['LUIGI_CONFIG_PATH'])
@classmethod
def add_config_path(cls, path):
cls._config_paths.append(path)
cls.reload()
@classmethod
def instance(cls, *args, **kwargs):
""" Singleton getter """
if cls._instance is None:
cls._instance = cls(*args, **kwargs)
loaded = cls._instance.reload()
logging.getLogger('luigi-interface').info('Loaded %r', loaded)
return cls._instance
@classmethod
def reload(cls):
return cls.instance().read(cls._config_paths)
def _get_with_default(self, method, section, option, default, expected_type=None):
"""
Gets the value of the section/option using method.
Returns default if value is not found.
Raises an exception if the default value is not None and doesn't match the expected_type.
"""
try:
return method(self, section, option)
except (NoOptionError, NoSectionError):
if default is LuigiConfigParser.NO_DEFAULT:
raise
if expected_type is not None and default is not None and \
not isinstance(default, expected_type):
raise
return default
def get(self, section, option, default=NO_DEFAULT):
return self._get_with_default(ConfigParser.get, section, option, default)
def getboolean(self, section, option, default=NO_DEFAULT):
return self._get_with_default(ConfigParser.getboolean, section, option, default, bool)
def getint(self, section, option, default=NO_DEFAULT):
return self._get_with_default(ConfigParser.getint, section, option, default, int)
def getfloat(self, section, option, default=NO_DEFAULT):
return self._get_with_default(ConfigParser.getfloat, section, option, default, float)
def getintdict(self, section):
try:
return dict((key, int(value)) for key, value in self.items(section))
except NoSectionError:
return {}
def set(self, section, option, value=None):
if not ConfigParser.has_section(self, section):
ConfigParser.add_section(self, section)
return ConfigParser.set(self, section, option, value)
def get_config():
"""
Convenience method (for backwards compatibility) for accessing config singleton.
"""
return LuigiConfigParser.instance()