fix

utils/locks.py

@@ -0,0 +1,27 @@
+import requests
+
+
+LOCKS_URL = 'http://locks'
+
+
+class Conflict(Exception):
+    pass
+
+
+def acquire(name: str, ttl: int = 1):
+    resp = requests.post(f'{LOCKS_URL}/api/v1/acquire', json={
+        'name': name,
+        'ttl': ttl,
+    })
+    if resp.status_code == 409:
+        raise Conflict
+    if resp.status_code != 202:
+        raise Exception
+
+
+def release(name: str):
+    resp = requests.post(f'{LOCKS_URL}/api/v1/release', json={
+        'name': name,
+    })
+    if resp.status_code != 202:
+        raise Exception

utils/queues.py

@@ -23,6 +23,7 @@ class TasksHandlerMixin:
                 time.sleep(0.2)
                 continue
             try:
+                self.before_execute(task)
                 self.process(task['payload'])
             except Exception as exc:
                 print(f'Error processing message id={task["id"]}, payload={task["payload"]}, exc={exc}')
@@ -31,9 +32,16 @@ class TasksHandlerMixin:
                 resp = requests.post(f'{QUEUES_URL}/api/v1/finish', json={'id': task['id']})
                 if resp.status_code != 202:
                     raise QueuesException
+                self.after_execute(task)
             except:
                 print(f'Failed to finish task id={task["id"]}')
 
+    def before_execute(self, task: dict):
+        pass
+
+    def after_execute(self, task: dict):
+        pass
+
     @property
     def queue_name(self):
         raise NotImplemented