Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
if wanna_ipns:
ipns_key = f'{IPNS_KEYS_NAME_PREFIX}_{name}'
try:
out = config.ipfs.key.gen(ipns_key, IPNS_KEYS_TYPE)
except ipfshttpclient.exceptions.Error:
use_existing = inquirer.shortcuts.confirm(f'There is already IPNS key with name \'{ipns_key}\', '
f'do you want to use it?', default=True)
if use_existing:
keys = config.ipfs.key.list()
out = next((x for x in keys['Keys'] if x['Name'] == ipns_key), None)
if out is None:
raise exceptions.RepoException('We were not able to generate or fetch the IPNS key')
else:
while True:
ipns_key = inquirer.shortcuts.text('Then please provide non-existing name for the IPNS key')
try:
out = config.ipfs.key.gen(ipns_key, IPNS_KEYS_TYPE)
break
except ipfshttpclient.exceptions.Error:
click.echo('There is already existing key with this name!')
continue
ipns_addr = f'/ipns/{out["Id"]}/'
else:
keys = config.ipfs.key.list()
key_object = next((x for x in keys['Keys'] if x['Name'] == ipns_key), None)
if key_object is None:
def bootstrap_property(cls, name: str, category: str, message: str, value: typing.Any = None,
default: typing.Any = None,
validate: typing.Callable = None):
if value is not None:
if validate is not None and not validate(None, value):
raise exceptions.RepoException(f'Invalid {name}: {value}!')
return value
return getattr(inquirer.shortcuts, category)(message, validate=validate, default=default)
def _remove_glob(self, path: pathlib.Path, glob: str):
"""
Removes all files from path that matches the glob string.
:param path:
:param glob:
:return:
"""
for path_to_delete in path.glob(glob):
path_to_delete = path_to_delete.resolve()
if not path_to_delete.exists():
continue
if path not in path_to_delete.parents:
raise exceptions.RepoException(
f'Trying to delete file outside the repo temporary directory! {path_to_delete}')
if path_to_delete.is_file():
path_to_delete.unlink()
else:
shutil.rmtree(str(path_to_delete))
if build_bin is None:
build_bin = inquirer.shortcuts.text('Path to build binary, if you want to do some pre-processing '
'before publishing', default='')
if after_publish_bin is None:
after_publish_bin = inquirer.shortcuts.text('Path to after-publish binary, if you want to do some '
'actions after publishing', default='')
if publish_dir is None:
publish_dir = inquirer.shortcuts.text('Directory to be published inside the repo. Path related to the root '
'of the repo', default='/')
ipns_lifetime = ipns_lifetime or '24h'
if not validate_time_span(ipns_lifetime):
raise exceptions.RepoException('Passed lifetime is not valid! Supported units are: h(our), m(inute), '
's(seconds)!')
ipns_ttl = ipns_ttl or '15m'
if not validate_time_span(ipns_ttl):
raise exceptions.RepoException('Passed ttl is not valid! Supported units are: h(our), m(inute), '
's(seconds)!')
if ipns_key is None and after_publish_bin is None and zone_id is None:
raise exceptions.RepoException(
'You have choose not to use IPNS, not modify DNSLink entry on CloudFlare and you also have not '
'specified any after publish command. This does not make sense! What do you want to do '
'with this setting?! I have no idea, so aborting!')
return cls(config=config, name=name, git_repo_url=git_repo_url, branch=branch, secret=secret, pin=pin,
publish_dir=publish_dir,
ipns_key=ipns_key, ipns_addr=ipns_addr, build_bin=build_bin, after_publish_bin=after_publish_bin,
def __init__(self, git_repo_url, **kwargs):
if not is_github_url(git_repo_url):
raise exceptions.RepoException('The passed Git repo URL is not related to GitHub!')
super().__init__(git_repo_url=git_repo_url, **kwargs)
:param cwd: Directory in which the binary will be invoked
:param cmd: Binary definition invoked with shell
:param args:
:raises exceptions.RepoException: If the binary exited with non-zero status
:return:
"""
os.chdir(str(cwd))
full_cmd = f'{cmd} {" ".join(args)}'
logger.info(f'Running shell command "{full_cmd}" with cwd={cwd}')
r = subprocess.run(full_cmd, shell=True, capture_output=True)
if r.returncode != 0:
r.stderr and logger.debug(f'STDERR: {r.stderr.decode("utf-8")}')
r.stdout and logger.debug(f'STDOUT: {r.stdout.decode("utf-8")}')
raise exceptions.RepoException(f'\'{cmd}\' binary exited with non-zero code!')
if after_publish_bin is None:
after_publish_bin = inquirer.shortcuts.text('Path to after-publish binary, if you want to do some '
'actions after publishing', default='')
if publish_dir is None:
publish_dir = inquirer.shortcuts.text('Directory to be published inside the repo. Path related to the root '
'of the repo', default='/')
ipns_lifetime = ipns_lifetime or '24h'
if not validate_time_span(ipns_lifetime):
raise exceptions.RepoException('Passed lifetime is not valid! Supported units are: h(our), m(inute), '
's(seconds)!')
ipns_ttl = ipns_ttl or '15m'
if not validate_time_span(ipns_ttl):
raise exceptions.RepoException('Passed ttl is not valid! Supported units are: h(our), m(inute), '
's(seconds)!')
if ipns_key is None and after_publish_bin is None and zone_id is None:
raise exceptions.RepoException(
'You have choose not to use IPNS, not modify DNSLink entry on CloudFlare and you also have not '
'specified any after publish command. This does not make sense! What do you want to do '
'with this setting?! I have no idea, so aborting!')
return cls(config=config, name=name, git_repo_url=git_repo_url, branch=branch, secret=secret, pin=pin,
publish_dir=publish_dir,
ipns_key=ipns_key, ipns_addr=ipns_addr, build_bin=build_bin, after_publish_bin=after_publish_bin,
republish=republish, ipns_lifetime=ipns_lifetime, ipns_ttl=ipns_ttl, dns_id=dns_id,
zone_id=zone_id)
def from_toml_dict(cls, data: dict, config: config_module.Config) -> 'GenericRepo':
"""
Deserialize the passed data dict of TOML config into instance
:param data:
:param config:
:return:
"""
try:
return cls(config=config, **helpers.flatten(data))
except TypeError:
raise exceptions.RepoException('Passed repo\'s data are not valid for creating valid Repo instance!')