Compare commits
9 Commits
queues
...
889a20609f
| Author | SHA1 | Date | |
|---|---|---|---|
| 889a20609f | |||
| 1c3d1566e5 | |||
| 2b88479cec | |||
| 7238d725dd | |||
| 4a3fa3c2ab | |||
| 45e1c0fe8a | |||
| 989f7026ad | |||
| 90378be7c5 | |||
| 95e4f5e8da |
@@ -28,7 +28,6 @@ services:
|
|||||||
PLATFORM_SECURITY_TOKEN: $PLATFORM_SECURITY_TOKEN
|
PLATFORM_SECURITY_TOKEN: $PLATFORM_SECURITY_TOKEN
|
||||||
networks:
|
networks:
|
||||||
- queues-development
|
- queues-development
|
||||||
- configurator
|
|
||||||
command: worker
|
command: worker
|
||||||
deploy:
|
deploy:
|
||||||
mode: replicated
|
mode: replicated
|
||||||
@@ -115,5 +114,3 @@ networks:
|
|||||||
external: true
|
external: true
|
||||||
queues-development:
|
queues-development:
|
||||||
external: true
|
external: true
|
||||||
configurator:
|
|
||||||
external: true
|
|
||||||
|
|||||||
@@ -28,7 +28,6 @@ services:
|
|||||||
PLATFORM_SECURITY_TOKEN: $PLATFORM_SECURITY_TOKEN
|
PLATFORM_SECURITY_TOKEN: $PLATFORM_SECURITY_TOKEN
|
||||||
networks:
|
networks:
|
||||||
- queues
|
- queues
|
||||||
- configurator
|
|
||||||
command: worker
|
command: worker
|
||||||
deploy:
|
deploy:
|
||||||
mode: replicated
|
mode: replicated
|
||||||
@@ -128,5 +127,3 @@ networks:
|
|||||||
external: true
|
external: true
|
||||||
queues:
|
queues:
|
||||||
external: true
|
external: true
|
||||||
configurator:
|
|
||||||
external: true
|
|
||||||
|
|||||||
@@ -25,7 +25,4 @@ class Daemon(base.Daemon, queues.TasksHandlerMixin):
|
|||||||
reply_markup = payload.get('reply_markup')
|
reply_markup = payload.get('reply_markup')
|
||||||
if reply_markup:
|
if reply_markup:
|
||||||
body['reply_markup'] = reply_markup
|
body['reply_markup'] = reply_markup
|
||||||
try:
|
self.bot.send_message(**body, parse_mode='Markdown')
|
||||||
self.bot.send_message(**body, parse_mode='Markdown')
|
|
||||||
except Exception as exc:
|
|
||||||
print('Error', str(exc))
|
|
||||||
|
|||||||
@@ -15,8 +15,11 @@ class PlatformClient:
|
|||||||
self.stage = stage
|
self.stage = stage
|
||||||
self.configs = configs
|
self.configs = configs
|
||||||
self.experiments = experiments
|
self.experiments = experiments
|
||||||
self.endpoint = 'http://configurator/'
|
self.endpoint = 'https://platform.sprinthub.ru/'
|
||||||
self.fetch_url = urllib.parse.urljoin(self.endpoint, '/api/v1/fetch')
|
self.configs_url = urllib.parse.urljoin(self.endpoint, 'configs/get')
|
||||||
|
self.experiments_url = urllib.parse.urljoin(self.endpoint, 'experiments/get')
|
||||||
|
self.staff_url = urllib.parse.urljoin(self.endpoint, 'is_staff')
|
||||||
|
self.fetch_url = urllib.parse.urljoin(self.endpoint, 'fetch')
|
||||||
self.config_storage = {}
|
self.config_storage = {}
|
||||||
self.experiment_storage = {}
|
self.experiment_storage = {}
|
||||||
self.staff_storage = {}
|
self.staff_storage = {}
|
||||||
@@ -41,6 +44,7 @@ class PlatformClient:
|
|||||||
try:
|
try:
|
||||||
response = get(
|
response = get(
|
||||||
url,
|
url,
|
||||||
|
headers={'X-Security-Token': self.platform_security_token},
|
||||||
params=params
|
params=params
|
||||||
)
|
)
|
||||||
if response.status_code == 200:
|
if response.status_code == 200:
|
||||||
|
|||||||
@@ -17,22 +17,22 @@ class QueuesException(Exception):
|
|||||||
class TasksHandlerMixin:
|
class TasksHandlerMixin:
|
||||||
def poll(self):
|
def poll(self):
|
||||||
while True:
|
while True:
|
||||||
response = requests.get(f'{QUEUES_URL}/api/v1/take', headers={'queue': self.queue_name}).json()
|
response = requests.get(f'{QUEUES_URL}/api/v1/take', headers={'queue': self.queue_name})
|
||||||
task = response.get('task')
|
if response.status_code == 404:
|
||||||
if not task:
|
|
||||||
time.sleep(0.2)
|
time.sleep(0.2)
|
||||||
continue
|
continue
|
||||||
|
data = response.json()
|
||||||
try:
|
try:
|
||||||
self.process(task['payload'])
|
self.process(data['payload'])
|
||||||
except Exception as exc:
|
except Exception as exc:
|
||||||
print(f'Error processing message id={task["id"]}, payload={task["payload"]}, exc={exc}')
|
print(f'Error processing message id={data["id"]}, payload={data["payload"]}, exc={exc}')
|
||||||
continue
|
continue
|
||||||
try:
|
try:
|
||||||
resp = requests.post(f'{QUEUES_URL}/api/v1/finish', json={'id': task['id']})
|
resp = requests.post(f'{QUEUES_URL}/api/v1/finish', json={'id': data['id']})
|
||||||
if resp.status_code != 202:
|
if resp.status_code != 202:
|
||||||
raise QueuesException
|
raise QueuesException
|
||||||
except:
|
except:
|
||||||
print(f'Failed to finish task id={task["id"]}')
|
print(f'Failed to finish task id={data["id"]}')
|
||||||
|
|
||||||
@property
|
@property
|
||||||
def queue_name(self):
|
def queue_name(self):
|
||||||
|
|||||||
Reference in New Issue
Block a user