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_start_date(date):
# start_date format "%Y-%m-%dT%H:%M:%S"
now = datetime.datetime.now()
start_date = now + datetime.timedelta(0, 2, 0)
if (utils.date_to_timestamp(date) >
utils.date_to_timestamp(now.isoformat().split('.')[0])):
start_date = datetime.datetime.strptime(
date, "%Y-%m-%dT%H:%M:%S")
return start_date
def get_backups_from_timestamp(backups, restore_from_timestamp):
for backup in backups:
backup_created_date = backup.created_at.split('.')[0]
backup_created_timestamp = (
utils.date_to_timestamp(backup_created_date))
if backup_created_timestamp >= restore_from_timestamp:
yield backup
def get_end_date(start, end):
# start end format "%Y-%m-%dT%H:%M:%S"
end_date = datetime.datetime.strptime(end, "%Y-%m-%dT%H:%M:%S")
if (utils.date_to_timestamp(start) > utils.date_to_timestamp(end)):
end_date = None
return end_date
kwargs_date = {}
def execute(self):
# remove backups by freezer admin action
backup_media = self.conf.backup_media
if backup_media == 'cindernative':
admin_os = admin.AdminOs(self.conf.client_manager)
admin_os.del_off_limit_fullbackup(
self.conf.cindernative_vol_id,
self.conf.fullbackup_rotation)
return {}
if self.conf.remove_from_date:
timestamp = utils.date_to_timestamp(self.conf.remove_from_date)
else:
timestamp = datetime.datetime.now() - \
datetime.timedelta(days=float(self.conf.remove_older_than))
timestamp = int(time.mktime(timestamp.timetuple()))
if self.conf.backup_media == 'cinder':
if self.conf.cinder_vol_id:
old_backups = self.get_cinder_old_backups(
timestamp,
self.conf.cinder_vol_id
)
self.remove_backup_dirs(old_backups,
self.conf.cinder_vol_id)
return {}
else:
def execute(self):
conf = self.conf
LOG.info('Executing Restore...')
restore_timestamp = None
restore_abs_path = conf.restore_abs_path
if conf.restore_from_date:
restore_timestamp = utils.date_to_timestamp(conf.restore_from_date)
if conf.backup_media == 'fs':
self.engine.restore(
hostname_backup_name=self.conf.hostname_backup_name,
restore_resource=restore_abs_path,
overwrite=conf.overwrite,
recent_to_date=restore_timestamp,
backup_media=conf.mode)
try:
if conf.consistency_checksum:
backup_checksum = conf.consistency_checksum
restore_checksum = checksum.CheckSum(restore_abs_path,
ignorelinks=True)
if restore_checksum.compare(backup_checksum):
LOG.info('Consistency check success.')
else: