Celery-Progress - progress_recorder.set_progress() 返回 TypeEr

Question

我在一个Django项目中试图使用Celery-Progress创建一个进度条。

但是当我尝试在以下函数中使用 progress_recorder.set_progress 方法(使用芹菜工人(时：

def increase_task_progress():
global progress_recorder_current, progress_recorder_total, progress_recorder
progress_recorder_current += 1
progress_recorder.set_progress(progress_recorder_current, progress_recorder_total) <<<

它给出了以下异常：

Traceback (most recent call last):
File "/usr/lib/python3.7/threading.py", line 917, in _bootstrap_inner
self.run()
File "/usr/lib/python3.7/threading.py", line 865, in run
self._target(*self._args, **self._kwargs)
File "/projects/report/excel.py", line 193, in generate_data_sheets
increase_task_progress()
File "/projects/report/excel.py", line 199, in increase_task_progress
progress_recorder.set_progress(progress_recorder_current, progress_recorder_total)
File "/projects/report/my_venv/lib/python3.7/site-packages/celery_progress/backend.py", line 38, in set_progress
'percent': percent,
File "/projects/report/my_venv/lib/python3.7/site-packages/celery/app/task.py", line 930, in update_state
self.backend.store_result(task_id, meta, state, **kwargs)
File "/projects/report/my_venv/lib/python3.7/site-packages/celery/backends/base.py", line 342, in store_result
request=request, **kwargs)
File "/projects/report/my_venv/lib/python3.7/site-packages/celery/backends/base.py", line 714, in _store_result
self.set(self.get_key_for_task(task_id), self.encode(meta))
File "/projects/report/my_venv/lib/python3.7/site-packages/celery/backends/base.py", line 590, in get_key_for_task
self.task_keyprefix, key_t(task_id), key_t(key),
TypeError: sequence item 1: expected a bytes-like object, NoneType found

这些是我的代码，直到达到increase_task_progress功能。

views.py

from django.http                import HttpResponse
from . import tasts
def myView(request):
report_id = request.POST.get('report_id', 0)
result = tasks.generate_report.delay(report_id)
response = HttpResponse(content=task.task_id) 
return response    

tasks.py

from . import excel
from celery                     import shared_task
@shared_task(bind=True)
def generate_report(self, report_id):
task = excel.generate_xls_report(
report_id=report_id
task_obj=self
)
return task

excel.py

from celery_progress.backend    import ProgressRecorder
from xlsxwriter.workbook        import Workbook
from .                          import database
import threading, io
workbook = None
progress_recorder = None
progress_recorder_current = 0
progress_recorder_total = 4
def generate_xls_report(report_id, task_obj):
global progress_recorder, workbook
progress_recorder = ProgressRecorder(task=task_obj)
output = io.BytesIO()
workbook = Workbook(output, {'constant_memory': True})
db_conn = database.get_db_conn()
thread_product = threading.Thread(target=generate_data_sheets, args=("product", db_conn))
thread_product.start()
thread_sales = threading.Thread(target=generate_data_sheets, args=("sales", db_conn))
thread_sales.start()
thread_product.join()
thread_sales.join()
db_conn.close()
workbook.close()
output.seek(0)
print("Program finished!")
return workbook
def generate_data_sheets(table_name, db_conn):
#Generate the excel data sheet code HERE...
increase_task_progress()
print("Table {} done".format(table_name))
def increase_task_progress():
global progress_recorder_current, progress_recorder_total, progress_recorder
progress_recorder_current += 1
progress_recorder.set_progress(progress_recorder_current, progress_recorder_total)

拜托，我的代码有什么问题？

谢谢

Answer 1

我设法通过创建一个名为"ReportGenerator"的类并使用线程来解决此问题。Lock(( for python.

请参阅下面的已解决代码：

class ReportGenerator(object):
def __init__(self, start_datetime, end_datetime, task_obj):
self.lock = threading.Lock()    
self.task_obj = task_obj
self.task_id = self.task_obj.request.id
self.progress_recorder_current = 0
self.progress_recorder_total = 6
def xlsx_generator(self):
#Xlsx report first steps here
#Threads starts and joins that calls "generate_data_sheet()" method
return workbook
def generate_data_sheet(self):
#Put data in xlsx rows...
self.increase_task_progress()
def increase_task_progress(self):
self.lock.acquire()
self.progress_recorder_current += 1
percent = (Decimal(self.progress_recorder_current) / Decimal(self.progress_recorder_total)) * Decimal(100)
percent = float(round(percent, 2))
try:
self.task_obj.update_state(
task_id=self.task_id, # I had to pass the task_id to update_state
state="PROGRESS",
meta={
'current': self.progress_recorder_current,
'total': self.progress_recorder_total,
'percent': percent,
}
)
finally:
self.lock.release()