trying to run the first example but got the following error.
the code is cloned from the repository.
Traceback (most recent call last): File "/Users/astronaut/PycharmProjects/using_celery_with_flask/venv/lib/python3.6/site-packages/flask/app.py", line 1997, in __call__ return self.wsgi_app(environ, start_response) File "/Users/astronaut/PycharmProjects/using_celery_with_flask/venv/lib/python3.6/site-packages/flask/app.py", line 1985, in wsgi_app response = self.handle_exception(e) File "/Users/astronaut/PycharmProjects/using_celery_with_flask/venv/lib/python3.6/site-packages/flask/app.py", line 1540, in handle_exception reraise(exc_type, exc_value, tb) File "/Users/astronaut/PycharmProjects/using_celery_with_flask/venv/lib/python3.6/site-packages/flask/_compat.py", line 33, in reraise raise value File "/Users/astronaut/PycharmProjects/using_celery_with_flask/venv/lib/python3.6/site-packages/flask/app.py", line 1982, in wsgi_app response = self.full_dispatch_request() File "/Users/astronaut/PycharmProjects/using_celery_with_flask/venv/lib/python3.6/site-packages/flask/app.py", line 1614, in full_dispatch_request rv = self.handle_user_exception(e) File "/Users/astronaut/PycharmProjects/using_celery_with_flask/venv/lib/python3.6/site-packages/flask/app.py", line 1517, in handle_user_exception reraise(exc_type, exc_value, tb) File "/Users/astronaut/PycharmProjects/using_celery_with_flask/venv/lib/python3.6/site-packages/flask/_compat.py", line 33, in reraise raise value File "/Users/astronaut/PycharmProjects/using_celery_with_flask/venv/lib/python3.6/site-packages/flask/app.py", line 1612, in full_dispatch_request rv = self.dispatch_request() File "/Users/astronaut/PycharmProjects/using_celery_with_flask/venv/lib/python3.6/site-packages/flask/app.py", line 1598, in dispatch_request return self.view_functions[rule.endpoint](**req.view_args) File "/Users/astronaut/PycharmProjects/using_celery_with_flask/flask-celery-example/app.py", line 75, in index send_async_email.delay(msg) File "/Users/astronaut/PycharmProjects/using_celery_with_flask/venv/lib/python3.6/site-packages/celery/app/task.py", line 413, in delay return self.apply_async(args, kwargs) File "/Users/astronaut/PycharmProjects/using_celery_with_flask/venv/lib/python3.6/site-packages/celery/app/task.py", line 536, in apply_async **options File "/Users/astronaut/PycharmProjects/using_celery_with_flask/venv/lib/python3.6/site-packages/celery/app/base.py", line 737, in send_task amqp.send_task_message(P, name, message, **options) File "/Users/astronaut/PycharmProjects/using_celery_with_flask/venv/lib/python3.6/site-packages/celery/app/amqp.py", line 554, in send_task_message **properties File "/Users/astronaut/PycharmProjects/using_celery_with_flask/venv/lib/python3.6/site-packages/kombu/messaging.py", line 169, in publish compression, headers) File "/Users/astronaut/PycharmProjects/using_celery_with_flask/venv/lib/python3.6/site-packages/kombu/messaging.py", line 252, in _prepare body) = dumps(body, serializer=serializer) File "/Users/astronaut/PycharmProjects/using_celery_with_flask/venv/lib/python3.6/site-packages/kombu/serialization.py", line 221, in dumps payload = encoder(data) File "/usr/local/Cellar/python3/3.6.4/Frameworks/Python.framework/Versions/3.6/lib/python3.6/contextlib.py", line 99, in __exit__ self.gen.throw(type, value, traceback) File "/Users/astronaut/PycharmProjects/using_celery_with_flask/venv/lib/python3.6/site-packages/kombu/serialization.py", line 54, in _reraise_errors reraise(wrapper, wrapper(exc), sys.exc_info()[2]) File "/Users/astronaut/PycharmProjects/using_celery_with_flask/venv/lib/python3.6/site-packages/vine/five.py", line 178, in reraise raise value.with_traceback(tb) File "/Users/astronaut/PycharmProjects/using_celery_with_flask/venv/lib/python3.6/site-packages/kombu/serialization.py", line 50, in _reraise_errors yield File "/Users/astronaut/PycharmProjects/using_celery_with_flask/venv/lib/python3.6/site-packages/kombu/serialization.py", line 221, in dumps payload = encoder(data) File "/Users/astronaut/PycharmProjects/using_celery_with_flask/venv/lib/python3.6/site-packages/kombu/utils/json.py", line 72, in dumps **dict(default_kwargs, **kwargs)) File "/usr/local/Cellar/python3/3.6.4/Frameworks/Python.framework/Versions/3.6/lib/python3.6/json/__init__.py", line 238, in dumps **kw).encode(obj) File "/usr/local/Cellar/python3/3.6.4/Frameworks/Python.framework/Versions/3.6/lib/python3.6/json/encoder.py", line 199, in encode chunks = self.iterencode(o, _one_shot=True) File "/usr/local/Cellar/python3/3.6.4/Frameworks/Python.framework/Versions/3.6/lib/python3.6/json/encoder.py", line 257, in iterencode return _iterencode(o, 0) File "/Users/astronaut/PycharmProjects/using_celery_with_flask/venv/lib/python3.6/site-packages/kombu/utils/json.py", line 62, in default return super(JSONEncoder, self).default(o) File "/usr/local/Cellar/python3/3.6.4/Frameworks/Python.framework/Versions/3.6/lib/python3.6/json/encoder.py", line 180, in default o.__class__.__name__) kombu.exceptions.EncodeError: Object of type 'Message' is not JSON serializable

Is it possible to process input json using celery from multiple client ? If so, kindly advice how to process the input files in asynchronous mode.

how to write a unit test for this .

@celery.task(bind=True)
def long_task(self):
    """Background task that runs a long function with progress reports."""
    verb = ['Starting up', 'Booting', 'Repairing', 'Loading', 'Checking']
    adjective = ['master', 'radiant', 'silent', 'harmonic', 'fast']
    noun = ['solar array', 'particle reshaper', 'cosmic ray', 'orbiter', 'bit']
    message = ''
    total = random.randint(10, 50)
    for i in range(total):
        if not message or random.random() < 0.25:
            message = '{0} {1} {2}...'.format(random.choice(verb),
                                              random.choice(adjective),
                                              random.choice(noun))
        self.update_state(state='PROGRESS',
                          meta={'current': i, 'total': total,
                                'status': message})
        time.sleep(1)
    return {'current': 100, 'total': 100, 'status': 'Task completed!',
            'result': 42}

When i write a unit test, it fails at unable to open the database file. fails at update_state line.

def test_long_task(self, mocker):
        s = mocker.patch('service.random.randint')
        s.return_value = 5
        res = service.long_task()
        assert res == {'current': 100, 'total': 100, 'status': 'Task completed!',
            'result': 42}

PS: i've used the same code and must to write a unit test for sonarqube coverage and for my CI/CD pipeline. There it fails with error as unable to connect with database. i've used sqlite for celery backend instead of redis.

When I use your code fot testing , it raised the error, because the celery receive the json data, not the Message object.

Traceback (most recent call last):
File "/home/yzhao_sherry/work/venv/lib/python2.7/site-packages/flask/app.py", line 1997, in call
return self.wsgi_app(environ, start_response)
File "/home/yzhao_sherry/work/venv/lib/python2.7/site-packages/flask/app.py", line 1985, in wsgi_app
response = self.handle_exception(e)
File "/home/yzhao_sherry/work/venv/lib/python2.7/site-packages/flask/app.py", line 1540, in handle_exception
reraise(exc_type, exc_value, tb)
File "/home/yzhao_sherry/work/venv/lib/python2.7/site-packages/flask/app.py", line 1982, in wsgi_app
response = self.full_dispatch_request()
File "/home/yzhao_sherry/work/venv/lib/python2.7/site-packages/flask/app.py", line 1614, in full_dispatch_request
rv = self.handle_user_exception(e)
File "/home/yzhao_sherry/work/venv/lib/python2.7/site-packages/flask/app.py", line 1517, in handle_user_exception
reraise(exc_type, exc_value, tb)
File "/home/yzhao_sherry/work/venv/lib/python2.7/site-packages/flask/app.py", line 1612, in full_dispatch_request
rv = self.dispatch_request()
File "/home/yzhao_sherry/work/venv/lib/python2.7/site-packages/flask/app.py", line 1598, in dispatch_request
return self.view_functionsrule.endpoint
File "/home/yzhao_sherry/work/venv/pypro/flask-celery-example/app.py", line 285, in index
send_async_email.apply_async(args=[msg], countdown=60)
File "/home/yzhao_sherry/work/venv/lib/python2.7/site-packages/celery/app/task.py", line 536, in apply_async
**options
File "/home/yzhao_sherry/work/venv/lib/python2.7/site-packages/celery/app/base.py", line 737, in send_task
amqp.send_task_message(P, name, message, **options)
File "/home/yzhao_sherry/work/venv/lib/python2.7/site-packages/celery/app/amqp.py", line 554, in send_task_message
**properties
File "/home/yzhao_sherry/work/venv/lib/python2.7/site-packages/kombu/messaging.py", line 169, in publish
compression, headers)
File "/home/yzhao_sherry/work/venv/lib/python2.7/site-packages/kombu/messaging.py", line 252, in _prepare
body) = dumps(body, serializer=serializer)
File "/home/yzhao_sherry/work/venv/lib/python2.7/site-packages/kombu/serialization.py", line 221, in dumps
payload = encoder(data)
File "/usr/lib64/python2.7/contextlib.py", line 35, in exit
self.gen.throw(type, value, traceback)
File "/home/yzhao_sherry/work/venv/lib/python2.7/site-packages/kombu/serialization.py", line 54, in _reraise_errors
reraise(wrapper, wrapper(exc), sys.exc_info()[2])
File "/home/yzhao_sherry/work/venv/lib/python2.7/site-packages/kombu/serialization.py", line 50, in _reraise_errors
yield
File "/home/yzhao_sherry/work/venv/lib/python2.7/site-packages/kombu/serialization.py", line 221, in dumps
payload = encoder(data)
File "/home/yzhao_sherry/work/venv/lib/python2.7/site-packages/kombu/utils/json.py", line 72, in dumps
**dict(default_kwargs, **kwargs))
File "/usr/lib64/python2.7/json/init.py", line 251, in dumps
sort_keys=sort_keys, **kw).encode(obj)
File "/usr/lib64/python2.7/json/encoder.py", line 207, in encode
chunks = self.iterencode(o, _one_shot=True)
File "/usr/lib64/python2.7/json/encoder.py", line 270, in iterencode
return _iterencode(o, 0)
File "/home/yzhao_sherry/work/venv/lib/python2.7/site-packages/kombu/utils/json.py", line 62, in default
return super(JSONEncoder, self).default(o)
File "/usr/lib64/python2.7/json/encoder.py", line 184, in default
raise TypeError(repr(o) + " is not JSON serializable")
EncodeError: <flask_mail.Message object at 0x7fe0b576f510> is not JSON serializable

I installed celery-server 3.0.0 to flask and then I started the server, but when I used the server to run a python code backgroud I find this error.

~/Bureau$ sudo python exme.py Traceback (most recent call last): File "exme.py", line 2, in from celery import Celery ImportError: No module named celery

from flask import Flask
from celery import Celery
app = Flask(name)
app.config.update(
CELERY_BROKER_URL='redis://localhost:6379',
CELERY_RESULT_BACKEND='redis://localhost:6379'
)
celery = Celery(app.name, broker=app.config['CELERY_BROKER_URL'])
celery.conf.update(app.config)
@celery.task()
def add_together(a, b):
return a + b
@APP.route('/test',methods=['POST'])
def test():
try:
result=add_together.delay(5,2)
return result
except Exception as e:

return e

I need help please.

This is not a issue about your code, just a question.

As most mail provider asks user to login before using smtp, so I faced the error: "Error: need EHLO and AUTH first !"
Do you have a good idea about how to make flask mail login?

I might have configured something badly but when pressing the longtask button, it fires the GET /status/taskid every 2s and doesn't stop, and always returns {"current": 0, "state": "PENDING", "status": "Pending...", "total": 1}

出现了上述问题，出错的地方大概在如下图所示的地方

I need to sart reddis from the command line or on startup.
Or will the backend be automatically launched?

Traceback (most recent call last):
  File "./app.py", line 10, in <module>
    app = Flask(__name__)
  File "/Users/megamorf/.local/share/virtualenvs/flask-celery-example--m1TFcnE/lib/python3.8/site-packages/flask/app.py", line 559, in __init__
    self.add_url_rule(
  File "/Users/megamorf/.local/share/virtualenvs/flask-celery-example--m1TFcnE/lib/python3.8/site-packages/flask/app.py", line 67, in wrapper_func
    return f(self, *args, **kwargs)
  File "/Users/megamorf/.local/share/virtualenvs/flask-celery-example--m1TFcnE/lib/python3.8/site-packages/flask/app.py", line 1217, in add_url_rule
    self.url_map.add(rule)
  File "/Users/megamorf/.local/share/virtualenvs/flask-celery-example--m1TFcnE/lib/python3.8/site-packages/werkzeug/routing.py", line 1388, in add
    rule.bind(self)
  File "/Users/megamorf/.local/share/virtualenvs/flask-celery-example--m1TFcnE/lib/python3.8/site-packages/werkzeug/routing.py", line 730, in bind
    self.compile()
  File "/Users/megamorf/.local/share/virtualenvs/flask-celery-example--m1TFcnE/lib/python3.8/site-packages/werkzeug/routing.py", line 794, in compile
    self._build = self._compile_builder(False).__get__(self, None)
  File "/Users/megamorf/.local/share/virtualenvs/flask-celery-example--m1TFcnE/lib/python3.8/site-packages/werkzeug/routing.py", line 951, in _compile_builder
    code = compile(module, "<werkzeug routing>", "exec")
TypeError: required field "type_ignores" missing from Module
(flask-celery-example)

See:
pallets/werkzeug#1551

I'm sorta having a problem adopting the approach introduced in the blog post.
I'm trying this on a Windows 10 platform.
for some reason, and on other projects which didn't have flask I had to open a new terminal for celery and command this:
celery -A tasks worker -l info -P eventlet
questions troubled me so far and I really appreciate getting an answer:

1. (the dummiest one probably) when I'm about to run my project should I do the same. I mean opening a terminal and commanding celery?
2. if no, how can I set the "-P eventlet" part in my app?
3. if yes, what would be the command exactly?
4. I'm instantiating Celery and defining a make_celery function in a file named extensions.py and calling it in my create_app. Am I correct?

I have these environment variables in Window 7, but somehow during runtime they are still null?

When i use celery and apply_async(), i got such error:

EncodeError: Can't pickle class 'werkzeug.datastructures.ImmutableMultiDict': it's not the same object as werkzeug.datastructures.ImmutableMultiDict

@celery.task
def send_async_email(app, msg):
with app.app_context():
mail.send(msg)

def send_email(to, subject, template, *_kwargs):
app = current_app._get_current_object()
with app.app_context():
msg = Message(app.config['FLASKY_MAIL_SUBJECT_PREFIX'] + ' ' + subject,
sender=app.config['FLASKY_MAIL_SENDER'], recipients=[to])
msg.body = render_template(template + '.txt', *_kwargs)
msg.html = render_template(template + '.html', **kwargs)
send_async_email.apply_async(args=[current_app, msg])

but if i don't send current_app to send_async_email, then i got "RuntimeError: working outside of application context" in celery terminal outputs.

Can you help me ?

python 2.7.11 does not work with the version of kombu in requriements.txt.

Hi Miguel,

I was able to get this celery example working - thanks for a great tutorial. However, I have trouble to integrate Celery with Flasky from your book. I am just not sure where to put various information into Flasky to make it work. For example, I have put the following line in config.py:

# Celery configuration
CELERY_BROKER_URL = 'redis://localhost:6379/0'
CELERY_RESULT_BACKEND = 'redis://localhost:6379/0'

And in init.py:

from celery import Celery

but I don't know where to add:

# Initialize Celery
celery = Celery(app.name, broker=app.config['CELERY_BROKER_URL'])
celery.conf.update(app.config)

will the following work?

celery = Celery()

And then what to do? Please help and thanks.

Quick question: in the long_task celery function, i notice that you have used time.sleep(1)

Not sure I fully understand the implications of the sleep statement here ? I tried removing it and cant seem to see any differences in the output atleast cosmetically. Would appreciate any thoughts ?

for i in range(total):
    if not message or random.random() < 0.25:
        message = '{0} {1} {2}...'.format(random.choice(verb),
                                          random.choice(adjective),
                                          random.choice(noun))
    self.update_state(state='PROGRESS',
                      meta={'current': i, 'total': total,
                            'status': message})
    time.sleep(1)

Thanks

I ran your code locally and it works well.
I build similar system of long running task and capturing custom states. But in my case it fails with this error

https://stackoverflow.com/questions/31852219/why-asyncresult-returns-nonetype-after-task-is-done

And it is resolved after i initialized celery object with backend as below,

celery = Celery(app.name, broker=app.config['CELERY_BROKER_URL'], backend=app.config['CELERY_RESULT_BACKEND'])

Please let me know how it is working in your case.??

How can I start the worker and flask app only with one script?
Thanks!

Get this error while running the example.
I installed the necessary requirements from the requirements.txt.
Got this error when I clicked on the Start Long Calculations on the Webpage. Any pointers would be appreciated.

D:\flask-celery-example-master>python app.py

• Restarting with stat
• Debugger is active!
• Debugger PIN: 316-263-277
• Running on http://127.0.0.1:5000/ (Press CTRL+C to quit)
  127.0.0.1 - - [06/Mar/2019 17:51:06] "GET / HTTP/1.1" 200 -
  127.0.0.1 - - [06/Mar/2019 17:51:06] "GET /favicon.ico HTTP/1.1" 404 -
  127.0.0.1 - - [06/Mar/2019 17:51:34] "POST / HTTP/1.1" 500 -
  Traceback (most recent call last):
  File "C:\Users\AppData\Local\Programs\Python\Python36\lib\site-packages\flask\app.py", line 1836, in call
  return self.wsgi_app(environ, start_response)
  File "C:\Users\AppData\Local\Programs\Python\Python36\lib\site-packages\flask\app.py", line 1820, in wsgi_app
  response = self.make_response(self.handle_exception(e))
  File "C:\Users\AppData\Local\Programs\Python\Python36\lib\site-packages\flask\app.py", line 1403, in handle_exception
  reraise(exc_type, exc_value, tb)
  File "C:\Users\AppData\Local\Programs\Python\Python36\lib\site-packages\flask_compat.py", line 33, in reraise
  raise value
  File "C:\Users\AppData\Local\Programs\Python\Python36\lib\site-packages\flask\app.py", line 1817, in wsgi_app
  response = self.full_dispatch_request()
  File "C:\Users\AppData\Local\Programs\Python\Python36\lib\site-packages\flask\app.py", line 1477, in full_dispatch_request
  rv = self.handle_user_exception(e)
  File "C:\Users\AppData\Local\Programs\Python\Python36\lib\site-packages\flask\app.py", line 1381, in handle_user_exception
  reraise(exc_type, exc_value, tb)
  File "C:\Users\AppData\Local\Programs\Python\Python36\lib\site-packages\flask_compat.py", line 33, in reraise
  raise value
  File "C:\Users\AppData\Local\Programs\Python\Python36\lib\site-packages\flask\app.py", line 1475, in full_dispatch_request
  rv = self.dispatch_request()
  File "C:\Users\AppData\Local\Programs\Python\Python36\lib\site-packages\flask\app.py", line 1461, in dispatch_request
  return self.view_functionsrule.endpoint
  File "D:\flask-celery-example-master\app.py", line 75, in index
  send_async_email.delay(msg)
  File "C:\Users\AppData\Local\Programs\Python\Python36\lib\site-packages\celery\app\task.py", line 453, in delay
  return self.apply_async(args, kwargs)
  File "C:\Users\AppData\Local\Programs\Python\Python36\lib\site-packages\celery\app\task.py", line 555, in apply_async
  **dict(self._get_exec_options(), **options)
  File "C:\Users\AppData\Local\Programs\Python\Python36\lib\site-packages\celery\app\base.py", line 353, in send_task
  reply_to=reply_to or self.oid, **options
  File "C:\Users\AppData\Local\Programs\Python\Python36\lib\site-packages\celery\app\amqp.py", line 305, in publish_task
  **kwargs
  File "C:\Users\AppData\Local\Programs\Python\Python36\lib\site-packages\kombu\messaging.py", line 172, in publish
  routing_key, mandatory, immediate, exchange, declare)
  File "C:\Users\AppData\Local\Programs\Python\Python36\lib\site-packages\kombu\connection.py", line 457, in ensured
  interval_max)
  File "C:\Users\AppData\Local\Programs\Python\Python36\lib\site-packages\kombu\connection.py", line 369, in ensure_connection
  interval_start, interval_step, interval_max, callback)
  File "C:\Users\AppData\Local\Programs\Python\Python36\lib\site-packages\kombu\utils_init.py", line 246, in retry_over_time
  return fun(*args, **kwargs)
  File "C:\Users\AppData\Local\Programs\Python\Python36\lib\site-packages\kombu\connection.py", line 237, in connect
  return self.connection
  File "C:\Users\AppData\Local\Programs\Python\Python36\lib\site-packages\kombu\connection.py", line 741, in connection
  self._connection = self._establish_connection()
  File "C:\Users\AppData\Local\Programs\Python\Python36\lib\site-packages\kombu\connection.py", line 696, in establish_connection
  conn = self.transport.establish_connection()
  File "C:\Users\AppData\Local\Programs\Python\Python36\lib\site-packages\kombu\transport\virtual_init.py", line 809, in establish_connection
  self.avail_channels.append(self.create_channel(self))
  File "C:\Users\AppData\Local\Programs\Python\Python36\lib\site-packages\kombu\transport\virtual_init.py", line 791, in create_channel
  channel = self.Channel(connection)
  File "C:\Users\AppData\Local\Programs\Python\Python36\lib\site-packages\kombu\transport\redis.py", line 456, in init
  self._pool.disconnect()
  File "C:\Users\AppData\Local\Programs\Python\Python36\lib\site-packages\redis\connection.py", line 898, in disconnect
  connection.disconnect()
  File "C:\Users\AppData\Local\Programs\Python\Python36\lib\site-packages\kombu\transport\redis.py", line 819, in disconnect
  channel._on_connection_disconnect(self)
  File "C:\Users\AppData\Local\Programs\Python\Python36\lib\site-packages\kombu\transport\redis.py", line 476, in _on_connection_disconnect
  raise get_redis_ConnectionError()
  redis.exceptions.ConnectionError
  127.0.0.1 - - [06/Mar/2019 17:51:45] "POST /longtask HTTP/1.1" 500 -
  127.0.0.1 - - [06/Mar/2019 17:51:45] "GET /?debugger=yes&cmd=resource&f=style.css HTTP/1.1" 200 -
  127.0.0.1 - - [06/Mar/2019 17:51:45] "GET /?debugger=yes&cmd=resource&f=jquery.js HTTP/1.1" 200 -
  127.0.0.1 - - [06/Mar/2019 17:51:45] "GET /?debugger=yes&cmd=resource&f=debugger.js HTTP/1.1" 200 -
  127.0.0.1 - - [06/Mar/2019 17:51:45] "GET /?debugger=yes&cmd=resource&f=ubuntu.ttf HTTP/1.1" 200 -
  127.0.0.1 - - [06/Mar/2019 17:51:45] "GET /?debugger=yes&cmd=resource&f=console.png HTTP/1.1" 200 -
  127.0.0.1 - - [06/Mar/2019 17:51:45] "GET /?debugger=yes&cmd=resource&f=console.png HTTP/1.1" 200 -

Old flask email extension long deprecated. New version kept complaining about compatibility with the older version of Flask specified. I also had "missing extension issues". I just pulled the list in manually and everything functions. Great demonstration as always, I sincerely appreciate your efforts.

I am using python 3.4.5 on a Centos 7 machine:

My pip freeze base on a line by line manual install of requirements:

amqp==2.2.2
anyjson==0.3.3
billiard==3.5.0.3
blinker==1.4
celery==4.1.0
click==6.7
Flask==0.12.2
Flask-Mail==0.9.1
itsdangerous==0.24
Jinja2==2.10
kombu==4.1.0
MarkupSafe==1.0
pytz==2018.4
redis==2.10.6
vine==1.1.4
Werkzeug==0.14.1

Thanks,
Justin

Issue: on Quick Setup 5 'venv/bin/python app.py', app fails to run

Cause: unsupported version of Flask in requirements.txt

Fix: upgrade Flask>=2.2.2