Coverage for other/scheduler.py : 58%

from datetime import date, datetime, timedelta 

import os 

import pickle 

""" 

A collection of functions for quickly registering periodic tasks using the 

decorator syntax. Note that, while these functions are used as decorators, they 

actually return the original function, so you can decorate multiple times 

without worry. 

""" 

def _create_decorator(schedule): 

""" 

Creates a decorator that registers periodic tasks. 

""" 

def decorator(fn): 

_registry.register_task(fn, schedule) 

return fn 

return decorator 

def hourly(): 

return _create_decorator(lambda last_run: last_run < datetime.now().replace(second=0, microsecond=0) - timedelta( 

minutes=datetime.now().minute)) 

def daily(): 

29 ↛ exitline 29 didn't run the lambda on line 29 return _create_decorator(lambda last_run: last_run.date() < date.today()) 

def weekly(): 

return _create_decorator(lambda last_run: last_run.date() < date.today() - timedelta(date.today().weekday())) 

def monthly(): 

37 ↛ exitline 37 didn't run the lambda on line 37 return _create_decorator(lambda last_run: last_run.date() <= date.today() - timedelta(date.today().day)) 

def every(days=0, seconds=0, microseconds=0, milliseconds=0, minutes=0, hours=0, weeks=0): 

period = timedelta(days, seconds, microseconds, milliseconds, minutes, hours, weeks) 

return _create_decorator(lambda last_run: last_run <= datetime.now() - period) 

class TaskRegistry: 

class Task: 

timestamp_folder = os.path.dirname('./.timestamps/') 

def __init__(self, fn, schedule): 

self.schedule = schedule 

self.fn = fn 

self.task_id = '{}.{}'.format(fn.__module__, fn.__name__) 

54 ↛ 55line 54 didn't jump to line 55, because the condition on line 54 was never true if not os.path.isdir(self.timestamp_folder): 

os.mkdir(self.timestamp_folder) 

try: 

with open(os.path.join(self.timestamp_folder, self.task_id), 'rb') as timestamp_file: 

self.last_run = pickle.load(timestamp_file) 

except FileNotFoundError: 

self.last_run = datetime.min # Closest approximation to "never" 

def run(self): 

if self.schedule(self.last_run): 

self.fn() 

self.last_run = datetime.now() 

with open(os.path.join(self.timestamp_folder, self.task_id), 'wb') as timestamp_file: 

pickle.dump(self.last_run, timestamp_file) 

def __init__(self): 

self.tasks = [] 

def register_task(self, fn, schedule): 

self.tasks.append(self.Task(fn, schedule)) 

def run_all_tasks(self): 

for task in self.tasks: 

task.run() 

def run_all_tasks(): 

_registry.run_all_tasks() 

_registry = TaskRegistry()