Coverage for portality / bll / services / concurrency_prevention.py: 92%

1from portality.core import app 

2import redis 

3 

4 

5class ConcurrencyPreventionService: 

6 def __init__(self): 

7 self.rs = redis.Redis(host=app.config.get("REDIS_HOST"), port=app.config.get("REDIS_PORT")) 

8 

9 def check_concurrency(self, key, _id): 

10 """ 

11 Checks whether concurrent request has been submitted 

12 Returns true if clash is detected 

13 """ 

14 value = self.rs.get(key) 

15 return value is not None and value != _id 

16 

17 def store_concurrency(self, key, _id, timeout=None): 

18 if timeout is None: 

19 timeout = app.config.get("UR_CONCURRENCY_TIMEOUT", 10) 

20 if timeout > 0: 

21 self.rs.set(key, _id, ex=timeout)