Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
self.path = self.path + '/*'
files = open_files(self.path, mode='rb', **options)
self.path = make_path_posix(self.path)
self.name = self.name or self.path
self.description = self.description or f'Catalog generated from all files found in {self.path}'
if not set(f.path for f in files) == set(
f.path for f in self._cat_files):
# glob changed, reload all
self._cat_files = files
self._cats.clear()
for f in files:
name = os.path.split(f.path)[-1].replace(
'.yaml', '').replace('.yml', '')
kwargs = self.kwargs.copy()
kwargs['path'] = f.path
d = make_path_posix(os.path.dirname(f.path))
if f.path not in self._cats:
entry = LocalCatalogEntry(name, "YAML file: %s" % name,
'yaml_file_cat', True,
kwargs, [], {}, self.metadata, d)
if self._flatten:
# store a concrete Catalog
try:
self._cats[f.path] = entry()
except IOError as e:
logger.info('Loading "%s" as a catalog failed: %s'
'' % (entry, e))
else:
# store a catalog entry
self._cats[f.path] = entry
for name, entry in list(self._cats.items()):
if self._flatten:
def get_dir(path):
if '://' in path:
protocol, _ = split_protocol(path)
out = get_filesystem_class(protocol)._parent(path)
if "://" not in out:
# some FSs strip this, some do not
out = protocol + "://" + out
return out
path = make_path_posix(
os.path.join(os.getcwd(), os.path.dirname(path)))
if path[-1] != '/':
path += '/'
return path
def cfile():
return make_path_posix(
os.getenv('INTAKE_CONF_FILE', posixpath.join(confdir, 'conf.yaml')))
def set_cache_dir(self, cache_dir):
for c in self.cache:
c._cache_dir = make_path_posix(cache_dir)
def set_cache_dir(self, cache_dir):
for c in self.cache:
c._cache_dir = make_path_posix(cache_dir)
# ensure that format_string is in posix format
format_string = make_path_posix(format_string)
# split the string into bits
literal_texts, field_names, format_specs, conversions = zip(*fmt.parse(format_string))
if not any(field_names):
return {}
for i, conversion in enumerate(conversions):
if conversion:
raise ValueError(('Conversion not allowed. Found on {}.'
.format(field_names[i])))
# ensure that resolved string is in posix format
resolved_string = make_path_posix(resolved_string)
# get a list of the parts that matter
bits = _get_parts_of_format_string(resolved_string, literal_texts, format_specs)
for i, (field_name, format_spec) in enumerate(zip(field_names, format_specs)):
if field_name:
try:
if format_spec.startswith('%'):
args[field_name] = datetime.strptime(bits[i], format_spec)
elif format_spec[-1] in list('bcdoxX'):
args[field_name] = int(bits[i])
elif format_spec[-1] in list('eEfFgGn'):
args[field_name] = float(bits[i])
elif format_spec[-1] == '%':
args[field_name] = float(bits[i][:-1])/100
else:
def global_data_dir():
"""Return the global Intake catalog dir for the current environment"""
prefix = False
if VIRTUALENV_VAR in os.environ:
prefix = os.environ[VIRTUALENV_VAR]
elif CONDA_VAR in os.environ:
prefix = sys.prefix
elif which('conda'):
# conda exists but is not activated
prefix = conda_prefix()
if prefix:
# conda and virtualenv use Linux-style directory pattern
return make_path_posix(os.path.join(prefix, 'share', 'intake'))
else:
return appdirs.site_data_dir(appname='intake', appauthor='intake')
"""Utility for cleaning up paths."""
storage_option = infer_storage_options(path)
protocol = storage_option['protocol']
if protocol in ('http', 'https'):
# Most FSs remove the protocol but not HTTPFS. We need to strip
# it to match properly.
path = os.path.normpath(path.replace("{}://".format(protocol), ''))
elif protocol == 'file':
# Remove trailing slashes from file paths.
path = os.path.normpath(path)
# Remove colons
path = path.replace(':', '')
# Otherwise we just make sure that path is posix
return make_path_posix(path)