fix
All checks were successful
Deploy Dev / Build (pull_request) Successful in 4s
Deploy Dev / Push (pull_request) Successful in 8s
Deploy Dev / Deploy dev (pull_request) Successful in 17s

This commit is contained in:
2024-11-22 01:47:46 +03:00
parent 17c9b1b693
commit ac1c8745dc
5 changed files with 35 additions and 34 deletions

View File

@@ -1,18 +1,13 @@
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'
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,23 +17,16 @@ 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()
response = requests.get(f'{QUEUES_URL}/api/v1/take', headers={'queue': self.queue_name})
if response.status_code == 404:
time.sleep(0.2)
continue
data = response.json()
try:
self.process(data['payload'])
except Exception as exc:
print(f'Error processing message id={data["id"]}, payload={data["payload"]}, exc={exc}')
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:
@@ -55,7 +43,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,