from crontab import CronTab import re from ..calls.create import Task import logging def post(task_id=None, period=None, user=None, force=False): """ Change schedule of the given task Parameters ---------- task_id: str Id of task to be change period: str Cron formatted schedule string user: str Task user Returns ------- result: str result[0]: message result[1]: bool if it is deleted or not Notes ----- % in any of string make script falls. Examples -------- >>> # Change stopped task >>> import requests >>> import json >>> params = {"user": 'test', "task_id": '8485#&', "period": 'minute.every(1)'} >>> requests.post(url='/MG/routines/change', json=params) ['8485#& succesfully changed', True] >>> # Change started task >>> params = {"user": 'test', "task_id": '8485#&'} >>> requests.post(url='/MG/routines/change', json=params) ['8485#& succesfully stopped', True] >>> params = {"user": 'test', "task_id": '8485#&', "period": 'minute.every(1)'} >>> requests.post(url='/MG/routines/change', json=params) ['8485#& succesfully changed', True] >>> # Wrong task_id >>> params = {"user": 'test', "task_id": '0000000'} >>> requests.post(url='/MG/routines/change', json=params) ['Wrong task id. This task does not exists.', False] """ period = period user = Task.clean_string(user) task_id = Task.clean_string(task_id) user_cron = CronTab(user=True) jobs = [job for job in user_cron.find_comment('{}:{}'.format(user, task_id))] if len(jobs) > 1: logging.warning('There are two task with the same id, {}.'.format(task_id)) result = 'There are two task with the same id.' elif len(jobs) == 0: result = ['Wrong task id. This task does not exists.', False] else: job = jobs[0] eval('job.{}'.format(period)) job.enable(True) user_cron.write() result = ['{} successfully changed'.format(task_id), True] return result