How to Run a Coroutine After a Delay in Python

January 1, 2023 Python Asyncio

You can run an asyncio coroutine after a delay by wrapping it in a coroutine that waits before executing the target coroutine.

In this tutorial, you will discover how to run a coroutine after a delay with asyncio in Python.

Let's get started.

Need to Run a Delayed Task

A coroutine can schedule another coroutine or task for execution.

The scheduled coroutine or task will execute as soon as it is able within the asyncio event loop.

This can be achieved by suspending and awaiting the coroutine directly, or wrapping the coroutine in a task and executing it independently.

When developing asyncio programs, we may need to schedule a coroutine and have it begin executing after a delay.

The high-level asyncio does not provide the capability to schedule a coroutine to execute later, after a delay.

How can we run a coroutine after a delay in asyncio?

How to Run a Delayed Task

We can develop a custom wrapper coroutine to execute a target coroutine after a delay.

The wrapper coroutine may take two arguments, a coroutine and a time in seconds.

It will sleep for the given delay interval in seconds, then await the provided coroutine.

The delay() coroutine below implements this.

# coroutine that will start another coroutine after a delay in seconds
async def delay(coro, seconds):
    # suspend for a time limit in seconds
    await asyncio.sleep(seconds)
    # execute the other coroutine
    await coro

To use the wrapper coroutine, a coroutine object can be created and either awaited directly or executed independently as a task.

For example, the caller may suspend and schedule the delayed coroutine and wait for it to be done:

...
# execute a coroutine after a delay
await delay(coro, 10)

Alternatively, the caller may schedule the delayed coroutine to run independently:

...
# execute a coroutine after a delay independently
_ = asyncio.create_task(delay(coro, 10))

Now that we know how to execute a coroutine after a delay, let's look at some worked examples.

Example of Running a Delayed Coroutine

We can explore how to execute a coroutine after a delay.

In this example, we will define a coroutine task that reports a message, simulates work with a sleep, then reports a final message.

The main coroutine will execute the task coroutine after a delay and directly wait for it to complete.

The complete example is listed below.

# SuperFastPython.com
# example of starting a coroutine after a delay
import asyncio

# coroutine that will start another coroutine after a delay in seconds
async def delay(coro, seconds):
    # suspend for a time limit in seconds
    await asyncio.sleep(seconds)
    # execute the other coroutine
    await coro

# task coroutine to do something
async def task_coro():
    # report a message
    print('Coroutine is running')
    # simulate some long running task
    await asyncio.sleep(3)
    # report another message
    print('Coroutine is done')

# main coroutine
async def main():
    # report a message
    print('Main starting the delay')
    # run the delayed coroutine
    await delay(task_coro(), 3)
    # report a final message
    print('Main all done')

# run the asyncio program
asyncio.run(main())

Running the example first creates the main() coroutine and uses it as the entry point into the asyncio program.

The main() coroutine runs and reports a message.

It then creates the task_coro() coroutine and wraps it in the delay() coroutine. The main() coroutine then suspends and awaits the delayed coroutine to complete.

The delay() coroutine runs and suspends for the given time limit, in this case, three seconds.

It then resumes and executes the provided coroutine, which in this case is the task_coro() coroutine.

The task_coro() coroutine runs, reporting a message, sleeping for 3 seconds, and reporting a final message before terminating.

The delay() coroutine then terminates and the main() coroutine resumes and reports a final message.

This highlights how we can execute a coroutine after a delay.

Main starting the delay
Coroutine is running
Coroutine is done
Main all done

Next, let's look at how we might execute a coroutine after a delay by scheduling it as an independent task.

Example of Running a Delayed Task

We can explore how to run a coroutine after a delay by scheduling it as an independent task.

In this example, we will wrap our target task coroutine in the delay coroutine to ensure it is not started until after a delay. We will then schedule the delay wrapper coroutine to execute independently as a task.

The main coroutine will then sleep a moment to allow the scheduled task to execute. It will then simulate proceeding with other tasks, in this case sleeping long enough to allow the delayed coroutine to complete.

Tying this together, the complete example is listed below.

# SuperFastPython.com
# example of starting a coroutine after a delay
import asyncio

# coroutine that will start another coroutine after a delay in seconds
async def delay(coro, seconds):
    # suspend for a time limit in seconds
    await asyncio.sleep(seconds)
    # execute the other coroutine
    await coro

# task coroutine to do something
async def task_coro():
    # report a message
    print('Coroutine is running')
    # simulate some long running task
    await asyncio.sleep(3)
    # report another message
    print('Coroutine is done')

# main coroutine
async def main():
    # report a message
    print('Main starting the delay')
    # schedule the delayed task
    task = asyncio.create_task(delay(task_coro(), 3))
    # start the task executing
    await asyncio.sleep(0)
    # simulate doing other things...
    await asyncio.sleep(7)
    # report a final message
    print('Main all done')

# run the asyncio program
asyncio.run(main())

Running the example first creates the main() coroutine and uses it as the entry point into the asyncio program.

The main() coroutine runs and reports a message.

It then creates the task_coro() coroutine and wraps it in the delay() coroutine which in turn is wrapped in an asyncio.Task. This schedules the delayed coroutine for independent execution.

The main() coroutine then suspends for a moment and allows the scheduled coroutine to run.

The delay() coroutine runs and suspends for the given time limit, in this case, three seconds.

The main() coroutine resumes and continues on with other activities. In this case, it simply sleeps for seven seconds.

The task_coro() coroutine runs after the delay, reporting a message, sleeping for 3 seconds, and reporting a final message before terminating. The delay() coroutine then terminates.

The main() coroutine resumes after its sleep, reports a message, and then terminates.

This highlights how we can execute a coroutine after a delay independently of the calling coroutine.

Main starting the delay
Coroutine is running
Coroutine is done
Main all done

Takeaways

You now know how to run a coroutine after a delay.

If you enjoyed this tutorial, you will love my book: Python Asyncio Jump-Start. It covers everything you need to master the topic with hands-on examples and clear explanations.