Выводим результаты работы периодических задач Celery на фронт.
Для этих целей можно установить flower. Но у него есть ряд недостатков:
- Доступ на сервере настраивается витиевато: либо дополнительно писать конфиги в Nginx либо подключаться через google api. На мой взгляд слишком далеко от кода все это, потом не вспомнить где это лежит.
- Нет мониторинга записанных в БД задач. Мониторинг только в лайв режиме.
Все недостатки легко обойти быстро соорудив страницу в админке для просмотра результатов работы периодических задач.
Весь код:
from flask import Blueprint, json
from flask_login import login_required
from app import db
celery_bp = Blueprint('celery_bp', __name__)
def serialize_query(msg):
return {
"task_id": msg.task_id,
"status": msg.status,
"result": msg.result.tobytes().decode('utf-8', errors='ignore').replace(
'\x05C\x00\x00\x00\x00\x00\x00\x00?', ''),
"date_done": msg.date_done.strftime('%H:%M | %d %B %Y'),
}
@login_required
@celery_bp.route('/get-logs/', methods=['GET'])
def get_logs():
query = [serialize_query(x) for x in db.engine.execute('select * from celery_taskmeta order by date_done')]
return json.dumps(query)
from flask_login import login_required
from app import db
celery_bp = Blueprint('celery_bp', __name__)
def serialize_query(msg):
return {
"task_id": msg.task_id,
"status": msg.status,
"result": msg.result.tobytes().decode('utf-8', errors='ignore').replace(
'\x05C\x00\x00\x00\x00\x00\x00\x00?', ''),
"date_done": msg.date_done.strftime('%H:%M | %d %B %Y'),
}
@login_required
@celery_bp.route('/get-logs/', methods=['GET'])
def get_logs():
query = [serialize_query(x) for x in db.engine.execute('select * from celery_taskmeta order by date_done')]
return json.dumps(query)
Это получение данных с бэкенда. Здесь используются Flask, Blueprint, SQLAlchemy. Остается только на фронте отправить запрос на указанный урл, распарсить json и вывести на страницу.
Ошибки
TypeError: Object of type memoryview is not JSON serializable
Из особенностей Celery. Он сохраняет результат в БД в формате memoryview. Мы этот формат переводим в байты командой:
tobytes()
Затем декодируем игнорируя ошибки:
.decode('utf-8', errors='ignore')
Почему-то у меня celery сохраняет memoryview с неправильными символами на что собственно и ругается интерпретатор. Дальше эти символы удаляем командой:
.replace('\x05C\x00\x00\x00\x00\x00\x00\x00?', '')
