Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def __enter__(self):
current_time = call_time = time.time()
while call_time + self._timeout > current_time:
self._lockfile = open(self._filepath, 'w')
try:
portalocker.lock(self._lockfile, portalocker.constants.LOCK_NB | portalocker.constants.LOCK_EX)
return self
except portalocker.exceptions.LockException:
pass
current_time = time.time()
check_interval = self._check_interval if self._timeout > self._check_interval else self._timeout
time.sleep(check_interval)
raise RuntimeError('Timeout was reached')
runtime exception
"""
for _ in range(self.max_tries, 0, -1):
if self.token.is_access_expired:
try:
with Lock(self.token_path, 'r+',
fail_when_locked=True, timeout=0):
log.debug('Locked oauth token file')
if con.refresh_token() is False:
raise RuntimeError('Token Refresh Operation not '
'working')
log.info('New oauth token fetched')
log.debug('Unlocked oauth token file')
return None
except LockException:
self.fs_wait = True
log.warning('Oauth file locked. Sleeping for 2 seconds...'
f'retrying {_ - 1} more times.')
time.sleep(2)
log.debug('Waking up and rechecking token file for update'
' from other instance...')
self.token = self.load_token()
else:
log.info('Token was refreshed by another instance...')
self.fs_wait = False
return False
# if we exit the loop, that means we were locked out of the file after
# multiple retries give up and throw an error - something isn't right
raise RuntimeError('Could not access locked token file after '
f'{self.max_tries}')