# How do I get a Cron like scheduler in Python?

## The Question :

383 people think this question is useful

I’m looking for a library in Python which will provide at and cron like functionality.

I’d quite like have a pure Python solution, rather than relying on tools installed on the box; this way I run on machines with no cron.

For those unfamiliar with cron: you can schedule tasks based upon an expression like:

 0 2 * * 7 /usr/bin/run-backup # run the backups at 0200 on Every Sunday
0 9-17/2 * * 1-5 /usr/bin/purge-temps # run the purge temps command, every 2 hours between 9am and 5pm on Mondays to Fridays.



The cron time expression syntax is less important, but I would like to have something with this sort of flexibility.

If there isn’t something that does this for me out-the-box, any suggestions for the building blocks to make something like this would be gratefully received.

Edit I’m not interested in launching processes, just “jobs” also written in Python – python functions. By necessity I think this would be a different thread, but not in a different process.

To this end, I’m looking for the expressivity of the cron time expression, but in Python.

Cron has been around for years, but I’m trying to be as portable as possible. I cannot rely on its presence.

• I also would like to know how to do this. It would be more useful to have a cross platform solution than depend on platform specific components.
• This is not off-topic, this is a very important and useful question
• I propose to use celery. It’s a task queue with focus on real-time processing, while also supporting task scheduling. You can have a look at how we can handle the periodic tasks PS: I can’t give an example as this question is not accepting answers, I agree with @Connor

634 people think this answer is useful

If you’re looking for something lightweight checkout schedule:

import schedule
import time

def job():
print("I'm working...")

schedule.every(10).minutes.do(job)
schedule.every().hour.do(job)
schedule.every().day.at("10:30").do(job)

while 1:
schedule.run_pending()
time.sleep(1)



Disclosure: I’m the author of that library.

68 people think this answer is useful

You could just use normal Python argument passing syntax to specify your crontab. For example, suppose we define an Event class as below:

from datetime import datetime, timedelta
import time

# Some utility classes / functions first
class AllMatch(set):
"""Universal set - match everything"""
def __contains__(self, item): return True

allMatch = AllMatch()

def conv_to_set(obj):  # Allow single integer to be provided
if isinstance(obj, (int,long)):
return set([obj])  # Single item
if not isinstance(obj, set):
obj = set(obj)
return obj

# The actual Event class
class Event(object):
def __init__(self, action, min=allMatch, hour=allMatch,
day=allMatch, month=allMatch, dow=allMatch,
args=(), kwargs={}):
self.mins = conv_to_set(min)
self.hours= conv_to_set(hour)
self.days = conv_to_set(day)
self.months = conv_to_set(month)
self.dow = conv_to_set(dow)
self.action = action
self.args = args
self.kwargs = kwargs

def matchtime(self, t):
"""Return True if this event should trigger at the specified datetime"""
return ((t.minute     in self.mins) and
(t.hour       in self.hours) and
(t.day        in self.days) and
(t.month      in self.months) and
(t.weekday()  in self.dow))

def check(self, t):
if self.matchtime(t):
self.action(*self.args, **self.kwargs)



(Note: Not thoroughly tested)

Then your CronTab can be specified in normal python syntax as:

c = CronTab(
Event(perform_backup, 0, 2, dow=6 ),
Event(purge_temps, 0, range(9,18,2), dow=range(0,5))
)



This way you get the full power of Python’s argument mechanics (mixing positional and keyword args, and can use symbolic names for names of weeks and months)

The CronTab class would be defined as simply sleeping in minute increments, and calling check() on each event. (There are probably some subtleties with daylight savings time / timezones to be wary of though). Here’s a quick implementation:

class CronTab(object):
def __init__(self, *events):
self.events = events

def run(self):
t=datetime(*datetime.now().timetuple()[:5])
while 1:
for e in self.events:
e.check(t)

t += timedelta(minutes=1)
while datetime.now() < t:
time.sleep((t - datetime.now()).seconds)



A few things to note: Python’s weekdays / months are zero indexed (unlike cron), and that range excludes the last element, hence syntax like “1-5” becomes range(0,5) – ie [0,1,2,3,4]. If you prefer cron syntax, parsing it shouldn’t be too difficult however.

14 people think this answer is useful

More or less same as above but concurrent using gevent 🙂

"""Gevent based crontab implementation"""

from datetime import datetime, timedelta
import gevent

# Some utility classes / functions first
def conv_to_set(obj):
"""Converts to set allowing single integer to be provided"""

if isinstance(obj, (int, long)):
return set([obj])  # Single item
if not isinstance(obj, set):
obj = set(obj)
return obj

class AllMatch(set):
"""Universal set - match everything"""
def __contains__(self, item):
return True

allMatch = AllMatch()

class Event(object):
"""The Actual Event Class"""

def __init__(self, action, minute=allMatch, hour=allMatch,
day=allMatch, month=allMatch, daysofweek=allMatch,
args=(), kwargs={}):
self.mins = conv_to_set(minute)
self.hours = conv_to_set(hour)
self.days = conv_to_set(day)
self.months = conv_to_set(month)
self.daysofweek = conv_to_set(daysofweek)
self.action = action
self.args = args
self.kwargs = kwargs

def matchtime(self, t1):
"""Return True if this event should trigger at the specified datetime"""
return ((t1.minute     in self.mins) and
(t1.hour       in self.hours) and
(t1.day        in self.days) and
(t1.month      in self.months) and
(t1.weekday()  in self.daysofweek))

def check(self, t):
"""Check and run action if needed"""

if self.matchtime(t):
self.action(*self.args, **self.kwargs)

class CronTab(object):
"""The crontab implementation"""

def __init__(self, *events):
self.events = events

def _check(self):
"""Check all events in separate greenlets"""

t1 = datetime(*datetime.now().timetuple()[:5])
for event in self.events:
gevent.spawn(event.check, t1)

t1 += timedelta(minutes=1)
s1 = (t1 - datetime.now()).seconds + 1
print "Checking again in %s seconds" % s1
job = gevent.spawn_later(s1, self._check)

def run(self):
"""Run the cron forever"""

self._check()
while True:
gevent.sleep(60)

import os
"""Just an example that sends a bell and asd to all terminals"""

os.system('echo asd | wall')

cron = CronTab(
)
cron.run()



11 people think this answer is useful

None of the listed solutions even attempt to parse a complex cron schedule string. So, here is my version, using croniter. Basic gist:

schedule = "*/5 * * * *" # Run every five minutes

nextRunTime = getNextCronRunTime(schedule)
while True:
roundedDownTime = roundDownTime()
if (roundedDownTime == nextRunTime):
####################################
### Do your periodic thing here. ###
####################################
nextRunTime = getNextCronRunTime(schedule)
elif (roundedDownTime > nextRunTime):
# We missed an execution. Error. Re initialize.
nextRunTime = getNextCronRunTime(schedule)
sleepTillTopOfNextMinute()



Helper routines:

from croniter import croniter
from datetime import datetime, timedelta

# Round time down to the top of the previous minute
def roundDownTime(dt=None, dateDelta=timedelta(minutes=1)):
roundTo = dateDelta.total_seconds()
if dt == None : dt = datetime.now()
seconds = (dt - dt.min).seconds
rounding = (seconds+roundTo/2) // roundTo * roundTo
return dt + timedelta(0,rounding-seconds,-dt.microsecond)

# Get next run time from now, based on schedule specified by cron string
def getNextCronRunTime(schedule):
return croniter(schedule, datetime.now()).get_next(datetime)

# Sleep till the top of the next minute
def sleepTillTopOfNextMinute():
t = datetime.utcnow()
sleeptime = 60 - (t.second + t.microsecond/1000000.0)
time.sleep(sleeptime)



5 people think this answer is useful

I know there are a lot of answers, but another solution could be to go with decorators. This is an example to repeat a function everyday at a specific time. The cool think about using this way is that you only need to add the Syntactic Sugar to the function you want to schedule:

@repeatEveryDay(hour=6, minutes=30)
def sayHello(name):
print(f"Hello {name}")

sayHello("Bob") # Now this function will be invoked every day at 6.30 a.m



And the decorator will look like:

def repeatEveryDay(hour, minutes=0, seconds=0):
"""
Decorator that will run the decorated function everyday at that hour, minutes and seconds.
:param hour: 0-24
:param minutes: 0-60 (Optional)
:param seconds: 0-60 (Optional)
"""
def decoratorRepeat(func):

@functools.wraps(func)
def wrapperRepeat(*args, **kwargs):

def getLocalTime():
return datetime.datetime.fromtimestamp(time.mktime(time.localtime()))

# Get the datetime of the first function call
td = datetime.timedelta(seconds=15)
if wrapperRepeat.nextSent == None:
now = getLocalTime()
wrapperRepeat.nextSent = datetime.datetime(now.year, now.month, now.day, hour, minutes, seconds)
if wrapperRepeat.nextSent < now:
wrapperRepeat.nextSent += td

# Waiting till next day
while getLocalTime() < wrapperRepeat.nextSent:
time.sleep(1)

# Call the function
func(*args, **kwargs)

# Get the datetime of the next function call
wrapperRepeat.nextSent += td
wrapperRepeat(*args, **kwargs)

wrapperRepeat.nextSent = None
return wrapperRepeat

return decoratorRepeat



4 people think this answer is useful

There isn’t a “pure python” way to do this because some other process would have to launch python in order to run your solution. Every platform will have one or twenty different ways to launch processes and monitor their progress. On unix platforms, cron is the old standard. On Mac OS X there is also launchd, which combines cron-like launching with watchdog functionality that can keep your process alive if that’s what you want. Once python is running, then you can use the sched module to schedule tasks.

2 people think this answer is useful

Another trivial solution would be:

from aqcron import At
from time import sleep
from datetime import datetime

# Event scheduling
event_1 = At( second=5 )
event_2 = At( second=[0,20,40] )

while True:
now = datetime.now()

# Event check
if now in event_1: print "event_1"
if now in event_2: print "event_2"

sleep(1)



And the class aqcron.At is:

# aqcron.py

class At(object):
def __init__(self, year=None,    month=None,
day=None,     weekday=None,
hour=None,    minute=None,
second=None):
loc = locals()
loc.pop("self")
self.at = dict((k, v) for k, v in loc.iteritems() if v != None)

def __contains__(self, now):
for k in self.at.keys():
try:
if not getattr(now, k) in self.at[k]: return False
except TypeError:
if self.at[k] != getattr(now, k): return False
return True



1 people think this answer is useful

I like how the pycron package solves this problem.

import pycron
import time

while True:
if pycron.is_now('0 2 * * 0'):   # True Every Sunday at 02:00
print('running backup')
time.sleep(60)               # The process should take at least 60 sec
# to avoid running twice in one minute
else:
time.sleep(15)               # Check again in 15 seconds