How do I get a Cron like scheduler in Python?

Question

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.

Diantha

Answer 1

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.

Threecornered

You should mention that you're the maintainer of schedule. It worked well for me. It would be even nicer if it had cron like syntax and supported decorators (see crython but don't use this library because it doesn't work; the scheduling doesn't seem to be written well). – Fascicle

Is there a way to pass a parameter to the job? I'd like to do something like this: schedule.every().hour.do(job(myParam)) – Rattoon

schedule.every().hour.do(job) does this run every clock hour? Like 01:00, 02:00, 03:00, etc.? even if the start time isn't a full hour? – Deoxyribose

suppose this code is in scheduler.py. will this code run automatically ? – Immoderation

Just sending a simple get request makes this drift by about 1 second every three iterations The second solution here works well #8600661 – Filiate

For @ZenSkunkworx and others still wondering about this. You might be able to make it into a lambda function to pass a function with parameter(s). Something like schedule.every().hour.do(lambda: job(myParam)) – Ser

@darrel-holt and @zen-skunkworx: The do() function forwards extra arguments you pass to it to the job function: schedule.readthedocs.io/en/stable/api.html#schedule.Job.do For example, you can do this: schedule.every().hour.do(job, param1, param2) No need to use a lambda. Hope this helps :) – Threecornered

@Threecornered Could you support schedule.run() just like sched instead of running on loop? Thanks. – Thoracoplasty

Adding schedule.clear() to your example above and in your repo would be helpful – Prevost

@Threecornered , any particular reason on why you included time.sleep ? – Mate

@Mate Usually adding a sleep(1) or even sleep(0.001) in a small infinite loop is done to prevent python from using 100% of a core of your CPU. – Hendiadys

@Threecornered I have a problem with this, i'm trying to schedule adiary task, but it's does not work – While

does it prevent conflicts and multiple instances? i want to run only 1 instance of the job at a time, the next run should skip/pause till the previous job will finish – Rosemare

Maybe am very new to this, but why is there an infinite loop and wouldn't that keep my thread engaged? Should I be running this scheduler on a separate thread? – Demiurge

It can't run jobs between 8-18 hours every 10 minutes. Cron can, therefore this is not the same functionality. – Surprint

Can not import schedule. Any issues with pip? – Milesmilesian

@Threecornered explain me one thing. If I have similar function for example I want to print some string every 1 minute. What's the difference if I simply do print("xxx") time.sleep(60) in while True: loop vs I will use your script. You use infinite loop to so it doesn't work same ? – Ronni

Can we use this for the run a python file? – Precipitin

bad without standard cron syntax – Moralez

What happens if this scheduler takes longer to run the tasks, than necessary to regain control and launch on-time the next batch of tasks? Does it internally run the tasks in new threads, to ensure immediate return of the control? Or do we have to run it in a separate thread ourselves? – Harper

@Threecornered Does this work even when you shut down / reboot your computer? If no, how could I get around that? – Seek

Will the scheduled py script run with the schedule library when the system is off? – Stagestruck

I want to define a scheduler in python version 3 that runs at every minute within a time range. I wrote below code- schedule.every(1).minutes.day.at("00:10").to.day.at("06:10").do(somthing)---------------- but it's not working. – Orv

I wish this library would stop from being recommended as it doesn't really work for non trivial things. It does not come close to replacing the cron syntax – Contortion

NOT___WORTH___CONSIDERING! @Threecornered doesn't maintain this code and it has bugs. DESPITE pull requests that change 26 lines of code to get it working. – Hanging

How can I run a script every 5 minutes in the time interval between 12:00 and 12:30 using the schedule module? – Hernia

@MichaelSchmidt Schedule is being actively maintained again. – Hillaryhillbilly

Answer 2

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.

Residuum

You might want to add some import statements for the inexperienced. I ended up putting all the classes in a single file with from datetime import * from time import sleep and changed time.sleep to sleep. Nice, simple elegant solution. Thanks. – Historical

Just wondering, why is this being preferred over Kronos? Is sched that buggy (since kronos uses sched)? Or is this just outdated? – Kerri

Thanks brian, i use your solution in production and it's working quite well. However, as others have pointed out, there is a subtle bug in your run code. Also i found it overly complicated for the needs. – Venetian

This is cool, but still doesn't support slash notation, for execution every hour, min, etc... – Amidase

Excellent idea to write your own classes, e.g. when I don't have sudo access on a server and thus cannot pip install anything :) – Resiniferous

@Resiniferous Consider using pip install --user and virtualenv. You shouldn't really pip install anything as root anyway. – Uthrop

@JonathonReinhart Good call on that - I agree not to do into root areas, but that privilege is needed sometimes to install into /usr/local/. When I can run things as my own user, that's always the best; when I need to set it up for the team at work to run... it becomes a bigger issue. ¯\_(ツ)_/¯ – Resiniferous

Answer 3

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 
def test_task():
    """Just an example that sends a bell and asd to all terminals"""

    os.system('echo asd | wall')  

cron = CronTab(
  Event(test_task, 22, 1 ),
  Event(test_task, 0, range(9,18,2), daysofweek=range(0,5)),
)
cron.run()

Gourmet

Just a note that datetime.timetuple() will start with year, month, day... etc... – Weekly

Nice.! but will it work for just one time task? – Heliogabalus

looks like you need gevent.joinall() to run the processes and also not sure why you have empty while loop in the run() – Heliogabalus

Answer 4

Croniter: Parse a CRON string and get sequence of next event times

• Documentation for Croniter can be found here

None of the listed solutions take the approach of parsing a CRON schedule string. So, here is my version, using croniter. Basic gist:

import croniter
import datetime

now = datetime.datetime.now()
iter = croniter.croniter('0 0 * * *', now)

next_datetime = croniter.get_next(datetime.datetime) # datetime not now!

remaining_seconds = next_datetime - now
remaining_seconds = remaining_seconds.total_seconds()
print(f'seconds until next event: {remaining_seconds}')

Easy!

---

More complex usecases with more detailed logic:

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)

Hamstring

How someone could go into the "missed an execution" elif? Atm I am using a schedule like "* * * * *" then adding some time.sleep greater than 1 minute in the "Do your periodic thing" if, but I always see the stuff in that if statement. When it takes more than 1 minute I just see the while loop skipping that missing loop execution. – Gregarine

@Gregarine The process could have been suspended, the clock could have been changed manually or by ntp etc. etc. Croniter is used in Airflow and it seems more full featured than the Crontab module and others. – Carbolated

What if there are multiple jobs to be scheduled ? Each having its own schedule_time, in that case how do we make sure we are not missing either of the executions ?? – Markitamarkka

Answer 5

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

Journalist

This is not a good idea, because your code "print('running backup')" will be launching entire minute with 5s interval. So in this case delay should be 60 seconds. – Herlindaherm

Answer 6

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

Operant

this is an attractive solution but it looks like it hinges upon using time.sleep(1) to do the "waiting" until the appropriate time; what kind of performance impact does this have? – Jana

@Jana It causes the entire program to pause for 1 second. It's better than pass though because pass results in higher CPU usage. – Lanthorn

Answer 7

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.

Broida

Answer 8

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

Lemur

Be careful when posting copy and paste boilerplate/verbatim answers to multiple questions, these tend to be flagged as "spammy" by the community. If you're doing this then it usually means the questions are duplicates so flag them as such instead: https://mcmap.net/q/76116/-python-event-schedular-like-quartz-closed – Disgraceful

Answer 9

I don't know if something like that already exists. It would be easy to write your own with time, datetime and/or calendar modules, see http://docs.python.org/library/time.html

The only concern for a python solution is that your job needs to be always running and possibly be automatically "resurrected" after a reboot, something for which you do need to rely on system dependent solutions.

Reinforce

Roll your own is an option - though the best code is code you don't have to write. Resurrection, I suppose is something I may need to consider. – Diantha