How do I get a Cron like scheduler in Python? [closed] How do I get a Cron like scheduler in Python? [closed] python python

How do I get a Cron like scheduler in Python? [closed]


If you're looking for something lightweight checkout schedule:

import scheduleimport timedef 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.


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, timedeltaimport time# Some utility classes / functions firstclass AllMatch(set):    """Universal set - match everything"""    def __contains__(self, item): return TrueallMatch = 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 classclass 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.


More or less same as above but concurrent using gevent :)

"""Gevent based crontab implementation"""from datetime import datetime, timedeltaimport gevent# Some utility classes / functions firstdef 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 objclass AllMatch(set):    """Universal set - match everything"""    def __contains__(self, item):         return TrueallMatch = 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()


matomo