parameters: ()
docstring: 清除celery定时任务
docstring_summary: 清除celery定时任务
docstring_tokens: 清除celery定时任务
function: def clean_celery_periodic_tasks():
      """清除celery定时任务"""
      need_cleaned_tasks = [
          'handle_be_interrupted_change_auth_task_periodic',
      ]
      logger.info('Start clean celery periodic tasks: {}'.format(need_cleaned_tasks))
      for task_name in need_cleaned_tasks:
          logger.info('Start clean task: {}'.format(task_name))
          task = get_celery_periodic_task(task_name)
          if task is None:
              logger.info('Task does not exist: {}'.format(task_name))
              continue
          disable_celery_periodic_task(task_name)
          delete_celery_periodic_task(task_name)
          task = get_celery_periodic_task(task_name)
          if task is None:
              logger.info('Clean task success: {}'.format(task_name))
          else:
              logger.info('Clean task failure: {}'.format(task))
function_tokens: def clean_celery_periodic_tasks ( ) : need_cleaned_tasks = [ 'handle_be_interrupted_change_auth_task_periodic' , ] logger . info ( 'Start clean celery periodic tasks: {}' . format ( need_cleaned_tasks ) ) for task_name in need_cleaned_tasks : logger . info ( 'Start clean task: {}' . format ( task_name ) ) task = get_celery_periodic_task ( task_name ) if task is None : logger . info ( 'Task does not exist: {}' . format ( task_name ) ) continue disable_celery_periodic_task ( task_name ) delete_celery_periodic_task ( task_name ) task = get_celery_periodic_task ( task_name ) if task is None : logger . info ( 'Clean task success: {}' . format ( task_name ) ) else : logger . info ( 'Clean task failure: {}' . format ( task ) )