How to Run a Periodic Background Task in Python
You can run a periodic task in the background via a daemon thread.
In this tutorial you will discover how to run a periodic task in the background in Python.
Need for a Periodic Background Task
A thread is a thread of execution in a computer program.
Every Python program has at least one thread of execution called the main thread. Both processes and threads are created and managed by the underlying operating system.
Sometimes we may need to create additional threads in our program in order to execute code concurrently.
Python provides the ability to create and manage new threads via the threading module and the threading.Thread class.
You can learn more about Python threads in the guude:
In concurrent programming, we may need to run a task periodically in the background.
This may be for many reasons, such as:
- Reloading or refreshing data in the application.
- Storing or saving data from the application.
- Reporting or logging progress of the application.
Periodically means that the task is repeated after a consistent time interval, such as:
Background means that the task is not the main purpose of the application, but instead is a task that supports the main application.
How can we run a periodic task in the background in Python?
Run your loops using all CPUs, download my FREE book to learn how.
How to Create a Periodic Background Task
You can run a periodic task in the background using a daemon thread with a while loop and a sleep.
A daemon thread is a background thread. Daemon is pronounced “dee-mon“, like the alternate spelling “demon“.
A thread may be configured to be a daemon or not, and most threads in concurrent programming, including the main thread, are non-daemon threads (not background threads) by default. The difference between daemon threads and non-daemon threads is that the process will exit if only daemon threads are running, whereas it cannot exit if at least one non-daemon thread is running.
As such, daemon threads are helpful for executing tasks in the background to support the non-daemon threads in an application.
If you are new to daemon threads, you can learn more about them here:
A Python threading.Thread instance can be configured to be a daemon thread.
We can configure a new thread to be a daemon thread by specifying the “daemon” argument to True in the constructor of the threading.Thread class.
How can I periodically execute a function with asyncio in Python
When you feel that something should happen «in background» of your asyncio program, asyncio.Task might be good way to do it. You can read this post to see how to work with tasks.
Here’s possible implementation of class that executes some function periodically:
import asyncio from contextlib import suppress class Periodic: def __init__(self, func, time): self.func = func self.time = time self.is_started = False self._task = None async def start(self): if not self.is_started: self.is_started = True # Start task to call func periodically: self._task = asyncio.ensure_future(self._run()) async def stop(self): if self.is_started: self.is_started = False # Stop task and await it stopped: self._task.cancel() with suppress(asyncio.CancelledError): await self._task async def _run(self): while True: await asyncio.sleep(self.time) self.func() async def main(): p = Periodic(lambda: print('test'), 1) try: print('Start') await p.start() await asyncio.sleep(3.1) print('Stop') await p.stop() await asyncio.sleep(3.1) print('Start') await p.start() await asyncio.sleep(3.1) finally: await p.stop() # we should stop task finally if __name__ == '__main__': loop = asyncio.get_event_loop() loop.run_until_complete(main()) Start test test test Stop Start test test test [Finished in 9.5s] As you see on start we just start task that calls some functions and sleeps some time in endless loop. On stop we just cancel that task. Note, that task should be stopped at the moment program finished.
One more important thing that your callback shouldn’t take much time to be executed (or it’ll freeze your event loop). If you’re planning to call some long-running func , you possibly would need to run it in executor .
There is no built-in support for periodic calls, no.
Just create your own scheduler loop that sleeps and executes any tasks scheduled:
import math, time async def scheduler(): while True: # sleep until the next whole second now = time.time() await asyncio.sleep(math.ceil(now) - now) # execute any scheduled tasks async for task in scheduled_tasks(time.time()): await task() The scheduled_tasks() iterator should produce tasks that are ready to be run at the given time. Note that producing the schedule and kicking off all the tasks could in theory take longer than 1 second; the idea here is that the scheduler yields all tasks that should have started since the last check.
A variant that may be helpful: if you want your recurring call to happen every n seconds instead of n seconds between the end of the last execution and the beginning of the next, and you don’t want calls to overlap in time, the following is simpler:
async def repeat(interval, func, *args, **kwargs): """Run func every interval seconds. If func has not finished before *interval*, will run again immediately when the previous iteration finished. *args and **kwargs are passed as the arguments to func. """ while True: await asyncio.gather( func(*args, **kwargs), asyncio.sleep(interval), ) And an example of using it to run a couple tasks in the background:
async def f(): await asyncio.sleep(1) print('Hello') async def g(): await asyncio.sleep(0.5) print('Goodbye') async def main(): t1 = asyncio.ensure_future(repeat(3, f)) t2 = asyncio.ensure_future(repeat(2, g)) await t1 await t2 loop = asyncio.get_event_loop() loop.run_until_complete(main()) Alternative version with decorator for python 3.7
import asyncio import time def periodic(period): def scheduler(fcn): async def wrapper(*args, **kwargs): while True: asyncio.create_task(fcn(*args, **kwargs)) await asyncio.sleep(period) return wrapper return scheduler @periodic(2) async def do_something(*args, **kwargs): await asyncio.sleep(5) # Do some heavy calculation print(time.time()) if __name__ == '__main__': asyncio.run(do_something('Maluzinha do papai!', secret=42)) Based on @A. Jesse Jiryu Davis answer (with @Torkel Bjørnson-Langen and @ReWrite comments) this is an improvement which avoids drift.
import time import asyncio @asyncio.coroutine def periodic(period): def g_tick(): t = time.time() count = 0 while True: count += 1 yield max(t + count * period - time.time(), 0) g = g_tick() while True: print('periodic', time.time()) yield from asyncio.sleep(next(g)) loop = asyncio.get_event_loop() task = loop.create_task(periodic(1)) loop.call_later(5, task.cancel) try: loop.run_until_complete(task) except asyncio.CancelledError: pass