for testing and deploying your application
for finding and fixing issues
for empowering human code reviews
import threading
import time
from datetime import datetime
import schedule
from chaoswg import Task
class TaskScheduler(threading.Thread):
__class__
def __init__(self, interval=60):
# init a daemonized thread
threading.Thread.__init__(self, daemon=True)
self.interval = interval
# run an update every hour
schedule.every().hour.do(self.schedule_tasks)
@staticmethod
def schedule_tasks():
now = datetime.utcnow()
schedule_tasks = Task.get_schedule_tasks()
for task in schedule_tasks:
if task['last_done'] is not None:
delta = now - task['last_done']
if delta.days >= task['schedule_days']:
Task.set_todo(task['id'])
else:
# Task was not done yet
def run(self):
while True:
schedule.run_pending()
time.sleep(self.interval)