I am trying to use celery but i faced a problem.
I checked in celery_taskmeta table in database and I saw the task status is SUCCESS, but it raise an Error in the log.
this is my celery code:
@celery.task(name="celery_sample", bind=True, track_started=True)
def celery_sample_task(self, input_data: dict):
try:
celery_sample = CelerySampleService()
ret = celery_sample .sample_function(input_data)
result = ret['result']
save_path = './services/resul/'
file_name = datetime.now().strftime("%Y_%m_%d-%I_%M_%S_%p")
full_file_path = os.path.join(save_path, file_name + '.txt')
file = open(full_file_path, 'w')
file.write(result)
file.close()
return full_file_path
except Exception as e:
return str(e)
I think celery status is always SUCCESS if I use try…except?
Any idea?
DDManh is a new contributor to this site. Take care in asking for clarification, commenting, and answering.
Check out our Code of Conduct.