Can I have local callbacks on celery task objects?

Question

I've rigged up a small demo - it's intended to call a time-consuming function on Celery. I want to execute

import itertools
import time
import logging

from celery.result import AsyncResult

from myproj.tasks.time_consuming_thing import time_consuming_thing

log: logging.Logger = logging.getLogger()


def log_the_result(result):
    print("Result: %r" % result)

def main():
    for i in itertools.count(0):
        log.info("About to schedule a task: #%i", i)

        result: AsyncResult = time_consuming_thing.delay()
        result.then(callback=log_the_result)

        time.sleep(10)


if __name__ == "__main__":
    logging.basicConfig()
    logging.getLogger("").setLevel(logging.INFO)
    main()

What actually seems to happen is... nothing:

I can see the worker is returning a value, but the value never seems to make it to the consumer. The callback function is never called.

What can I do so that the callback function is called with the return value of the result?

Answer 1

In order to use then functionality you have to use either aio, threading, or gevent.

For gevent, you would use something like this (copy & pasted from the github thread above):

import gevent.monkey
gevent.monkey.patch_all()

import itertools
import time
import logging

from celery.result import AsyncResult

from myproj.tasks.time_consuming_thing import time_consuming_thing

log: logging.Logger = logging.getLogger()


def log_the_result(result):
    print("Result: %r" % result)

def main():
    for i in itertools.count(0):
        log.info("About to schedule a task: #%i", i)

        result: AsyncResult = time_consuming_thing.delay()
        result.then(callback=log_the_result)

        time.sleep(10)


if __name__ == "__main__":
    logging.basicConfig()
    logging.getLogger("").setLevel(logging.INFO)
    main()