Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
else:
# break never reached
raise error
except Exception as exc:
# cleanup in case of fault or cancellation will run in background
async def cleanup():
self.log.debug('Cleaning up lock "%s"', resource)
with contextlib.suppress(LockError):
await self.redis.unset_lock(resource, lock_identifier)
asyncio.ensure_future(cleanup())
raise
return Lock(self, resource, lock_identifier, valid=True)
async def is_locked(self, resource_or_lock):
"""
Checks if the resource or the lock is locked by any redlock instance.
:param resource_or_lock: resource name or aioredlock.Lock instance
:returns: True if locked else False
"""
if isinstance(resource_or_lock, Lock):
resource = resource_or_lock.resource
elif isinstance(resource_or_lock, str):
resource = resource_or_lock
else:
raise TypeError(
'Argument should be ether aioredlock.Lock instance or string, '
'%s is given.', type(resource_or_lock)
)
return await self.redis.is_locked(resource)