celery-progress

import os import time from celery import shared_task from celery_progress.backend import ProgressRecorder # settings.py # INSTALLED_APPS = [ # ..., # 'celery_progress', # ] # CELERY_BROKER_URL = os.environ.get('CELERY_BROKER_URL', 'redis://localhost:6379/0') # CELERY_RESULT_BACKEND = os.environ.get('CELERY_RESULT_BACKEND', 'redis://localhost:6379/0') # urls.py (in your project's main urls.py) # from django.urls import path, include # urlpatterns = [ # ..., # path('celery-progress/', include('celery_progress.urls')), # ] @shared_task(bind=True) def my_long_task(self, total_steps): progress_recorder = ProgressRecorder(self) for i in range(total_steps): time.sleep(1) # Simulate work progress_recorder.set_progress(i + 1, total_steps, description=f'Step {i+1} of {total_steps}') return 'Task Completed!' # views.py # from django.shortcuts import render # from .tasks import my_long_task # def start_task_view(request): # if request.method == 'POST': # result = my_long_task.delay(10) # Start the task asynchronously # return render(request, 'task_progress.html', {'task_id': result.task_id}) # return render(request, 'start_task.html') # task_progress.html (template) # <div class='progress-wrapper'> # <div id='progress-bar' class='progress-bar' style="background-color: #68a9ef; width: 0%;">&nbsp;</div> # </div> # <div id="progress-bar-message">Waiting for progress to start...</div> # <div id="celery-result"></div> # # {% load static %} # <script src="{% static 'celery_progress/celery_progress.js' %}"></script> # <script> # document.addEventListener("DOMContentLoaded", function() { # var progressUrl = "{% url 'celery_progress:task_status' task_id %}"; # CeleryProgressBar.initProgressBar(progressUrl, { # onSuccess: function(result, elm) { # document.getElementById('celery-result').innerHTML = 'Result: ' + result; # } # }); # }); # </script>