fix

.deploy/deploy-dev.yaml

@@ -9,7 +9,8 @@ services:
     environment:
       TELEGRAM_TOKEN: $TELEGRAM_TOKEN_DEV
       STAGE: "development"
-      QUEUES_TOKEN: $QUEUES_TOKEN_DEV
+    networks:
+      - queues
     deploy:
       mode: replicated
       restart_policy:
@@ -25,7 +26,9 @@ services:
       MONGO_HOST: "mongo.develop.sprinthub.ru"
       MONGO_PASSWORD: $MONGO_PASSWORD_DEV
       STAGE: "development"
-      QUEUES_TOKEN: $QUEUES_TOKEN_DEV
+      PLATFORM_SECURITY_TOKEN: $PLATFORM_SECURITY_TOKEN
+    networks:
+      - queues
     deploy:
       mode: replicated
       restart_policy:
@@ -40,7 +43,8 @@ services:
     environment:
       TELEGRAM_TOKEN: $TELEGRAM_TOKEN_DEV
       STAGE: "development"
-      QUEUES_TOKEN: $QUEUES_TOKEN_DEV
+    networks:
+      - queues
     deploy:
       mode: replicated
       restart_policy:
@@ -71,3 +75,5 @@ services:
 networks:
   common-infra-nginx:
     external: true
+  queues:
+    external: true

.deploy/deploy-prod.yaml

@@ -9,7 +9,8 @@ services:
     environment:
       TELEGRAM_TOKEN: $TELEGRAM_TOKEN_PROD
       STAGE: "production"
-      QUEUES_TOKEN: $QUEUES_TOKEN_PROD
+    networks:
+      - queues
     deploy:
       mode: replicated
       restart_policy:
@@ -25,7 +26,9 @@ services:
       MONGO_HOST: "mongo.sprinthub.ru"
       MONGO_PASSWORD: $MONGO_PASSWORD_PROD
       STAGE: "production"
-      QUEUES_TOKEN: $QUEUES_TOKEN_PROD
+      PLATFORM_SECURITY_TOKEN: $PLATFORM_SECURITY_TOKEN
+    networks:
+      - queues
     deploy:
       mode: replicated
       restart_policy:
@@ -40,7 +43,8 @@ services:
     environment:
       TELEGRAM_TOKEN: $TELEGRAM_TOKEN_PROD
       STAGE: "production"
-      QUEUES_TOKEN: $QUEUES_TOKEN_PROD
+    networks:
+      - queues
     deploy:
       mode: replicated
       restart_policy:
@@ -71,3 +75,5 @@ services:
 networks:
   common-infra-nginx:
     external: true
+  queues:
+    external: true

daemons/worker.py

@@ -47,7 +47,7 @@ class Daemon(base.BaseDaemon, queues.TasksHandlerMixin):
     def execute(self):
         self.poll()
 
-    def reply(text: str, chat_id: int, message_id: int):
+    def reply(self, text: str, chat_id: int, message_id: int):
         queues.set_task(
             'pizda_bot_mailbox', 
             {
@@ -59,7 +59,7 @@ class Daemon(base.BaseDaemon, queues.TasksHandlerMixin):
             1,
         )
 
-    def send(text: str, chat_id: int):
+    def send(self, text: str, chat_id: int):
         queues.set_task(
             'pizda_bot_mailbox', 
             {

main.py

@@ -8,6 +8,9 @@ if arg == 'poll':
 elif arg == 'worker':
     from daemons import worker
     daemon = worker.Daemon()
+elif arg == 'mailbox':
+    from daemons import mailbox
+    daemon = mailbox.Daemon()
 else:
     from api import app
     app.run(host="0.0.0.0", port=1238)

utils/queues.py

@@ -5,14 +5,10 @@ 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'
+if stage == 'local':
+    QUEUES_URL = 'http://localhost:1239'
 else:
-    QUEUES_URL = None
-
-token = os.getenv('QUEUES_TOKEN')
+    QUEUES_URL = 'http://queues:1239'
 
 
 class QueuesException(Exception):
@@ -22,25 +18,18 @@ 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})
+            response = requests.get(f'{QUEUES_URL}/api/v1/take', headers={'queue': self.queue_name})
             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:
+            except Exception as exc:
+                print(f'Error processing message id={data["id"]}, payload={data["payload"]}, exc={exc}')
                 continue
             try:
-                resp = requests.post(f'{QUEUES_URL}/api/v1/finish', json={'id': data['id']}, headers={'X-Queues-Token': token})
+                resp = requests.post(f'{QUEUES_URL}/api/v1/finish', json={'id': data['id']})
                 if resp.status_code != 202:
                     raise QueuesException
             except:
@@ -55,7 +44,7 @@ class TasksHandlerMixin:
 
 
 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={
+    resp = requests.post(f'{QUEUES_URL}/api/v1/put', headers={'queue': queue_name}, json={
         'payload': payload,
         'seconds_to_execute': seconds_to_execute,
         'delay': delay,