Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def _acquire_non_blocking(acquire, timeout, retry_period, path):
if retry_period is None:
retry_period = 0.05
start_time = time.time()
while True:
success = acquire()
if success:
return
elif (timeout is not None and
time.time() - start_time > timeout):
raise LockError("Couldn't lock {0}".format(path))
else:
time.sleep(retry_period)