How to stream stdout from a Celery Task and display it on a web app?

user_6091189 0
Viewed 520

I have setup an asynchronous python task in a Flask Restful API. The asynchronous task can run for up to several hours. While that long task runs in Celery, it periodically outputs text on the celery terminal. My question: how can I stream the stdout, and monitor the stdout outputs on a web-app.

The basic structure of how I have set up my task in Python is like this:

@celery.task(name='my_long_task')
def my_long_task(arg_1):
    proc = subprocess.Popen(f'python xyz.py', shell=True)

def start_background_task():
    arg_1 = "my name"
    my_long_task.delay(arg_1)

I have explored Flower and programmatically logging Celery tasks, but so far I have not been able to access the "live stream" of the stdout of a celery task. Any suggestions on how to achieve this will be very helpful.

flask-restful celery python
2 Answers

Don't use subprocess, import your python file as a module and call functions from it, such as:

import xyz

@celery.task(name='my_long_task')
def my_long_task(arg_1):
    xyz.main(arg_1)

Then when you start the task, capture the task id:

def start_background_task():
    arg_1 = "my name"
    return my_long_task.delay(arg_1).id

With the id captured, you can do queries on its status and results:

from celery.result import AsyncResult

task = AsyncResult("your-task-id")
task.result  # OR
task.state

See Celery docs for more info: https://docs.celeryproject.org/en/latest/faq.html#how-do-i-get-the-result-of-a-task-if-i-have-the-id-that-points-there

user_19363210

I haven't found a reliable solution for "streaming" output either. What I'm doing is reading the output and sending it to RabbitMQ:

import kombu, amqp
from ... import app
from subprocess import Popen, PIPE

@app.task
def long_running_task():
    connection = kombu.Connection(...)
    channel = connection.channel()
    channel.queue_declare("QUEUE_NAME")

    p = Popen(
        "echo 1; sleep 20; echo 2; sleep 40; echo 3; sleep 60; echo done",
        shell=True,
        stdout=PIPE,
    )
    for line in iter(p.stdout.readline, ""):
        if not line:
            break
        channel.basic_publish(amqp.Message(line), routing_key=self.queue)

and then consuming that output from RabbitMQ elsewhere:

from ...tasks import long_running_task

def printer(msg):
    print(msg.body)


def some_function():
    job = long_running_task.apply_async(queue="<QUEUE>")

    connection = kombu.Connection(...)
    channel = connection.channel()
    # just in case the job hasn't started yet
    channel.queue_declare("QUEUE_NAME")

    channel.basic_consume(queue="QUEUE_NAME", callback=printer)

    while not job.ready():
        try:
            self.connection.drain_events(timeout=2)
        except socket.timeout:
            pass

    print(job.get())
user_66551500
Related