Explain Codes LogoExplain Codes Logo

How to repeatedly execute a function every x seconds?

python
scheduling
asyncio
threading
Anton ShumikhinbyAnton Shumikhin·Aug 8, 2024
TLDR

For regular function execution in Python, threading.Timer is an efficient tool. It activates a function after a specified interval and can be recursive. The below snippet is the essence of this idea:

from threading import Timer def repeater(interval, function): Timer(interval, repeater, [interval, function]).start() function() # Example: Run `my_task` every 2 seconds. def my_task(): print("Task accomplished") # Now let's grab a coffee! repeater(2, my_task)

The code above schedules my_task to initiate every 2 seconds by recursively calling repeater.

Advanced scheduling methods: Choosing the right tool

Scheduling a function to activate at regular intervals is an everyday task. While threading.Timer provides a simple way to do it, Python has more methods at hand for different complex scenarios and requirements.

Event scheduling with sched

import sched, time # Initialize scheduler with a monotonic time source for accurate timings. scheduler = sched.scheduler(time.monotonic, time.sleep) def scheduled_job(): print('Function initiated.') def scheduler_tool(exec_interval, func, scheduled_event=None): if scheduled_event: scheduler.cancel(scheduled_event) new_event = scheduler.enter(exec_interval, 1, func) scheduler.run(blocking=False) return new_event # Schedule `scheduled_job` every 5 seconds. current_event = scheduler_tool(5, scheduled_job)

A monotonic clock source ensures schedule remains accurate over elapse time. It won't even mind if you change your system clock!

AsyncIO: A multitasker's delight

For convoluted I/O-bound or network tasks, asyncio is Python's inbuilt library to write concurrent code using the async/await syntax.

import asyncio async def periodic_operation(interval): while True: await asyncio.sleep(interval) print("Periodic operation is on!") # Victory dance time. # Run the periodic job every 3 seconds loop = asyncio.get_event_loop() loop.run_until_complete(periodic_operation(3))

Avoiding common pitfalls

There are some important factors to consider when implementing periodic execution which might affect the reliability and effectiveness of your method.

bye-bye while time.sleep() loop

It might be easy to just use a simple while loop with time.sleep(interval), but this method can cause drift over time due to the execution time of the task itself.

When a tortoise runs your functions

If your function's execution time may vary and potentially exceed your interval, consider using threading to ensure intervals remain constant:

import threading def run_at_intervals(interval, function): next_call = time.time() while True: threading.Thread(target=function).start() next_call = next_call + interval time.sleep(max(0, next_call - time.time())) # Schedule `long_running_function` to start every 10 seconds def long_running_function(): # Time-intensive computations here print("Long-running function is a ninja!") # Faster than Flash! run_at_intervals(10, long_running_function)

Dr. Strange New Precision with Twisted's Reactor Pattern

For high precision and robust logic, including error handling and daemonization, consider using Twisted's LoopingCall:

from twisted.internet import task, reactor def periodic_execution(): print("Twisted: Periodic execution in action!") interval = 6.0 # seconds lc = task.LoopingCall(periodic_execution) lc.start(interval) reactor.run()

Twisted's LoopingCall will make sure your tasks activate at correct intervals, accounting for function's execution time.

Make way for on-the-go adjustments

Using classes like RepeatedTimer can provide real time adjustments such as modifying intervals, and ensuring that your function calls do not overlap if they take longer than the original interval.

Keeping it simple, yet refreshing

While simplicity is appealing, always stick with pure Python solutions that do not require additional dependencies. But for advanced scheduling needs, exploring third-party libraries like APScheduler or even Celery for distributed tasks might offer the just-right balance between ease of use and functionality.