Where can I find celery magazines in a flask?

I am working on a simple web application that runs a python script, which can take about 5-10 minutes. Instead of making the user wait on the site, I run the task in the background using Celery and send the user an email notification when the script completes.

My question is simple: where are the celery magazines? Currently, if an assignment fails because it encounters a fatal error, I have no way to find out.

Celery registration docs show how to set up the logs, but it is not clear to me where I can access the logs (I want to save as a file) after completing the task.

app.py

#----------------------------------------------------------------------------#
# Imports
#----------------------------------------------------------------------------#

from flask import Flask, render_template, request, flash, send_file, redirect, url_for    
from werkzeug import secure_filename
import logging
from logging import Formatter, FileHandler
from forms import *

import my_module
from tasks import make_celery
from celery.utils.log import get_task_logger


#----------------------------------------------------------------------------#
# App Config.
#----------------------------------------------------------------------------#

app = Flask(__name__)
app.config.from_object('config')

celery = make_celery(app)
logger = get_task_logger(__name__)

format = "%(asctime)s - [%(levelname)s] %(message)s"
logging.basicConfig(filename='app-errors.log', 
                    filemode='a',
                    format=format,
                    level=logging.DEBUG)
console = logging.StreamHandler()
console.setLevel(logging.DEBUG)
logging.getLogger(__name__).addHandler(console)

#----------------------------------------------------------------------------#
# Controllers.
#----------------------------------------------------------------------------#

@celery.task
def background_task(formdict):
    logger.debug('starting work...')
    result = my_module.main(formdict)
    return result

@app.route('/', methods=('GET', 'POST'))
def index():
    form = MyForm()
    if form.validate_on_submit():
        task = background_task.delay(form.data)

        return redirect(url_for('jobcomplete'))
    else:
        print form.errors
        filename = None
    return render_template('pages/home.html', form=form)
+4

Source: https://habr.com/ru/post/1611362/


All Articles