Task status monitoring in Flask

Introduction

Task status monitoring helps you know if a background job is running, done, or failed. It keeps users informed about progress.

When you start a long process like file upload or data processing and want to show progress.

When you run background tasks and want to check if they finished successfully.

When you want to update a web page automatically with the current state of a task.

When you need to debug or log the status of tasks in your Flask app.

Syntax

Flask

from flask import Flask, jsonify

app = Flask(__name__)
tasks = {}

@app.route('/start-task')
def start_task():
    task_id = 'task1'
    tasks[task_id] = 'running'
    # simulate task
    tasks[task_id] = 'completed'
    return jsonify({'task_id': task_id, 'status': tasks[task_id]})

@app.route('/task-status/<task_id>')
def task_status(task_id):
    status = tasks.get(task_id, 'not found')
    return jsonify({'task_id': task_id, 'status': status})

Use a dictionary or database to store task statuses.

Define routes to start tasks and check their status.

Examples

Store and update task status in a dictionary.

Flask

tasks = {}
tasks['task1'] = 'running'
tasks['task1'] = 'completed'

Check and return the status of a task by its ID.

Flask

@app.route('/task-status/<task_id>')
def task_status(task_id):
    status = tasks.get(task_id, 'not found')
    return jsonify({'task_id': task_id, 'status': status})

Sample Program

This Flask app starts a background task that runs for 3 seconds. You can start the task via /start-task and check its status via /task-status/task1.

Flask

from flask import Flask, jsonify
import threading
import time

app = Flask(__name__)
tasks = {}

def background_task(task_id):
    tasks[task_id] = 'running'
    time.sleep(3)  # simulate work
    tasks[task_id] = 'completed'

@app.route('/start-task')
def start_task():
    task_id = 'task1'
    thread = threading.Thread(target=background_task, args=(task_id,))
    thread.start()
    return jsonify({'task_id': task_id, 'status': 'started'})

@app.route('/task-status/<task_id>')
def task_status(task_id):
    status = tasks.get(task_id, 'not found')
    return jsonify({'task_id': task_id, 'status': status})

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

Important Notes

Use threads or background workers to avoid blocking the main Flask app.

Task IDs should be unique if you run multiple tasks.

For production, consider using task queues like Celery for better management.

Summary

Task status monitoring lets users see progress of background jobs.

Store task states in a dictionary or database.

Use Flask routes to start tasks and check their status.