inspired by book 'redis in action'
(conn, lock_name, acquire_timeout=10, lock_timeout=10)
| 17 | |
| 18 | |
| 19 | def acquire_lock(conn, lock_name, acquire_timeout=10, lock_timeout=10): |
| 20 | """inspired by book 'redis in action' """ |
| 21 | identifier = str(uuid.uuid4()) |
| 22 | lock_name = LOCKER_PREFIX + lock_name |
| 23 | end = time.time() + acquire_timeout |
| 24 | |
| 25 | while time.time() < end: |
| 26 | if conn.set(lock_name, identifier, lock_timeout, nx=True): |
| 27 | return identifier |
| 28 | elif not conn.ttl(lock_name) or conn.ttl(lock_name) == -1: |
| 29 | conn.expire(lock_name, lock_timeout) |
| 30 | time.sleep(0.1) |
| 31 | |
| 32 | return False |
| 33 | |
| 34 | |
| 35 | def release_lock(conn, lock_name, identifier): |
no outgoing calls
no test coverage detected