Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def convert_csv_string_to_list(value, delimiter=',', trim=False):
"""
Convert comma or other character delimited strings to a list.
:param value: The value to convert.f
:param delimiter: Optionally Change the default delimiter ',' if required.
:param trim: Optionally trim the individual list items.
:return: The delimited value as a list.
"""
if not isinstance(value, (string_types, text_type)):
return value
with suppress(AttributeError, ValueError):
value = value.split(delimiter) if value else []
if trim:
value = [_.strip() for _ in value]
return value
def backupVersionedFile(old_file, version):
"""Back up an old version of a file.
:param old_file: Original file, to take a backup from
:param version: Version of file to store in backup
:return: True if success, False if failure
"""
num_tries = 0
with suppress(TypeError):
version = u'.'.join([str(i) for i in version]) if not isinstance(version, str) else version
new_file = u'{old_file}.v{version}'.format(old_file=old_file, version=version)
while not os.path.isfile(new_file):
if not os.path.isfile(old_file):
logger.debug(u"Not creating backup, {file} doesn't exist", file=old_file)
break
try:
logger.debug(u'Trying to back up {old} to new', old=old_file, new=new_file)
shutil.copy(old_file, new_file)
logger.debug(u"Backup done")
break
except Exception as e:
logger.warning(u'Error while trying to back up {old} to {new} : {error!r}',
def restore_versioned_file(backup_file, version):
"""Restore a file version to original state.
For example sickbeard.db.v41 passed with version int(41), will translate back to sickbeard.db.
sickbeard.db.v41. passed with version tuple(41,2), will translate back to sickbeard.db.
:param backup_file: File to restore
:param version: Version of file to restore
:return: True on success, False on failure
"""
num_tries = 0
with suppress(TypeError):
version = '.'.join([str(i) for i in version]) if not isinstance(version, str) else version
new_file, _ = backup_file[0:backup_file.find(u'v{version}'.format(version=version))]
restore_file = backup_file
if not os.path.isfile(new_file):
log.debug(u"Not restoring, {file} doesn't exist", {'file': new_file})
return False
try:
log.debug(u'Trying to backup {file} to {file}.r{version} before '
u'restoring backup', {'file': new_file, 'version': version})
shutil.move(new_file, new_file + '.' + 'r' + str(version))
except OSError as error:
log.warning(u'Error while trying to backup DB file {name} before'
def backup_versioned_file(old_file, version):
"""Back up an old version of a file.
:param old_file: Original file, to take a backup from
:param version: Version of file to store in backup
:return: True if success, False if failure
"""
num_tries = 0
with suppress(TypeError):
version = u'.'.join([str(i) for i in version]) if not isinstance(version, str) else version
new_file = u'{old_file}.v{version}'.format(old_file=old_file, version=version)
while not os.path.isfile(new_file):
if not os.path.isfile(old_file):
log.debug(u"Not creating backup, {old_file} doesn't exist",
{'old_file': old_file})
break
try:
log.debug(u'Trying to back up {old} to new',
{'old': old_file, 'new': new_file})
shutil.copy(old_file, new_file)
log.debug(u"Backup done")
break
def restoreVersionedFile(backup_file, version):
"""Restore a file version to original state.
For example sickbeard.db.v41 passed with version int(41), will translate back to sickbeard.db.
sickbeard.db.v41. passed with version tuple(41,2), will translate back to sickbeard.db.
:param backup_file: File to restore
:param version: Version of file to restore
:return: True on success, False on failure
"""
num_tries = 0
with suppress(TypeError):
version = '.'.join([str(i) for i in version]) if not isinstance(version, str) else version
new_file, _ = backup_file[0:backup_file.find(u'v{version}'.format(version=version))]
restore_file = backup_file
if not os.path.isfile(new_file):
logger.debug(u"Not restoring, %s doesn't exist" % new_file)
return False
try:
logger.debug(u"Trying to backup %s to %s.r%s before restoring backup" % (new_file, new_file, version))
shutil.move(new_file, new_file + '.' + 'r' + str(version))
except Exception as e:
logger.warning(u"Error while trying to backup DB file %s before proceeding with restore: %r" %
(restore_file, ex(e)))
def release_lock(self, path=None):
"""Release the lock on the currently-loaded session data."""
self.lock.close()
with contextlib2.suppress(FileNotFoundError):
os.remove(self.lock._path)
self.locked = False
def downgrade(cls, upd):
upd.print_log('Restore kubelet config...')
with suppress(Exception):
cls._restore_backup_remote(cls.KUBELET_CONFIG_FILE)
helpers.run('systemctl restart kubelet')
def upgrade(cls, upd):
upd.print_log('Update kubelet config...')
with suppress(Exception):
cls._backup_file_remote(cls.KUBELET_CONFIG_FILE)
cls._replace_str_in_file_remote(
cls.KUBELET_CONFIG_FILE,
'--cadvisor_port=\S+', '')
cls._replace_str_in_file_remote(
cls.KUBELET_CONFIG_FILE,
'KUBELET_ADDRESS=".*"', 'KUBELET_ADDRESS="0.0.0.0"')
helpers.run('systemctl restart kubelet')
def release_lock(self, path=None):
"""Release the lock on the currently-loaded session data."""
self.lock.close()
with contextlib2.suppress(FileNotFoundError):
os.remove(self.lock._path)
self.locked = False