Shulou(Shulou.com)06/01 Report--

This article mainly introduces the relevant knowledge of Flask how to achieve asynchronous task execution, the content is detailed and easy to understand, the operation is simple and fast, and has a certain reference value. I believe you will gain something after reading this Flask article on how to achieve asynchronous task execution. Let's take a look at it.

Flask is a famous lightweight synchronous web framework in Python. In some development, you may encounter tasks that need to be processed for a long time. At this time, you need to use an asynchronous method to let the long-time task run in the background. First, return the response status of this request to the front end to prevent the front-end interface from "stuttering". When the asynchronous task is processed, if you need to return the status, then return the status.

How can it be realized? The way threads are used

When you want to perform a time-consuming task, it is easiest and quickest to start a new thread to execute the task.

Through ThreadPoolExecutor to achieve

From flask import Flaskfrom time import sleepfrom concurrent.futures import ThreadPoolExecutor# DOCS https://docs.python.org/3/library/concurrent.futures.html#concurrent.futures.ThreadPoolExecutor# create thread pool executor executor = ThreadPoolExecutor (2) app = Flask (_ _ name__) @ app.route ('/ jobs') def run_jobs (): # leave the thread to execute the time-consuming task executor.submit (long_task, 'hello', 123) return' long task running.'# time-consuming task def long_task (arg1 Arg2): print ("args:% s% s!"% (arg1, arg2)) sleep (5) print ("Task is done!") if _ _ name__ = ='_ main__': app.run ()

You can use this method when you want to perform some relatively simple and time-consuming tasks, such as sending email, SMS CAPTCHA, and so on.

But the problem with this approach is that the front end cannot know the status of the task.

If you want the front end to know, you need to design some logic, such as storing the task execution state in redis, identifying it by a unique task id, and then writing an interface to obtain the task status through the task id, and then let the front end request the interface regularly to obtain the task status information.

It's a bit of a hassle to implement it all on your own, and Celery just implements this logic to use it.

Use Celery

To meet the need for the front end to get the status of the task, you can use Celery.

Celery is a distributed task queue for real-time task processing and scheduling, which is often used for web asynchronous tasks, scheduled tasks and so on. A separate article is written to describe the architecture of Celery, which will not be discussed in depth here.

Now I want the front end to use a progress bar to determine the execution of back-end tasks. It's easy to use Celery, first install Celery and redis through pip, and install redis because Celery chooses redis as the "message broker / message middleware".

Pip install celerypip install redis

Using Celery in Flask is actually very simple. Here, we simply go through the overall process of using Celery in Flask, and then implement a specific project.

1. Initialize Celery in Flask

From flask import Flaskfrom celery import Celeryapp = Flask (_ _ name__) # configure # configure the path of the message broker if it is on a remote server Then configure the URLapp.config of the redis in the remote server ['CELERY_BROKER_URL'] =' redis://localhost:6379/0'# to store the status or run results of the Celery task, you must configure app.config ['CELERY_RESULT_BACKEND'] =' redis://localhost:6379/0'# initialization Celerycelery = Celery (app.name, broker=app.config ['CELERY_BROKER_URL']) # pass the configuration in Flask directly to Celerycelery.conf.update (app.config)

In the above code, the celery object is initialized through the Celery class, and the application name passed in is connected to the message broker URL.

two。 Decorate the corresponding functions of time-consuming tasks through the celery.task decorator

@ celery.taskdef long_task (arg1, arg2): # logical return result of time-consuming tasks

Interfaces defined in 3.Flask perform time-consuming tasks asynchronously

@ app.route ('/', methods= ['GET',' POST']) def index (): task = long_task.delay (1,2) delay () method is a shortcut to the applyasync () method. With more applyasync () parameters, time-consuming tasks can be controlled more carefully. For example, if you want long_task () to execute @ app.route ('/', methods= ['GET',' POST']) def index (): task = long_task.apply_async (args= [1,2], countdown=60) after one minute.

Delay () and apply_async () return a task object that can get the status of the task and all kinds of related information.

With these three steps, you can use Celery.

Then we will specifically realize the requirement of "allowing the front end to judge the execution of back-end tasks through a progress bar".

# bind is True Self will be passed to the decorated method @ celery.task (bind=True) def long_task (self): 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: # get some information randomly message ='{0} {1} {2}... '.format (random.choice (verb), random.choice (adjective), random.choice (noun)) # Update Celery task status self.update_state (state='PROGRESS', meta= {' current': I, 'total': total) 'status': message}) time.sleep (1) # returns the dictionary return {' current': 100, 'total': 100,' status': 'Task completed dictionary,' result': 42}

In the above code, the celery.task () decorator uses the bind=True parameter, which causes Celery to pass in the Celery itself, which can be used to record and update the task status.

Then there is a for iteration, the logic of the iteration is meaningless, just randomly extract some words from the list to simulate the operation of some logic, to show that this is time-consuming logic, sleep through time.sleep (1) for a second.

Each time you get a vocabulary, the status of the Celery task is updated through self.update_state (). Celery contains some built-in states, such as SUCCESS, STARTED, and so on. The custom state "PROGRESS" is used here. In addition to the state, some information about this loop is stored in a dictionary through meta parameters (metadata). With this data, the front end can display a progress bar.

After defining the time-consuming method, define a Flask interface method to call the time-consuming method

@ app.route ('/ longtask', methods= ['POST']) def longtask (): # Asynchronous call task = long_task.apply_async () # returns 202, with location header return jsonify ({}), 202, {' Location': url_for ('taskstatus', task_id=task.id)}

To put it simply, the front end requests to / longtask through POST, allowing the back end to start performing time-consuming tasks.

The returned status code of 202202 usually indicates that a request is in progress, and then the Location header information is added to the Header of the return packet. The front end can obtain the complete url corresponding to the task id by reading the information of the Location in the Header in the packet.

After the frontend has the url corresponding to the task id, it also needs to provide an API to the frontend, so that the frontend can obtain the specific status of the task at the current time through the task id.

@ app.route ('/ status/') def taskstatus (task_id): task = long_task.AsyncResult (task_id) if task.state = = 'PENDING': # waiting for data in response = {' state': task.state, 'current': 0,' total': 1, 'status':' Pending...'} elif task.state! = 'FAILURE': # there is no failure response = {' state': task.state, # status # meta Through task.info.get (), you can get 'current': task.info.get (' current', 0), # current cycle progress' total': task.info.get ('total', 1), # total cycle progress' status': task.info.get ('status' '')} if 'result' in task.info: response [' result'] = task.info ['result'] else: # there are some problems with backend task execution response = {' state': task.state, 'current': 1,' total': 1, 'status': str (task.info), # specific exception reported} return jsonify (response)

In order to get the information in the task object, use the task id to initialize the AsyncResult class, get the task object, and then get the information about the current task from the task object.

This method returns a JSON containing the task status and the information specified in the meta, which the front end can use to build a progress bar.

If the task is in PENDING status, it means that the task has not yet started. In this state, there is no information in the task. Here, some data is returned artificially. If the task fails, the exception information contained in the task.info is returned. In addition, the task is executed normally, and the specific information in the task can be obtained through task.info.

In this way, the back-end logic is processed, and then the front-end logic is implemented. To implement a graphical progress bar, you can directly use nanobar.js. A progress bar can be implemented in two simple sentences. An example of its official website is as follows:

Var options = {classname: 'my-class', id:' my-id', / / where the progress bar will appear target: document.getElementById ('myDivId')}; / / initialize the progress bar object var nanobar = new Nanobar (options); nanobar.go (30); / / 30% progress bar nanobar.go (76); / 76% progress bar / / 100% progress bar end nanobar.go (100)

With nanobar.js, it's very simple.

Define a simple HTML interface first

Long running task with progress updatesStart Long Calculation

Realize the request to the background through JavaScript

/ / Button Click event $(function () {$('# start-bg-job') .click (start_long_task);}); / / request longtask interface function start_long_task () {/ / add element in html div = $('0%. '); $(' # progress') .append (div); / / create a progress bar object var nanobar = new Nanobar ({bg:'# 44fallows, target: div [0] .childNodes [0]}); / / ajax requests longtask $.ajax ({type: 'POST', url:' / longtask', / / to get data and get Location success: function (data, status, request) {status_url = request.getResponseHeader ('Location') from the response header / call the update_progress () method to update the progress bar update_progress (status_url, nanobar, div [0]);}, error: function () {alert ('Unexpected error');}}) } / / Update progress bar function update_progress (status_url, nanobar, status_div) {/ / getJSON () method is the JQuery built-in method. Here, request is made to the corresponding url in Location, that is, request "/ status/" $.getJSON (status_url, function (data) {/ / calculate progress percent = parseInt (data ['current'] * 100 / data [' total']); / / update progress bar nanobar.go (percent) / / Update text $(status_div.childNodes [1]) .text (percent +'%'); $(status_div.childNodes [2]) .text (data ['status']); if (data [' state']! = 'PENDING' & & data [' state']! = 'PROGRESS') {if (' result' in data) {/ / display the result $(status_div.childNodes [3]). Text ('Result:' + data ['result']) } else {/ / unexpected things happen $(status_div.childNodes [3]). Text ('Result:' + data ['state']);} else {/ / 2 seconds later run setTimeout again (function () {update_progress (status_url, nanobar, status_div);}, 2000);});})

You can read the overall logic of the code through comments.

At this point, the requirements have been implemented, run it.

Run Redis first

Redis-server

Then run celery

Celery worker-An app.celery-- loglevel=info

Finally run the Flask project

This is the end of python app.py 's article on "how to implement tasks asynchronously in Flask". Thank you for reading! I believe you all have a certain understanding of the knowledge of "how to implement tasks asynchronously in Flask". If you want to learn more, you are welcome to follow the industry information channel.

Welcome to subscribe "Shulou Technology Information " to get latest news, interesting things and hot topics in the IT industry, and controls the hottest and latest Internet news, technology news and IT industry trends.