/usr/lib/python2.7/dist-packages/parsec/config.py is in python-cylc 7.6.0-1.
This file is owned by root:root, with mode 0o755.
The actual contents of the file can be viewed below.
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 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 | #!/usr/bin/env python
# THIS FILE IS PART OF THE CYLC SUITE ENGINE.
# Copyright (C) 2008-2017 NIWA
#
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) 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 General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program. If not, see <http://www.gnu.org/licenses/>.
import re
from parsec import ParsecError
from parsec.fileparse import parse
from parsec.util import printcfg
from parsec.validate import validate, check_compulsory, expand, validator
from parsec.OrderedDict import OrderedDictWithDefaults
from parsec.util import replicate, itemstr
class ItemNotFoundError(ParsecError):
def __init__(self, msg):
self.msg = 'ERROR: item not found: %s' % msg
class NotSingleItemError(ParsecError):
def __init__(self, msg):
self.msg = 'ERROR: not a singular item: %s' % msg
class config(object):
"Object wrapper for parsec functions"
def __init__(self, spec, upgrader=None, output_fname=None, tvars=None):
self.sparse = OrderedDictWithDefaults()
self.dense = OrderedDictWithDefaults()
self.upgrader = upgrader
self.tvars = tvars
self.output_fname = output_fname
self.checkspec(spec)
self.spec = spec
def checkspec(self, spec, parents=[]):
"check that the file spec is a nested dict of validators"
for key, value in spec.items():
pars = parents + [key]
if isinstance(value, dict):
self.checkspec(value, pars)
else:
if not isinstance(value, validator):
raise ParsecError(
"Illegal file spec item: %s" % itemstr(
pars, repr(value)))
def loadcfg(self, rcfile, title=""):
"""Parse a config file, upgrade or deprecate items if necessary,
validate it against the spec, and if this is not the first load,
combine/override with the existing loaded config."""
sparse = parse(rcfile, self.output_fname, self.tvars)
if self.upgrader is not None:
self.upgrader(sparse, title)
self.validate(sparse)
if not self.sparse:
self.sparse = sparse
else:
# Already loaded, override with new items.
replicate(self.sparse, sparse)
def validate(self, sparse):
"Validate sparse config against the file spec."
validate(sparse, self.spec)
check_compulsory(sparse, self.spec)
def expand(self):
"Flesh out undefined items with defaults, if any, from the spec."
if not self.dense:
self.dense = expand(self.sparse, self.spec)
def get(self, keys=[], sparse=False):
"""
Retrieve items or sections, sparse or dense, by list of keys:
[sec1,sec2,item] =>
[sec1]
[[sec2]]
item = value
"""
if sparse:
cfg = self.sparse
else:
self.expand()
cfg = self.dense
parents = []
for key in keys:
try:
cfg = cfg[key]
except KeyError:
raise ItemNotFoundError(itemstr(parents, key))
else:
parents.append(key)
return cfg
def idump(self, items=[], sparse=False, pnative=False, prefix='',
oneline=False, none_str=''):
"""
items is a list of --item style inputs:
'[runtime][foo]script'.
"""
mkeys = []
null = True
for i in items:
null = False
i = i.lstrip('[')
i = i.rstrip(']')
j = re.split('\]\[*', i)
mkeys.append(j)
if null:
mkeys = [[]]
self.mdump(mkeys, sparse, pnative, prefix, oneline, none_str)
def mdump(self, mkeys=[], sparse=False, pnative=False, prefix='',
oneline=False, none_str=''):
if oneline:
items = []
for keys in mkeys:
item = self.get(keys, sparse)
if isinstance(item, list) or isinstance(item, dict):
raise NotSingleItemError(itemstr(keys))
if not item:
item = none_str or "None"
items.append(str(item))
# TODO - quote items if they contain spaces or comment delimiters?
print prefix + ' '.join(items)
else:
for keys in mkeys:
self.dump(keys, sparse, pnative, prefix, none_str)
def dump(self, keys=[], sparse=False, pnative=False, prefix='',
none_str=''):
cfg = self.get(keys, sparse)
if pnative:
print cfg
else:
printcfg(cfg, prefix=prefix, level=len(keys), none_str=none_str)
|