Flask Progress Reporting

When running long-running tasks in a Flask application, it's often beneficial to provide users with feedback on the task's progress. In this tutorial, we'll explore how to implement progress reporting for asynchronous tasks in Flask applications, allowing users to see real-time updates as their tasks execute.

Introduction to Progress Reporting

Asynchronous tasks are excellent for handling time-consuming operations without blocking the main application thread. However, one challenge is keeping users informed about the status of their tasks. Progress reporting solves this problem by:

Providing real-time feedback to users
Giving estimates on completion time
Improving user experience during long operations
Allowing users to make informed decisions (like canceling tasks that take too long)

Prerequisites

Before diving into progress reporting, make sure you have:

A working Flask application
Celery set up for asynchronous task processing
Redis or another message broker for Celery
Basic understanding of WebSockets (we'll use Flask-SocketIO)

Setting Up Your Environment

First, let's install the necessary packages:

bash
pip install Flask Flask-SocketIO celery redis

Basic Architecture for Progress Reporting

Our progress reporting system will consist of these components:

Celery Tasks: Long-running operations that report their progress
Flask-SocketIO: For real-time communication between server and client
Redis Backend: To store task state and progress information
Frontend Components: To display progress to users

Implementation Steps

1. Configure Celery with Result Backend

First, ensure Celery is properly set up with a result backend to store task state:

python
# app.py
from flask import Flask
from celery import Celery

app = Flask(__name__)
app.config.update(
    CELERY_BROKER_URL='redis://localhost:6379/0',
    CELERY_RESULT_BACKEND='redis://localhost:6379/0'
)

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

2. Create a Task with Progress Updates

Let's create a Celery task that reports its progress:

python
# tasks.py
from celery import current_task

@celery.task(bind=True)
def long_task(self, items_count):
    """
    Task that processes items and reports progress
    """
    import time
    
    for i in range(items_count):
        # Do some work
        time.sleep(0.5)  # Simulate work with delay
        
        # Update task state
        self.update_state(
            state='PROGRESS',
            meta={
                'current': i + 1,
                'total': items_count,
                'percent': (i + 1) * 100 / items_count,
                'status': f'Processing item {i + 1} of {items_count}'
            }
        )
    
    # Return final results
    return {
        'current': items_count,
        'total': items_count,
        'percent': 100,
        'status': 'Task completed!'
    }

3. Set Up Flask-SocketIO

Now, let's integrate Flask-SocketIO for real-time updates:

python
# app.py (continued)
from flask_socketio import SocketIO, emit

socketio = SocketIO(app, cors_allowed_origins="*")

# SocketIO event handlers
@socketio.on('connect')
def handle_connect():
    print('Client connected')

@socketio.on('disconnect')
def handle_disconnect():
    print('Client disconnected')

4. Create Flask Routes to Start Tasks and Check Progress

python
# app.py (continued)
from flask import jsonify, request, render_template
from tasks import long_task

@app.route('/')
def index():
    return render_template('index.html')

@app.route('/start_task', methods=['POST'])
def start_background_task():
    items_count = int(request.form.get('items_count', 10))
    task = long_task.delay(items_count)
    return jsonify({"task_id": task.id}), 202

@app.route('/task_status/<task_id>')
def task_status(task_id):
    task = long_task.AsyncResult(task_id)
    if task.state == 'PENDING':
        response = {
            'state': task.state,
            'status': 'Task pending...'
        }
    elif task.state == 'PROGRESS':
        response = {
            'state': task.state,
            'current': task.info.get('current', 0),
            'total': task.info.get('total', 1),
            'percent': task.info.get('percent', 0),
            'status': task.info.get('status', '')
        }
    elif task.state == 'SUCCESS':
        response = {
            'state': task.state,
            'current': task.info.get('current', 0),
            'total': task.info.get('total', 1),
            'percent': task.info.get('percent', 0),
            'status': task.info.get('status', '')
        }
    else:
        response = {
            'state': task.state,
            'status': str(task.info)  # Error message
        }
    return jsonify(response)

5. Create Frontend Templates for Progress Display

html
<!-- templates/index.html -->
<!DOCTYPE html>
<html>
<head>
    <title>Flask Progress Reporting</title>
    <meta charset="UTF-8">
    <meta name="viewport" content="width=device-width, initial-scale=1.0">
    <script src="https://cdnjs.cloudflare.com/ajax/libs/socket.io/4.0.1/socket.io.js"></script>
    <script src="https://code.jquery.com/jquery-3.6.0.min.js"></script>
    <style>
        .progress-container {
            width: 100%;
            max-width: 600px;
            margin: 20px auto;
        }
        .progress {
            height: 30px;
            background-color: #f5f5f5;
            border-radius: 5px;
            margin-bottom: 10px;
        }
        .progress-bar {
            height: 100%;
            background-color: #4CAF50;
            text-align: center;
            line-height: 30px;
            color: white;
            border-radius: 5px;
            transition: width 0.5s;
        }
        .status {
            margin-bottom: 20px;
            font-family: sans-serif;
        }
    </style>
</head>
<body>
    <div class="progress-container">
        <h2>Task Progress</h2>
        
        <form id="start-task-form">
            <label for="items_count">Number of items to process:</label>
            <input type="number" id="items_count" name="items_count" value="10" min="1" max="100">
            <button type="submit">Start Task</button>
        </form>
        
        <div class="progress">
            <div class="progress-bar" role="progressbar" style="width: 0%;">0%</div>
        </div>
        
        <div class="status">Task status will appear here...</div>
    </div>
    
    <script>
        $(document).ready(function() {
            let taskId = null;
            const progressBar = $('.progress-bar');
            const statusDiv = $('.status');
            
            // Connect to SocketIO
            const socket = io.connect(location.protocol + '//' + document.domain + ':' + location.port);
            
            // Handle form submission
            $('#start-task-form').submit(function(event) {
                event.preventDefault();
                
                // Reset progress
                progressBar.css('width', '0%').text('0%');
                statusDiv.text('Starting task...');
                
                // Submit form via AJAX to start the task
                $.ajax({
                    type: 'POST',
                    url: '/start_task',
                    data: $(this).serialize(),
                    success: function(data) {
                        taskId = data.task_id;
                        checkTaskStatus();
                    }
                });
            });
            
            // Function to check task status
            function checkTaskStatus() {
                if (!taskId) return;
                
                $.getJSON('/task_status/' + taskId, function(data) {
                    // Update progress bar
                    const percent = data.percent || 0;
                    progressBar.css('width', percent + '%').text(Math.round(percent) + '%');
                    
                    // Update status message
                    statusDiv.text(data.status || 'Processing...');
                    
                    // If task is not complete, check again in 1 second
                    if (data.state !== 'SUCCESS' && data.state !== 'FAILURE') {
                        setTimeout(checkTaskStatus, 1000);
                    }
                });
            }
        });
    </script>
</body>
</html>

6. Enhanced Real-time Updates with WebSockets

For a more efficient approach, we can push updates directly to the client using WebSockets instead of polling:

python
# tasks.py (modified)
from celery import current_task
from app import socketio

@celery.task(bind=True)
def long_task_with_websocket(self, items_count):
    """
    Task that processes items and reports progress via WebSockets
    """
    import time
    
    for i in range(items_count):
        # Do some work
        time.sleep(0.5)  # Simulate work with delay
        
        # Calculate progress
        current = i + 1
        percent = (current * 100) / items_count
        progress_data = {
            'task_id': self.request.id,
            'current': current,
            'total': items_count,
            'percent': percent,
            'status': f'Processing item {current} of {items_count}'
        }
        
        # Update task state in Celery
        self.update_state(
            state='PROGRESS',
            meta=progress_data
        )
        
        # Emit WebSocket event with progress
        socketio.emit('task_progress', progress_data)
    
    # Final results
    result = {
        'task_id': self.request.id,
        'current': items_count,
        'total': items_count,
        'percent': 100,
        'status': 'Task completed!'
    }
    
    # Emit completion event
    socketio.emit('task_completed', result)
    
    return result

Update your frontend JavaScript to handle the WebSocket events:

javascript
// Add this to the script in index.html
socket.on('task_progress', function(data) {
    // Only update if this is our task
    if (data.task_id === taskId) {
        const percent = data.percent || 0;
        progressBar.css('width', percent + '%').text(Math.round(percent) + '%');
        statusDiv.text(data.status || 'Processing...');
    }
});

socket.on('task_completed', function(data) {
    if (data.task_id === taskId) {
        progressBar.css('width', '100%').text('100%');
        statusDiv.text(data.status || 'Task completed!');
    }
});

Running the Application

Start your application components:

Start Redis:

bash
redis-server

Start Celery worker:

bash
celery -A app.celery worker --loglevel=info

Start the Flask application:

bash
python -m flask run

Real-World Use Cases

Progress reporting is particularly valuable in these scenarios:

1. File Processing

When uploading and processing large files, show progress as each file chunk is processed:

python
@celery.task(bind=True)
def process_large_file(self, file_path):
    total_lines = sum(1 for _ in open(file_path, 'r'))
    processed = 0
    
    with open(file_path, 'r') as file:
        for line in file:
            # Process the line
            process_data(line)
            
            processed += 1
            # Update progress every 100 lines
            if processed % 100 == 0:
                percent = (processed * 100) / total_lines
                self.update_state(
                    state='PROGRESS',
                    meta={
                        'current': processed, 
                        'total': total_lines,
                        'percent': percent
                    }
                )
    
    return {'status': 'File processing complete'}

2. Data Import/Export

When importing a large dataset into your application:

python
@celery.task(bind=True)
def import_data(self, data_source, items):
    total = len(items)
    
    for i, item in enumerate(items):
        # Import logic
        save_to_database(item)
        
        # Update progress every 10 items
        if i % 10 == 0:
            self.update_state(
                state='PROGRESS',
                meta={
                    'current': i + 1,
                    'total': total,
                    'percent': ((i + 1) * 100) / total,
                    'status': f'Imported {i + 1} of {total} items'
                }
            )
    
    return {'status': 'Import completed', 'count': total}

3. Data Analysis

When performing complex data analysis that takes time:

python
@celery.task(bind=True)
def analyze_dataset(self, dataset_id):
    # Load dataset
    dataset = load_dataset(dataset_id)
    steps = [
        ('data_cleaning', 0.2),
        ('feature_extraction', 0.3),
        ('model_training', 0.4),
        ('evaluation', 0.1)
    ]
    
    progress = 0
    results = {}
    
    for step_name, step_weight in steps:
        # Perform analysis step
        step_result = perform_analysis_step(dataset, step_name)
        results[step_name] = step_result
        
        # Update progress
        progress += step_weight * 100
        self.update_state(
            state='PROGRESS',
            meta={
                'current': progress,
                'total': 100,
                'percent': progress,
                'status': f'Completed {step_name}'
            }
        )
    
    return {
        'status': 'Analysis complete',
        'results': results
    }

Best Practices for Progress Reporting

Don't Report Too Frequently: Sending progress updates for every tiny change can overload your message broker. Update at reasonable intervals.
Include Meaningful Information: Don't just report percentages; include helpful status messages.
Handle Errors Gracefully: Make sure errors in the task are reported to the frontend.
Clean Up After Completion: Set appropriate expiry times for task results in your result backend.
Provide Cancellation Options: When possible, allow users to cancel long-running tasks.

Summary

In this tutorial, we've learned how to implement progress reporting for asynchronous tasks in Flask applications using Celery and WebSockets. We've covered:

Setting up Celery with a result backend to store task state
Creating tasks that report their progress
Using Flask routes to start tasks and check their status
Implementing WebSockets for real-time progress updates
Building a frontend interface to display task progress
Applying progress reporting to real-world scenarios

By implementing proper progress reporting, you can significantly enhance the user experience of your Flask applications, particularly when dealing with long-running operations.

Additional Resources

Exercises

Basic Progress: Modify the simple long-running task to include more detailed progress information (like estimated time remaining).
File Upload Progress: Create a Flask application that shows progress when uploading and processing a large CSV file.
Multiple Tasks: Extend the progress reporting system to handle multiple concurrent tasks for the same user.
Task Control: Add the ability to pause, resume, or cancel tasks that are in progress.
Error Handling: Enhance the progress reporting to handle and display errors that occur during task execution.

If you spot any mistakes on this website, please let me know at [email protected]. I’d greatly appreciate your feedback! :)

Introduction to Progress Reporting​

Prerequisites​

Setting Up Your Environment​

Basic Architecture for Progress Reporting​

Implementation Steps​

1. Configure Celery with Result Backend​

2. Create a Task with Progress Updates​

3. Set Up Flask-SocketIO​

4. Create Flask Routes to Start Tasks and Check Progress​

5. Create Frontend Templates for Progress Display​

6. Enhanced Real-time Updates with WebSockets​

Running the Application​

Real-World Use Cases​

1. File Processing​

2. Data Import/Export​

3. Data Analysis​

Best Practices for Progress Reporting​

Summary​

Additional Resources​

Exercises​