In Python, there are several solutions to implement background tasks, such as Celery or Redis Queue, which are good ways to implement a task queue system. But using both is cumbersome. Imagine a scenario where we use Flask to build an API to call background tasks from one terminal and stop the background tasks using another terminal.
Built a simple API using Flask, with two main methods, one for starting background tasks and the other for stopping.
To manage the life cycle of tasks, we use Event Objects
, which is a simple inter-thread communication mechanism.
The following are the required imported libraries, thread event declarations and background task methods:
from time import sleep from flask import Flask from flask_cors import CORS import threading thread_event = threading.Event() def backgroundTask(): while thread_event.is_set(): print('Background task running!') sleep(5)
The key here is the is_set()
method, which will return internal The value of the thread event flag: true
or false
.
First, use the set()
method to set the flag to true
, which will start a thread and run the backgroundTask
method continuously.
@app.route("/start", methods=["POST"]) def startBackgroundTask(): try: thread_event.set() thread = threading.Thread(target=backgroundTask) thread.start() return "Background task started!" except Exception as error: return str(error)
If you want to stop the task, call the clear()
method to set the flag to false
to stop the running thread.
@app.route("/stop", methods=["POST"]) def stopBackgroundTask(): try: thread_event.clear() return "Background task stopped!" except Exception as error: return str(error)
The above is the detailed content of How to use Python Flask to build efficient API applications and implement background tasks. For more information, please follow other related articles on the PHP Chinese website!