try

utils/queues.py

@@ -0,0 +1,64 @@
+import json
+import os
+import requests
+import time
+
+
+stage = os.getenv("STAGE", 'local')
+if stage == 'development':
+    QUEUES_URL = 'https://queues.develop.sprinthub.ru'
+elif stage == 'production':
+    QUEUES_URL = 'https://queues.sprinthub.ru'
+else:
+    QUEUES_URL = None
+
+token = os.getenv('QUEUES_TOKEN')
+
+
+class QueuesException(Exception):
+    ...
+
+
+class TasksHandlerMixin:
+    def poll(self):
+        while True:
+            if QUEUES_URL is None:
+                data = {'payload': json.loads(input('Input message: '))}
+            else:
+                response = requests.get(f'{QUEUES_URL}/api/v1/take', headers={'queue': self.queue_name, 'X-Queues-Token': token})
+                if response.status_code == 404:
+                    time.sleep(0.2)
+                    continue
+                if response.status_code == 403:
+                    raise NotImplemented('QUEUE_TOKEN is incorrect')
+                data = response.json()
+            try:
+                self.process(data['payload'])
+            except:
+                print(f'Error processing message id={data["id"]}, payload={data["payload"]}')
+                continue
+            if QUEUES_URL is None:
+                continue
+            try:
+                resp = requests.post(f'{QUEUES_URL}/api/v1/finish', json={'id': data['id']}, headers={'X-Queues-Token': token})
+                if resp.status_code != 202:
+                    raise QueuesException
+            except:
+                print(f'Failed to finish task id={data["id"]}')
+
+    @property
+    def queue_name(self):
+        raise NotImplemented
+
+    def process(self, payload):
+        raise NotImplemented
+
+
+def set_task(queue_name: str, payload: dict, seconds_to_execute: int, delay: int|None = None):
+    resp = requests.post(f'{QUEUES_URL}/api/v1/put', headers={'queue': queue_name, 'X-Queues-Token': token}, json={
+        'payload': payload,
+        'seconds_to_execute': seconds_to_execute,
+        'delay': delay,
+    })
+    if resp.status_code != 202:
+        raise QueuesException