Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def _add_from_url(self, dataset, path, url, nocopy=False, **kwargs):
"""Process an add from url and return the location on disk."""
u = parse.urlparse(url)
if u.scheme not in Dataset.SUPPORTED_SCHEMES:
raise NotImplementedError(
'{} URLs are not supported'.format(u.scheme)
)
# Respect the directory struture inside the source path.
relative_to = kwargs.pop('relative_to', None)
if relative_to:
dst_path = Path(url).resolve().absolute().relative_to(
Path(relative_to).resolve().absolute()
)
else:
dst_path = os.path.basename(url)
dst = path.joinpath(dst_path).absolute()
if u.scheme in ('', 'file'):
src = Path(u.path).absolute()
# if we have a directory, recurse
if src.is_dir():
files = {}
dst.mkdir(parents=True, exist_ok=True)
for f in src.iterdir():
files.update(
self._add_from_url(
def find_project_config_path(path=None):
"""Find project config path."""
path = Path(path) if path else Path.cwd()
abspath = path.absolute()
project_path = get_project_config_path(abspath)
if project_path:
return project_path
for parent in abspath.parents:
project_path = get_project_config_path(parent)
if project_path:
return project_path
def create_project_config_path(
path, mode=0o777, parents=False, exist_ok=False
):
"""Create new project configuration folder."""
# FIXME check default directory mode
project_path = Path(path).absolute().joinpath(RENKU_HOME)
project_path.mkdir(mode=mode, parents=parents, exist_ok=exist_ok)
return str(project_path)
'{} URLs are not supported'.format(u.scheme)
)
# Respect the directory struture inside the source path.
relative_to = kwargs.pop('relative_to', None)
if relative_to:
dst_path = Path(url).resolve().absolute().relative_to(
Path(relative_to).resolve().absolute()
)
else:
dst_path = os.path.basename(url)
dst = path.joinpath(dst_path).absolute()
if u.scheme in ('', 'file'):
src = Path(u.path).absolute()
# if we have a directory, recurse
if src.is_dir():
files = {}
dst.mkdir(parents=True, exist_ok=True)
for f in src.iterdir():
files.update(
self._add_from_url(
dataset,
dst,
f.absolute().as_posix(),
nocopy=nocopy
)
)
return files
def get_project_config_path(path=None):
"""Return project configuration folder if exist."""
project_path = Path(path or '.').absolute().joinpath(RENKU_HOME)
if project_path.exists() and project_path.is_dir():
return str(project_path)
# FIXME: do a proper check that the repos are not the same
if submodule_name not in (s.name for s in self.git.submodules):
# new submodule to add
if u.scheme == 'git+ssh':
url = 'git@{netloc}:{path}'.format(
netloc=u.netloc, path=u.path[1:]
)
self.git.create_submodule(
name=submodule_name, path=submodule_path.as_posix(), url=url
)
src = submodule_path / (target or '')
if target and relative_to:
relative_to = Path(relative_to)
if relative_to.is_absolute():
assert u.scheme in {
'', 'file'
}, ('Only relative paths can be used with URLs.')
target = (Path(url).resolve().absolute() / target).relative_to(
relative_to.resolve()
)
else:
# src already includes target so we do not have to append it
target = src.relative_to(submodule_path / relative_to)
# link the target into the data directory
dst = self.path / path / (target or '')
# if we have a directory, recurse
if src.is_dir():
def _add_from_url(self, dataset, path, url, nocopy=False, **kwargs):
"""Process an add from url and return the location on disk."""
u = parse.urlparse(url)
if u.scheme not in Dataset.SUPPORTED_SCHEMES:
raise NotImplementedError(
'{} URLs are not supported'.format(u.scheme)
)
# Respect the directory struture inside the source path.
relative_to = kwargs.pop('relative_to', None)
if relative_to:
dst_path = Path(url).resolve().absolute().relative_to(
Path(relative_to).resolve().absolute()
)
else:
dst_path = os.path.basename(url)
dst = path.joinpath(dst_path).absolute()
if u.scheme in ('', 'file'):
src = Path(u.path).absolute()
# if we have a directory, recurse
if src.is_dir():
files = {}
dst.mkdir(parents=True, exist_ok=True)
for f in src.iterdir():
files.update(
submodule_path = self.renku_path / 'vendors' / (u.netloc or 'local')
# Respect the directory struture inside the source path.
relative_to = kwargs.get('relative_to', None)
if u.scheme in ('', 'file'):
warnings.warn('Importing local git repository, use HTTPS')
# determine where is the base repo path
r = Repo(url, search_parent_directories=True)
src_repo_path = Path(r.git_dir).parent
submodule_name = src_repo_path.name
submodule_path = submodule_path / str(src_repo_path).lstrip('/')
# if repo path is a parent, rebase the paths and update url
if src_repo_path != Path(u.path):
top_target = Path(
u.path
).resolve().absolute().relative_to(src_repo_path)
if target:
target = top_target / target
else:
target = top_target
url = src_repo_path.as_posix()
elif u.scheme in {'http', 'https', 'git+https', 'git+ssh'}:
submodule_name = os.path.splitext(os.path.basename(u.path))[0]
submodule_path = submodule_path.joinpath(
os.path.dirname(u.path).lstrip('/'), submodule_name
)
else:
raise NotImplementedError(
'Scheme {} not supported'.format(u.scheme)
)