flask application with background threads

Question

I am creating a flask application, for one request I need to run some long running job which is not required to wait on the UI. I will create a thread and send a message to UI. The thread will calculate and update the database. But, UI will see a message upon submit. Below is my implementation, but it is running the thread and then sending the output to UI which is not I prefer. How can I run this thread in the background?

@app.route('/someJob')
def index():
    t1 = threading.Thread(target=long_running_job)
    t1.start()
    return 'Scheduled a job'

def long_running_job
    #some long running processing here

How can I make thread t1 to run the background and immediately send message in return?

Answer 1

Try this example, tested on Python 3.4.3 / Flask 0.11.1

from flask import Flask
from time import sleep
from concurrent.futures import ThreadPoolExecutor

# DOCS https://docs.python.org/3/library/concurrent.futures.html#concurrent.futures.ThreadPoolExecutor
executor = ThreadPoolExecutor(2)

app = Flask(__name__)


@app.route('/jobs')
def run_jobs():
    executor.submit(some_long_task1)
    executor.submit(some_long_task2, 'hello', 123)
    return 'Two jobs were launched in background!'


def some_long_task1():
    print("Task #1 started!")
    sleep(10)
    print("Task #1 is done!")


def some_long_task2(arg1, arg2):
    print("Task #2 started with args: %s %s!" % (arg1, arg2))
    sleep(5)
    print("Task #2 is done!")


if __name__ == '__main__':
    app.run()

Answer 2

Check out Flask-Executor which uses concurrent.futures in the background and makes your life very easy.

from flask_executor import Executor

executor = Executor(app)

@app.route('/someJob')
def index():
    executor.submit(long_running_job)
    return 'Scheduled a job'

def long_running_job
    #some long running processing here

This not only runs jobs in the background but gives them access to the app context. It also provides a way to store jobs so users can check back in to get statuses.

Answer 3

The best thing to do for stuff like this is use a message broker. There is some excellent software in the python world meant for doing just this:

• Celery (http://www.celeryproject.org/), and
• RQ (http://python-rq.org/).

Both are excellent choices.

It's almost never a good idea to spawn a thread the way you're doing it, as this can cause issues processing incoming requests, among other things.

If you take a look at the celery or RQ getting started guides, they'll walk you through doing this the proper way!

Answer 4

If you'd like to execute the long-running operation within the flask application context, then it's a bit easier to (as opposed to using ThreadPoolExecutor, taking care of exceptions):

1. Define a command line for your application (cli.py) - because all web applications should have an admin cli anyway.
2. subprocess.Popen (no wait) the command line in a web request.

For example:

# cli.py

import click
import yourpackage.app
import yourpackage.domain

app = yourpackage.app.create_app()

@click.group()
def cli():
    pass

@click.command()
@click.argument('foo_id')
def do_something(foo_id):
    with app.app_context():
        yourpackage.domain.do_something(foo_id)

if __name__ == '__main__':
    cli.add_command(do_something)
    cli()

Then,

# admin.py (flask view / controller)

bp = Blueprint('admin', __name__, url_prefix='/admin')

@bp.route('/do-something/<int:foo_id>', methods=["POST"])
@roles_required('admin')
def do_something(foo_id):
    yourpackage.domain.process_wrapper_do_something(foo_id)
    flash("Something has started.", "info")
    return redirect(url_for("..."))

And:

# domain.py

import subprocess

def process_wrapper_do_something(foo_id):
    command = ["python3", "-m", "yourpackage.cli", "do_something", str(foo_id)]
    subprocess.Popen(command)

def do_something(foo_id):
    print("I am doing something.")
    print("This takes some time.")

Answer 5

Agree with the marked answer from @rdegges. Sorry my account doesn't have enough credit to add comment under the answer, but I want to make it clear on "Why to use a message broker, instead of spawning a thread (or process)".

The other answers about ThreadPoolExecutor and flask_executor are creating a new thread (or process, as flask_executor is capable of) to execute the "long_running_job". These new threads/processes will have the same context as the main website:

For threads: The new thread will be able to access the context of the website application, change things, or break it, if this thread raises an exception; For processes: The new process will have a copy of the context of the website application. If the website somehow use a lot of memory in initialization, the new process will have a copy of it too, even if the process is not going to utilize this part of the memory.

On another hand, if you are using a message broker, and another application to retrieve the job message to work on it, the new application will have nothing to do with the website application, also it doesn't copy the memory from the web app.

In the future, when your application is big enough, you can place your application into another server (or servers), It is easy to scale out.

Answer 6

If you want use Celery with flask. Firstly check your operation and when operation completed redirect your celery task. You can check this link if you dont now celery: FLASK: How to establish connection with mysql server?