Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def initParser():
global parser
try:
parser = configparser.ConfigParser()
parser.read(constants.CONFIG_FILE)
if not(len(parser) > 1):
raise Exception()
logger.info(__name__, "Configparser initialized")
except:
logger.error(__name__, "Failed to load config.ini: Please ensure that a valid config file is located at {}".format(constants.CONFIG_FILE))
exit()
formatter = logging.Formatter('%(asctime)s - %(name)s - %(levelname)s - %(message)s')
# uncomment the next line to add filename and linenumber to logging output
#formatter = logging.Formatter('%(asctime)s - %(name)s - %(levelname)s {%(filename)s:%(lineno)d} %(message)s')
ch.setFormatter(formatter)
logger.addHandler(ch)
logging.info("C3TT publishing")
logging.debug("reading config")
### handle config
#make sure we have a config file
if not os.path.exists('client.conf'):
logging.error("Error: config file not found")
sys.exit(1)
config = configparser.ConfigParser()
config.read('client.conf')
source = config['general']['source']
dest = config['general']['dest']
source = "c3tt" #TODO quickfix for strange parser behavior
if source == "c3tt":
################### C3 Tracker ###################
#project = "projectslug"
group = config['C3Tracker']['group']
secret = config['C3Tracker']['secret']
if config['C3Tracker']['host'] == "None":
host = socket.getfqdn()
else:
host = config['C3Tracker']['host']
def parse_config(config_file=None):
if config_file is None:
config_file = constants.DEFAULT_CONFIG_FILE
config = ConfigParser()
try:
config.read(config_file)
except IOError:
logging.warning('Could not find configuration file %s' % config_file)
except ParsingError:
logging.error('There was an error parsing %s' % config_file)
sys.exit(1)
except Exception as e:
logging.error(
'There was an unknown error parsing %s: %s' %
config_file, e)
sys.exit(1)
if config.has_section('admin'):
config = config_to_dict(config)
section = config['admin']
def get_config():
config = configparser.ConfigParser()
config.read('./conf/diagnose.cnf', encoding='utf-8')
section_has = config.has_section('default')
if not section_has:
sys.exit("Error: The '[default]' not find")
processing = config.get("default", "processing")
host = config.get("default", "host")
user = config.get("default", "user")
password = config.get("default", "password")
port = config.get("default", "port")
database = config.get("default", "database")
log_level = config.get("default", "log_level")
type = config.get("default", "type")
conf_dict = {
'processing': processing,
'user': user,
def add_app_specific_ports(dep):
"""Add information of specific ports for the application
Args:
dep (str): deployment service description
Returns:
str: deployment service description with added specific port information for the application
"""
jupiter_config.set_globals()
INI_PATH = jupiter_config.APP_PATH + 'app_config.ini'
config = configparser.ConfigParser()
config.read(INI_PATH)
a = dep['spec']['ports']
for i in config['DOCKER_PORT']:
dic = {}
dic['name'] = i
dic['port'] = int(config['SVC_PORT'][i])
dic['targetPort'] = int(config['DOCKER_PORT'][i])
a.append(dic)
return dep
:param IN_filename: parameter file full path
:type IN_filename: string
:return: dictionary containing parameters
:rtype: dict
"""
print("[lakeSPProcessing] == readParamFile = %s ==" % IN_filename)
# 0 - Init output dictionary
OUT_params = {}
# Default values
OUT_params["flag_prod_shp"] = False
# 1 - Read parameter file
config = cfg.ConfigParser()
config.read(IN_filename)
# 2 - Retrieve PATHS
OUT_params["laketile_dir"] = config.get("PATHS", "LakeTile directory")
OUT_params["output_dir"] = config.get("PATHS", "Output directory")
#• 3 - Retrieve TILES_INFOS
OUT_params["cycle_num"] = config.getint("TILES_INFOS", "Cycle number")
OUT_params["pass_num"] = config.getint("TILES_INFOS", "Pass number")
# 4 - Retrieve OPTIONS
if "OPTIONS" in config.sections():
list_options = config.options("OPTIONS")
# Flag to also produce PIXCVec file as shapefile (=True); else=False (default)
if "produce shp" in list_options:
OUT_params["flag_prod_shp"] = config.getboolean("OPTIONS", "Produce shp")
def load_config():
config = configparser.ConfigParser()
config.set('DEFAULT', 'url', 'http://127.0.0.1:8080')
return config
def save(self, file):
d = self.__dict__.copy()
config = configparser.ConfigParser()
config['all'] = d
with open(file, 'w') as f:
config.write(f)
def verify(cls):
# load default configuration file
config = ConfigParser()
config.read(os.path.join(os.path.dirname(__file__), 'config.ini'))
os_db = config.get('NSRL', 'nsrl_os_db')
mfg_db = config.get('NSRL', 'nsrl_mfg_db')
file_db = config.get('NSRL', 'nsrl_file_db')
prod_db = config.get('NSRL', 'nsrl_prod_db')
databases = [os_db, mfg_db, file_db, prod_db]
# check for configured database path
results = list(map(os.path.exists, databases))
dbs_available = reduce(lambda x, y: x or y, results, False)
if not dbs_available:
raise PluginLoadError("{0}: verify() failed because "
"databases are not available."
"".format(cls.__name__))
def __init_config(self):
config = ConfigParser.ConfigParser()
conf_file = os.path.join(os.path.dirname(os.path.abspath(__file__)), os.pardir, 'logging.conf')
if not os.path.exists(conf_file):
if log.configuration_path is None:
self.enabled = False
return
conf_file = os.path.join(log.configuration_path, 'logging.conf')
if not os.path.exists(conf_file):
conf_file = os.path.join(log.configuration_path, 'logging.conf.default')
if not os.path.exists(conf_file):
self.enabled = False
return