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_installationpath_from_url(url):
"""Returns a relative path derived from the trailing end of a URL
This can be used to determine an installation path of a Dataset
from a URL, analog to what `git clone` does.
"""
ri = RI(url)
if isinstance(ri, (URL, DataLadRI)): # decode only if URL
path = ri.path.rstrip('/')
path = urlunquote(path) if path else ri.hostname
else:
path = url
path = path.rstrip('/')
if '/' in path:
path = path.split('/')
if path[-1] == '.git':
path = path[-2]
else:
path = path[-1]
if path.endswith('.git'):
path = path[:-4]
return path
message="installation `source` and destination `path` are identical. "
"If you are trying to add a subdataset simply use the `add` command")
return
# resolve the target location (if local) against the provided dataset
# or CWD:
if path is not None:
# MIH everything in here is highly similar to what common
# interface helpers do (or should/could do), but at the same
# is very much tailored to just apply to `install` -- I guess
# it has to stay special
# Should work out just fine for regular paths, so no additional
# conditioning is necessary
try:
path_ri = RI(path)
except Exception as e:
raise ValueError(
"invalid path argument {}: ({})".format(path, exc_str(e)))
try:
# Wouldn't work for SSHRI ATM, see TODO within SSHRI
# yoh: path should be a local path, and mapping note within
# SSHRI about mapping localhost:path to path is kinda
# a peculiar use-case IMHO
# TODO Stringification can be removed once PY35 is no longer
# supported
path = str(resolve_path(path_ri.localpath, dataset))
# any `path` argument that point to something local now
# resolved and is no longer a URL
except ValueError:
# `path` is neither a valid source nor a local path.
# TODO: The only thing left is a known subdataset with a
src : string or RI
Full or relative (then considered within base_url if provided) path
base_url : string or RI, optional
alternate_suffix : bool
Whether to generate URL candidates with and without '/.git' suffixes.
Returns
-------
candidates : list of str
List of RIs (path, url, ssh targets) to try to install from
"""
candidates = []
ri = RI(src)
if isinstance(ri, PathRI) and not isabs(ri.path) and base_url:
ri = RI(base_url)
if ri.path.endswith('/.git'):
base_path = ri.path[:-5]
base_suffix = '.git'
else:
base_path = ri.path
base_suffix = ''
if isinstance(ri, PathRI):
# this is a path, so stay native
ri.path = normpath(opj(base_path, src, base_suffix))
else:
# we are handling a URL, use POSIX path conventions
ri.path = posixpath.normpath(
posixpath.join(base_path, src, base_suffix))
src = str(ri)
def _check_and_update_remote_server_info(ds, remote):
# if we managed to copy to "http" url we should should try to trigger git
# update-server-info hook on the remote if there was ssh annexurl defined
# for it. Apparently we do that already in create_sibling ones, but here
# we need more checks and preparation
remote_url = ds.repo.config.get('remote.%s.url' % remote, None)
if remote_url:
remote_url = RI(remote_url)
if isinstance(remote_url, URL) and remote_url.scheme in (
'http', 'https'):
remote_annexurl = ds.repo.config.get('remote.%s.annexurl' % remote,
None)
if remote_annexurl:
remote_annexurl_ri = RI(remote_annexurl)
if is_ssh(remote_annexurl_ri):
ssh = ssh_manager.get_connection(remote_annexurl_ri)
ssh('git -C {} update-server-info'.format(
sh_quote(remote_annexurl_ri.path)))
return True
else:
lgr.debug(
"There is no annexurl defined but not ssh: %s, "
"dunno if "
"we could/should do anything", remote_annexurl
Parameters
----------
src : string or RI
Full or relative (then considered within base_url if provided) path
base_url : string or RI, optional
alternate_suffix : bool
Whether to generate URL candidates with and without '/.git' suffixes.
Returns
-------
candidates : list of str
List of RIs (path, url, ssh targets) to try to install from
"""
candidates = []
ri = RI(src)
if isinstance(ri, PathRI) and not isabs(ri.path) and base_url:
ri = RI(base_url)
if ri.path.endswith('/.git'):
base_path = ri.path[:-5]
base_suffix = '.git'
else:
base_path = ri.path
base_suffix = ''
if isinstance(ri, PathRI):
# this is a path, so stay native
ri.path = normpath(opj(base_path, src, base_suffix))
else:
# we are handling a URL, use POSIX path conventions
ri.path = posixpath.normpath(
posixpath.join(base_path, src, base_suffix))
def _get_git_url_from_source(source):
"""Return URL for cloning associated with a source specification
For now just resolves DataLadRIs
"""
# TODO: Probably RF this into RI.as_git_url(), that would be overridden
# by subclasses or sth. like that
if not isinstance(source, RI):
source_ri = RI(source)
else:
source_ri = source
if isinstance(source_ri, DataLadRI):
# we have got our DataLadRI as the source, so expand it
source = source_ri.as_git_url()
else:
source = str(source_ri)
return source
def is_url(ri):
"""Returns whether argument is a resource identifier what datalad should treat as a URL
This includes ssh "urls" which git understands.
Parameters
----------
ri : str or RI
The resource identifier (as a string or RI) to "analyze"
"""
if not isinstance(ri, RI):
try:
ri = RI(ri)
except: # MIH: MemoryError?
return False
return isinstance(ri, (URL, SSHRI))
def __new__(cls, ri=None, **kwargs):
"""Used as a possible factory for known RI types
Returns
-------
RI
uninitialized RI object of appropriate class with _str
set to string representation if was provided
"""
if cls is RI and ri is not None:
# RI class was used as a factory
cls = _guess_ri_cls(ri)
if cls is RI:
# should we fail or just pretend we are nothing??? ;-) XXX
raise ValueError("Could not deduce RI type for %r" % (ri,))
ri_obj = super(RI, cls).__new__(cls)
# Store internally original str
ri_obj._str = ri
return ri_obj
def get_connection(self, url):
"""Get a singleton, representing a shared ssh connection to `url`
Parameters
----------
url: str
ssh url
Returns
-------
SSHConnection
"""
# parse url:
from datalad.support.network import RI, is_ssh
if isinstance(url, RI):
sshri = url
else:
if ':' not in url and '/' not in url:
# it is just a hostname
lgr.debug("Assuming %r is just a hostname for ssh connection",
url)
url += ':'
sshri = RI(url)
if not is_ssh(sshri):
raise ValueError("Unsupported SSH URL: '{0}', use "
"ssh://host/path or host:path syntax".format(url))
from datalad import cfg
identity_file = cfg.get("datalad.ssh.identityfile")