Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def get_directories_in_properties(plugin_path):
build_properties_path = os.path.join(plugin_path, 'build.properties')
if not os.path.isfile(build_properties_path):
raise InvalidPluginException('missing build.properties in the plugin %s' % red(plugin_name(plugin_path)))
build_properties = javaproperties.load(open(build_properties_path, 'r'))
source_directory_in_properties = build_properties['source..'] if 'source..' in build_properties else None
output_directory_in_properties = build_properties['output..'] if 'output..' in build_properties else None
if source_directory_in_properties is not None and source_directory_in_properties.endswith('/'):
source_directory_in_properties = source_directory_in_properties[:-1]
if output_directory_in_properties is not None and output_directory_in_properties.endswith('/'):
output_directory_in_properties = output_directory_in_properties[:-1]
return source_directory_in_properties, output_directory_in_properties
"""
Return a MavenPom object from a POM file at `location` or provided as a
`text` string.
"""
if location and check_is_pom and not is_pom(location):
return
pom = MavenPom(location, text)
if not extra_properties:
extra_properties = {}
# do we have a pom.properties file side-by-side?
if location and os.path.exists(location):
parent = fileutils.parent_directory(location)
pom_properties = os.path.join(parent, 'pom.properties')
if os.path.exists(pom_properties):
with open(pom_properties) as props:
properties = javaproperties.load(props) or {}
if TRACE:
logger.debug('_get_mavenpom: properties: {}'.format(repr(properties)))
extra_properties.update(properties)
pom.resolve(**extra_properties)
# TODO: we cannot do much without these??
if check_is_pom and not has_basic_pom_attributes(pom):
if TRACE:
logger.debug('_get_mavenpom: has_basic_pom_attributes: {}'.format(
has_basic_pom_attributes(pom)))
return
return pom
def read_local_config(config_file):
if not os.path.exists(config_file):
raise KafkaStreamsError(f'Config file {config_file} does not exist')
with open(config_file, 'r') as cf:
props = javaproperties.load(cf)
for k, v in props.items():
ku = k.upper().replace('.', '_')
if ku not in globals().keys():
raise KafkaStreamsError(f'Unrecognised property {k} read from config file {config_file}')
globals()[ku] = v
log.debug('Config from "%s": %s = %s', config_file, k, v)
def __read_kv_from_file(file_path, format_, separator=None, prefix_to_add="", depth=None):
config_data = {}
try:
with io.open(file_path, 'r', encoding=__check_file_encoding(file_path)) as config_file:
if format_ == 'json':
config_data = json.load(config_file)
elif format_ == 'yaml':
for yaml_data in list(yaml.load_all(config_file)):
config_data.update(yaml_data)
elif format_ == 'properties':
config_data = javaproperties.load(config_file)
except ValueError:
raise CLIError(
'The input is not a well formatted %s file.' % (format_))
except OSError:
raise CLIError('File is not available.')
flattened_data = {}
index = 0
is_list = isinstance(config_data, list)
for key in config_data:
if is_list:
__flatten_key_value(prefix_to_add + str(index), key, flattened_data,
sys.maxsize if depth is None else int(depth), separator)
index += 1
else:
__flatten_key_value(
prefix_to_add + key, config_data[key], flattened_data, sys.maxsize if depth is None else int(depth), separator)