Python: Running a function periodically with asyncio

Updated: July 25, 2023 By: Goodman Post a comment

When developing applications with Python, there might be cases where you need to execute a function periodically, such as:

  • Performing a recurring task, such as checking the status of a server, sending a notification, updating a database, backing up files, etc.
  • Implementing a scheduler, a cron job, or a daemon process that runs in the background.

This concise, straight-to-the-point article will show you how to run a function repeatedly at regular intervals by making use of the asyncio module and the while loop. No third-party library is required.

What is the point?

The steps to get the job done are:

  1. Import the asyncio module.
  2. Define the function (asynchronous or synchronous) that you want to run periodically.
  3. Define an async function that wraps the function you’ve created in the previous step in a while loop and calls asyncio.sleep() with the desired interval between each invocation.
  4. Create an asyncio event loop and run the wrapper function until it completes or is canceled.

For more clarity, see the complete example below.

Example

This example simply prints the current time every 10 seconds:

import asyncio
from datetime import datetime

# Define a function that prints the current time
def print_time():
    print(f"Current time: {datetime.now().isoformat()}")

# Define another async function that runs print_time every 10 seconds
async def run_periodically():
    while True:
        # Call print_time
        print_time()
        # Sleep for 10 seconds
        await asyncio.sleep(10)

# Create an event loop
loop = asyncio.get_event_loop()
# Run the wrapper function until it completes or is cancelled
loop.run_until_complete(run_periodically())

The output will look like this:

Current time: 2023-07-24T23:16:19.241766
Current time: 2023-07-24T23:16:29.243151
Current time: 2023-07-24T23:16:39.244518
Current time: 2023-07-24T23:16:49.245967
Current time: 2023-07-24T23:16:59.247547
Current time: 2023-07-24T23:17:09.249159
Current time: 2023-07-24T23:17:19.250752
Current time: 2023-07-24T23:17:29.252354
Current time: 2023-07-24T23:17:39.252785
Current time: 2023-07-24T23:17:49.253983

This approach is not perfectly precise, as the actual interval between each invocation may vary depending on the system load and the execution time of the print_time() function. However, in most cases, the level of deviation is only a few microseconds and so is acceptable.

Stopping the loop after N times

You can modify or enhance the example above a little bit to suit your needs. For instance, if you only want to call the print_time() function a certain number of times (N) instead of triggering it perpetually, you can do the following:

import asyncio
from datetime import datetime

# Define a function that prints the current time
def print_time():
    print(f"Current time: {datetime.now().isoformat()}")

# Define another async function that runs print_time every 10 seconds
async def run_periodically():
    N = 5 # The print_time() function will run 5 times
    i = 0
    while i < N:
        # Call print_time
        print_time()
        # Sleep for 10 seconds
        await asyncio.sleep(10)
        i += 1

# Create an event loop
loop = asyncio.get_event_loop()
# Run the wrapper function until it completes or is cancelled
loop.run_until_complete(run_periodically())

You can also use the try/catch statement to handle exceptions as necessary. See also Handling Exceptions with Try/Except/Else/Finally in Python.