在我的Django项目中,我使用Celery(以及其他一些东西),有一个Celery任务会在后台上传文件到数据库。我使用轮询来跟踪上传进度,并显示上传的进度条。以下是一些细节描述上传过程的代码片段:
views.py:
当上传成功时,我想要执行类似以下操作:
(这两个代码片段主要来自之前关于Django + Celery进度条的StackOverflow问题。)
views.py:
from .tasks import upload_task
...
upload_task.delay(datapoints, user, description) # datapoints is a list of dictionaries, user and description are simple strings
tasks.py:
from taskman.celery import app, DBTask # taskman is the name of the Django app that has celery.py
from celery import task, current_task
@task(base=DBTask)
def upload_task(datapoints, user, description):
from utils.db.databaseinserter import insertIntoDatabase
for count in insertIntoDatabase(datapoints, user, description):
percent_completion = int(100 * (float(count) / float(len(datapoints))))
current_task.update_state(state='PROGRESS', meta={'percent':percent_completion})
databaseinserter.py:
def insertIntoDatabase(datapoints, user, description):
# iterate through the datapoints and upload them one by one
# at the end of an iteration, yield the number of datapoints completed so far
上传代码运行正常,进度条也正常工作。但我不确定如何发送Django消息来告诉用户上传完成(或在出现错误时发送Django消息通知用户错误)。当上传开始时,我在views.py中执行以下操作:
from django.contrib import messages
...
messages.info(request, "Upload is in progress")
当上传成功时,我想要执行类似以下操作:
messages.info(request, "Upload successful!")
由于Celery任务是“fire and forget”,因此我无法在views.py中执行此操作。是否可以在celery.py中执行此操作?在我的celery.py中的DBTask
类中,我定义了on_success
和on_failure
,那么我能否从那里发送Django消息?
此外,虽然我的轮询技术在技术上可行,但目前并不理想。当前轮询的方式是无论是否正在进行任务,它都会无休止地检查任务。它会迅速地淹没服务器控制台日志,并且可能对性能产生负面影响。我对编写轮询代码比较新,因此我不太确定最佳实践以及如何仅在需要时轮询。处理常量轮询和服务器日志阻塞的最佳方法是什么?以下是我的轮询代码。
views.py:
def poll_state(request):
data = 'Failure'
if request.is_ajax():
if 'task_id' in request.POST.keys() and request.POST['task_id']:
task_id = request.POST['task_id']
task = AsyncResult(task_id)
data = task.result or task.state
if data == 'SUCCESS' or data == 'FAILURE': # not sure what to do here; what I want is to exit the function early if the current task is already completed
return HttpResponse({}, content_type='application/json')
else:
data ='No task_id in the request'
logger.info('No task_id in the request')
else:
data = 'Not an ajax request'
logger.info('Not an ajax request')
json_data = json.dumps(data)
return HttpResponse(json_data, content_type='application/json')
相应的jQuery代码如下:
{% if task_id %}
jQuery(document).ready(function() {
var PollState = function(task_id) {
jQuery.ajax({
url: "poll_state",
type: "POST",
data: "task_id=" + task_id,
}).done(function(task) {
if (task.percent) {
jQuery('.bar').css({'width': task.percent + '%'});
jQuery('.bar').html(task.percent + '%');
}
else {
jQuery('.status').html(task);
};
PollState(task_id);
});
}
PollState('{{ task_id }}');
})
{% endif %}
(这两个代码片段主要来自之前关于Django + Celery进度条的StackOverflow问题。)