Using asyncio call_soon, call_at, call_later, and call_soon_threadsafe in Python
This article explains how Python's asyncio library schedules callbacks with call_soon, call_at, call_later, and call_soon_threadsafe, providing detailed usage descriptions and multiple example code snippets that demonstrate immediate, absolute‑time, relative‑time, and thread‑safe task scheduling within an event loop.
Introduction Python's asyncio library offers several functions— call_soon , call_at , call_later , and call_soon_threadsafe —to schedule callbacks in an event loop. The following sections describe each function and provide concrete examples.
call_soon Schedules a callback to run as soon as possible, i.e., in the next iteration of the event loop.
import asyncio
def callback():
print("Callback executed")
async def main():
print("Calling callback soon")
asyncio.get_event_loop().call_soon(callback)
await asyncio.sleep(0.1) # keep the loop running
asyncio.run(main())call_at Schedules a callback to run at a specific absolute time (Unix timestamp).
import asyncio
import time
def callback():
print("Callback executed at:", time.strftime('%X'))
async def main():
print("Calling callback at a specific time")
future_time = time.time() + 2
asyncio.get_event_loop().call_at(future_time, callback)
await asyncio.sleep(3) # keep the loop running
asyncio.run(main())call_later Schedules a callback to run after a relative delay.
import asyncio
def callback():
print("Callback executed after delay")
async def main():
print("Calling callback later")
asyncio.get_event_loop().call_later(2, callback)
await asyncio.sleep(3) # keep the loop running
asyncio.run(main())call_soon_threadsafe Allows safe scheduling of a callback from another thread.
import asyncio
import threading
def callback():
print("Callback executed")
async def main():
print("Calling callback soon (thread-safe)")
loop = asyncio.get_event_loop()
loop.call_soon_threadsafe(callback)
await asyncio.sleep(0.1) # keep the loop running
def run_async():
asyncio.run(main())
thread = threading.Thread(target=run_async)
thread.start()
thread.join()Combined Example Demonstrates using multiple scheduling methods together.
import asyncio
import time
def callback(msg):
print(f"{msg} executed at:", time.strftime('%X'))
async def main():
print("Setting up callbacks")
asyncio.get_event_loop().call_soon(callback, "Callback 1")
future_time = time.time() + 2
asyncio.get_event_loop().call_at(future_time, callback, "Callback 2")
asyncio.get_event_loop().call_later(1, callback, "Callback 3")
loop = asyncio.get_event_loop()
loop.call_soon_threadsafe(callback, "Callback 4")
await asyncio.sleep(3) # keep the loop running
asyncio.run(main())Conclusion The article covered four asyncio scheduling functions, illustrating how to schedule immediate, absolute‑time, relative‑time, and thread‑safe callbacks with clear code examples, enabling developers to manage asynchronous tasks effectively within Python's event loop.
Test Development Learning Exchange
Test Development Learning Exchange
How this landed with the community
Was this worth your time?
0 Comments
Thoughtful readers leave field notes, pushback, and hard-won operational detail here.