Schedule a repeating event in Python 3

Question

I'm trying to schedule a repeating event to run every minute in Python 3.

I've seen class sched.scheduler but I'm wondering if there's another way to do it. I've heard mentions I could use multiple threads for this, which I wouldn't mind doing.

I'm basically requesting some JSON and then parsing it; its value changes over time.

To use sched.scheduler I have to create a loop to request it to schedule the even to run for one hour:

scheduler = sched.scheduler(time.time, time.sleep)

# Schedule the event. THIS IS UGLY!
for i in range(60):
    scheduler.enter(3600 * i, 1, query_rate_limit, ())

scheduler.run()

What other ways to do this are there?

Answer 1

I figured out a mimic for setTimeout and setInterval from JavaScript, but in Python.

With this class you can:

1. Set a timeout with an interval ID (through variable, not required).
2. Set an interval in the same wise as a timeout.
3. Set an interval with an ID, and then set a timeout that will cancel the interval in a given amount of time.
4. Set a timeout with an ID and cancel the timeout with another setTimeout.

I know it seems complex but I needed all the functionality it had just like in JavaScript. Note that you will need to pre-create a function and use the function name as a parameter (see code).

#If I'm not mistaken, I believe the only modules you need are time and threading. The rest were for something I'm working on.
import time
import math
import pygame
import threading
import ctypes
from sys import exit

#Functions
class Timer:
    def __init__(self):
        self.timers = {}
        self.timer_id = 0
    
    def setInterval(self, fn, time, *args):
        def interval_callback():
            fn(*args)
            if timer_id in self.timers:
                self.timers[timer_id] = threading.Timer(time/1000, interval_callback)
                self.timers[timer_id].start()

        timer_id = self.timer_id
        self.timer_id += 1
        self.timers[timer_id] = threading.Timer(time/1000, interval_callback)
        self.timers[timer_id].start()
        return timer_id

    def clearInterval(self, timer_id):
        if timer_id in self.timers:
            self.timers[timer_id].cancel()
            del self.timers[timer_id]

    def setTimeout(self, fn, delay, *args, **kwargs):
        def timer_callback():
            self.timers.pop(timer_id, None)
            fn(*args, **kwargs)
        
        timer_id = self.timer_id
        self.timer_id += 1
        t = threading.Timer(delay / 1000, timer_callback)
        self.timers[timer_id] = t
        t.start()
        return timer_id
    
    def clearTimeout(self, timer_id):
        t = self.timers.pop(timer_id, None)
        if t is not None:
            t.cancel()
t = Timer()

#Usage Example
lall = t.setInterval(print, 1000, "hi")
t.setTimeout(t.clearInterval, 3000, lall)

Hope this helps.

Answer 2

Based on MestreLion answer, it solve a little problem with multithreading:

from threading import Timer, Lock

class Periodic(object):
    """
    A periodic task running in threading.Timers
    """

    def __init__(self, interval, function, *args, **kwargs):
        self._lock = Lock()
        self._timer = None
        self.function = function
        self.interval = interval
        self.args = args
        self.kwargs = kwargs
        self._stopped = True
        if kwargs.pop('autostart', True):
            self.start()

    def start(self, from_run=False):
        self._lock.acquire()
        if from_run or self._stopped:
            self._stopped = False
            self._timer = Timer(self.interval, self._run)
            self._timer.start()
        self._lock.release()

    def _run(self):
        self.start(from_run=True)
        self.function(*self.args, **self.kwargs)

    def stop(self):
        self._lock.acquire()
        self._stopped = True
        self._timer.cancel()
        self._lock.release()

Answer 3

Taking the original threading.Timer() class implementation and fixing the run() method I get something like:

class PeriodicTimer(Thread):
    """A periodic timer that runs indefinitely until cancel() is called."""
    def __init__(self, interval, function, args=None, kwargs=None):
        Thread.__init__(self)
        self.interval = interval
        self.function = function
        self.args = args if args is not None else []
        self.kwargs = kwargs if kwargs is not None else {}
        self.finished = Event()

    def cancel(self):
        """Stop the timer if it hasn't finished yet."""
        self.finished.set()

    def run(self):
        """Run until canceled"""
        while not self.finished.wait(self.interval):
            self.function(*self.args, **self.kwargs)

The wait() method called is using a condition variable, so it should be rather efficient.

Answer 4

Doc: Advanced Python Scheduler

@sched.cron_schedule(day='last sun')
def some_decorated_task():
    print("I am printed at 00:00:00 on the last Sunday of every month!")

Available fields:

| Field       | Description                                                    |
|-------------|----------------------------------------------------------------|
| year        | 4-digit year number                                            |
| month       | month number (1-12)                                            |
| day         | day of the month (1-31)                                        |
| week        | ISO week number (1-53)                                         |
| day_of_week | number or name of weekday (0-6 or mon,tue,wed,thu,fri,sat,sun) |
| hour        | hour (0-23)                                                    |
| minute      | minute (0-59)                                                  |
| second      | second (0-59)                                                  |

Answer 5

I ran into a similar issue a while back so I made a python module event-scheduler to address this. It has a very similar API to the sched library with a few differences:

1. It utilizes a background thread and is always able to accept and run jobs in the background until the scheduler is stopped explicitly (no need for a while loop).
2. It comes with an API to schedule recurring events at a user specified interval until explicitly cancelled.

It can be installed by pip install event-scheduler

from event_scheduler import EventScheduler

event_scheduler = EventScheduler()
event_scheduler.start()
# Schedule the recurring event to print "hello world" every 60 seconds with priority 1
# You can use the event_id to cancel the recurring event later
event_id = event_scheduler.enter_recurring(60, 1, print, ("hello world",))

Answer 6

There is a new package, called ischedule. For this case, the solution could be as following:

from ischedule import schedule, run_loop
from datetime import timedelta


def query_rate_limit():
    print("query_rate_limit")

schedule(query_rate_limit, interval=60)
run_loop(return_after=timedelta(hours=1))

Everything runs on the main thread and there is no busy waiting inside the run_loop. The startup time is very precise, usually within a fraction of a millisecond of the specified time.

Answer 7

You could use threading.Timer, but that also schedules a one-off event, similarly to the .enter method of scheduler objects.

The normal pattern (in any language) to transform a one-off scheduler into a periodic scheduler is to have each event re-schedule itself at the specified interval. For example, with sched, I would not use a loop like you're doing, but rather something like:

def periodic(scheduler, interval, action, actionargs=()):
    scheduler.enter(interval, 1, periodic,
                    (scheduler, interval, action, actionargs))
    action(*actionargs)

and initiate the whole "forever periodic schedule" with a call

periodic(scheduler, 3600, query_rate_limit)

Or, I could use threading.Timer instead of scheduler.enter, but the pattern's quite similar.

If you need a more refined variation (e.g., stop the periodic rescheduling at a given time or upon certain conditions), that's not too hard to accomodate with a few extra parameters.

Answer 8

Here's a quick and dirty non-blocking loop with Thread:

#!/usr/bin/env python3
import threading,time

def worker():
    print(time.time())
    time.sleep(5)
    t = threading.Thread(target=worker)
    t.start()


threads = []
t = threading.Thread(target=worker)
threads.append(t)
t.start()
time.sleep(7)
print("Hello World")

There's nothing particularly special, the worker creates a new thread of itself with a delay. Might not be most efficient, but simple enough. northtree's answer would be the way to go if you need more sophisticated solution.

And based on this, we can do the same, just with Timer:

#!/usr/bin/env python3
import threading,time

def hello():
    t = threading.Timer(10.0, hello)
    t.start()
    print( "hello, world",time.time() )

t = threading.Timer(10.0, hello)
t.start()
time.sleep(12)
print("Oh,hai",time.time())
time.sleep(4)
print("How's it going?",time.time())

Answer 9

See my sample

import sched, time

def myTask(m,n):
  print n+' '+m

def periodic_queue(interval,func,args=(),priority=1):
  s = sched.scheduler(time.time, time.sleep)
  periodic_task(s,interval,func,args,priority)
  s.run()

def periodic_task(scheduler,interval,func,args,priority):
  func(*args)
  scheduler.enter(interval,priority,periodic_task,
                   (scheduler,interval,func,args,priority))

periodic_queue(1,myTask,('world','hello'))

Answer 10

Based on Alex Martelli's answer, I have implemented decorator version which is more easier to integrated.

import sched
import time
import datetime
from functools import wraps
from threading import Thread


def async(func):
    @wraps(func)
    def async_func(*args, **kwargs):
        func_hl = Thread(target=func, args=args, kwargs=kwargs)
        func_hl.start()
        return func_hl
    return async_func


def schedule(interval):
    def decorator(func):
        def periodic(scheduler, interval, action, actionargs=()):
            scheduler.enter(interval, 1, periodic,
                            (scheduler, interval, action, actionargs))
            action(*actionargs)

        @wraps(func)
        def wrap(*args, **kwargs):
            scheduler = sched.scheduler(time.time, time.sleep)
            periodic(scheduler, interval, func)
            scheduler.run()
        return wrap
    return decorator


@async
@schedule(1)
def periodic_event():
    print(datetime.datetime.now())


if __name__ == '__main__':
    print('start')
    periodic_event()
    print('end')

Answer 11

You could use schedule. It works on Python 2.7 and 3.3 and is rather lightweight:

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)

Answer 12

Use Celery.

from celery.task import PeriodicTask
from datetime import timedelta


class ProcessClicksTask(PeriodicTask):
    run_every = timedelta(minutes=30)

    def run(self, **kwargs):
        #do something

Answer 13

You could use the Advanced Python Scheduler. It even has a cron-like interface.

Answer 14

My humble take on the subject:

from threading import Timer

class RepeatedTimer(object):
    def __init__(self, interval, function, *args, **kwargs):
        self._timer     = None
        self.function   = function
        self.interval   = interval
        self.args       = args
        self.kwargs     = kwargs
        self.is_running = False
        self.start()

    def _run(self):
        self.is_running = False
        self.start()
        self.function(*self.args, **self.kwargs)

    def start(self):
        if not self.is_running:
            self._timer = Timer(self.interval, self._run)
            self._timer.start()
            self.is_running = True

    def stop(self):
        self._timer.cancel()
        self.is_running = False

Usage:

from time import sleep

def hello(name):
    print "Hello %s!" % name

print "starting..."
rt = RepeatedTimer(1, hello, "World") # it auto-starts, no need of rt.start()
try:
    sleep(5) # your long-running job goes here...
finally:
    rt.stop() # better in a try/finally block to make sure the program ends!

Features:

• Standard library only, no external dependencies
• Uses the pattern suggested by Alex Martnelli
• start() and stop() are safe to call multiple times even if the timer has already started/stopped
• function to be called can have positional and named arguments
• You can change interval anytime, it will be effective after next run. Same for args, kwargs and even function!