self/botalka
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

master #60

Merged
emmatveev merged 5 commits from master into prod 2025-06-15 03:51:51 +03:00
Conversation 0 Commits 5 Files Changed 2 +8 -29
2 changed files with 8 additions and 29 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
daemons/mailbox.py
View File

@ -23,6 +23,7 @@ class Daemon(base.Daemon, queues.TasksHandlerMixin):
return 'botalka_mailbox' return 'botalka_mailbox'
def process(self, payload: dict): def process(self, payload: dict):
print(f"Get message {payload}")
message = Message.model_validate(payload) message = Message.model_validate(payload)
bot = platform.platform_client.get_config('bots')[message.project][message.name] bot = platform.platform_client.get_config('bots')[message.project][message.name]
if not bot['mailbox_enabled']: if not bot['mailbox_enabled']:

36
utils/queues.py
View File

@ -1,6 +1,5 @@
import datetime import datetime
import os import os
from threading import Thread
import zoneinfo import zoneinfo
import requests import requests
import time import time
@ -18,27 +17,6 @@ class QueuesException(Exception):
class TasksHandlerMixin: class TasksHandlerMixin:
def _send_metric(self, start, success, end):
try:
metric = requests.post('http://monitoring:1237/api/v1/metrics/task', json={
'timestamp': start.strftime("%Y-%m-%dT%H:%M:%S") + "Z",
'service': 'botalka',
'environment': stage,
'queue': self.queue_name,
'success': success,
'execution_time_ms': (end - start).microseconds // 1000,
})
if metric.status_code == 202:
print('metric ok')
else:
print(f'metric failed: {metric.status_code}')
except Exception as e:
print(f'metric failed: {e}')
def send_metric(self, start, success, end):
# Thread(target=self._send_metric, args=(start, success, end)).start()
# self._send_metric(start, success, end)
...
def poll(self): def poll(self):
while True: while True:
@ -60,13 +38,13 @@ class TasksHandlerMixin:
print(f'Error processing message id={task["id"]}, payload={task["payload"]}, exc={exc}') print(f'Error processing message id={task["id"]}, payload={task["payload"]}, exc={exc}')
success = False success = False
end = datetime.datetime.now().astimezone(zoneinfo.ZoneInfo("Europe/Moscow")) end = datetime.datetime.now().astimezone(zoneinfo.ZoneInfo("Europe/Moscow"))
try: if success:
resp = requests.post(f'{QUEUES_URL}/api/v1/finish', json={'id': task['id']}) try:
if resp.status_code != 202: resp = requests.post(f'{QUEUES_URL}/api/v1/finish', json={'id': task['id']})
raise QueuesException if resp.status_code != 202:
except: raise QueuesException
print(f'Failed to finish task id={task["id"]}') except:
self.send_metric(start, success, end) print(f'Failed to finish task id={task["id"]}')
@property @property
def queue_name(self): def queue_name(self):