Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
gen_check_result = gen_check.get_primary_result()
# Checks expire after 280 seconds, so keep track of how long this task has lasted.
start_time = time.time()
time_expired = False
# Purge the deleted files.
for view_conf_uuid in gen_check_result["full_output"]["items_to_purge"]:
# If we've taken more than 270 seconds to complete, break immediately
if time.time() - start_time > 270:
time_expired = True
break
purge_response = {}
try:
purge_response = ff_utils.purge_metadata(view_conf_uuid, key=connection.ff_keys)
except Exception as exc:
purge_response['comment'] = str(exc)
if purge_response.get('status') == 'success':
action_logs['items_purged'].append(view_conf_uuid)
else:
action_logs['failed_to_purge'][view_conf_uuid] = purge_response['comment']
action.status = 'DONE'
action.output = action_logs
return action
t0 = time.time()
check.full_output = {} # purged items by item type
search = '/search/?type=TrackingItem&tracking_type=download_tracking&status=deleted&field=uuid&limit=300'
search_res = ff_utils.search_metadata(search, key=connection.ff_keys)
search_uuids = [res['uuid'] for res in search_res]
client = es_utils.create_es_client(connection.ff_es, True)
# a bit convoluted, but we want the frame=raw, which does not include uuid
# use get_es_metadata to handle this. Use it as a generator
for to_purge in ff_utils.get_es_metadata(search_uuids, es_client=client, is_generator=True,
key=connection.ff_keys):
if round(time.time() - t0, 2) > time_limit:
break
purge_properties = to_purge['properties']
purge_properties['uuid'] = to_purge['uuid'] # add uuid to frame=raw
try:
purge_res = ff_utils.purge_metadata(to_purge['uuid'], key=connection.ff_keys)
except Exception as exc:
purge_status = 'error'
purge_detail = str(exc)
else:
purge_status = purge_res['status']
purge_detail = purge_properties if purge_status == 'success' else purge_res
purge_record = {'uuid': to_purge['uuid'], 'result': purge_detail}
if to_purge['item_type'] not in check.full_output:
check.full_output[to_purge['item_type']] = {}
if purge_status not in check.full_output[to_purge['item_type']]:
check.full_output[to_purge['item_type']][purge_status] = []
check.full_output[to_purge['item_type']][purge_status].append(purge_record)
purge_out_str = '. '.join(['%s: %s' % (it, len(check.full_output[it]['success']))
for it in check.full_output if check.full_output[it].get('success')])
check.description = 'Purged: ' + purge_out_str + '. Search used: %s' % search
if any([it for it in check.full_output if check.full_output[it].get('error')]):