Python Async Programming
As discussed in the corresponding Rust asynchronous-programming note, the general model can be understood through three components:
Python’s high-level asyncio abstractions correspond to these concepts as follows:
| General concept | Python asyncio concept |
|---|---|
| Executor | Event loop |
| Description of suspendable work | Coroutine |
| Scheduled unit of work | Task |
| Pending result or event | Future |
| Scheduling mechanism | Cooperative scheduling through await, callbacks, timers, and I/O readiness |
The event loop acts as both the executor and the central scheduler. It runs ready tasks, processes callbacks, manages timers, and monitors asynchronous I/O.
Asynchronous programming is particularly useful for I/O-bound workloads, including:
Suppose a server is handling 1,000 network connections. Most connections spend much of their lifetime waiting for clients or external services. Creating a dedicated operating-system thread for every connection can consume substantial resources. With asynchronous programming, one event-loop thread can manage many connections by running only the tasks that are currently able to make progress.
However, asynchronous programming does not automatically make CPU-intensive calculations faster. A long synchronous computation can still occupy the event-loop thread and prevent every other task on that loop from running. This is because of the cooperative scheduling, and the CPU-bounded task will not yield control back to scheduler and it will momopolize the main thread.
The event loop is the core executor and scheduler of an asyncio application.
Conceptually, the event loop repeatedly performs the following operations:
The event loop therefore behaves like an orchestra conductor: it does not necessarily perform the actual work itself, but it decides which ready operation receives control.
Application code normally starts the event loop using asyncio.run():
import asyncio
async def main() -> None:
print("Hello from asyncio")
asyncio.run(main())
asyncio.run(main()) manages the lifecycle of the event loop and runs the main() coroutine until it completes. Application developers normally use this high-level interface rather than manually creating and closing event-loop objects.
A function declared using async def is called a coroutine function.
async def fetch_data() -> str:
return "data"
Calling a regular function immediately begins executing its body:
def add(a: int, b: int) -> int:
return a + b
result = add(1, 2)
Calling a coroutine function behaves differently:
coroutine = fetch_data()
print(coroutine)
The call creates a coroutine object, but the body of fetch_data() does not immediately execute.
The distinction is:
fetch_data is a coroutine function.fetch_data() creates a coroutine object.A coroutine must eventually be awaited or scheduled as a task:
async def main() -> None:
result = await fetch_data()
print(result)
asyncio.run(main())
Coroutines are conceptually related to generators. Both can preserve their local state while suspended and later resume from the previous suspension point.
A Task is a coroutine that has been attached to an event loop and scheduled for execution.
The recommended high-level method for creating one is asyncio.create_task():
import asyncio
async def worker() -> None:
print("Worker started")
await asyncio.sleep(1)
print("Worker finished")
async def main() -> None:
task = asyncio.create_task(worker())
print("Worker has been scheduled")
await task
asyncio.run(main())
Calling worker() only creates a coroutine object:
coroutine = worker()
Calling asyncio.create_task(worker()) does two things:
asyncio.Task.Creating a task does not necessarily execute its entire body immediately. It makes the task eligible to run when the event loop next receives control.
A Future represents a result that may not be available yet.
A Future normally has one of three states:
When completed, it stores either:
A Future does not necessarily contain the computation that produces the result. Instead, it represents the status and eventual outcome of that computation.
A Task is a specialized kind of Future. It adds the ability to execute and manage a coroutine:
Future
└── Task
└── Runs a coroutine
This distinction is useful:
| Object | Purpose |
|---|---|
| Coroutine | Describes suspendable computation |
| Future | Represents a pending result |
| Task | Executes a coroutine and exposes its result as a Future |
Most application code directly uses coroutines and Tasks. Futures are more commonly encountered in lower-level libraries and event-loop integrations.
await KeywordThe await keyword pauses the current coroutine until an awaitable produces a result.
An awaitable may be:
__await__()async def main() -> None:
result = await fetch_data()
print(result)
Conceptually, await means:
Continue this coroutine once the awaited operation is ready. While it is genuinely waiting, allow the event loop to run other ready work.
The local variables and execution position of the coroutine are preserved while it is suspended.
The awaited operation is not yet complete, the current Task will register a wake-up callback on the Future or Task (for couroutine, just call the function wrapped in the coroutine). Then, it suspends the current coroutine and returns the control to the event loop.
result = await fetch_data()
Awaiting a coroutine runs that coroutine as part of the current Task. It does not automatically create an independently scheduled Task.
The child coroutine may still eventually return control to the event loop when it reaches an operation that genuinely suspends, such as:
await asyncio.sleep(1)
task = asyncio.create_task(fetch_data())
result = await task
The coroutine has already been scheduled independently as a Task. Awaiting it means that the current coroutine should resume after that Task finishes.
Therefore:
await coroutine()
and:
await asyncio.create_task(coroutine())
do not have exactly the same scheduling behaviour.
awaitConsider the following program:
import asyncio
import time
async def fetch(name: str, delay: int) -> str:
print(f"Starting {name}")
await asyncio.sleep(delay)
print(f"Finished {name}")
return name
async def main() -> None:
start = time.perf_counter()
first = await fetch("A", 2)
second = await fetch("B", 2)
elapsed = time.perf_counter() - start
print(first, second)
print(f"Elapsed: {elapsed:.2f} seconds")
asyncio.run(main())
Although fetch() is asynchronous, the two calls are still sequential:
fetch("A", 2) to finish.fetch("B", 2).The total execution time is approximately four seconds.
The presence of async and await alone does not automatically create concurrency.
To run the operations concurrently, schedule both before awaiting their results:
import asyncio
import time
async def fetch(name: str, delay: int) -> str:
print(f"Starting {name}")
await asyncio.sleep(delay)
print(f"Finished {name}")
return name
async def main() -> None:
start = time.perf_counter()
first_task = asyncio.create_task(fetch("A", 2))
second_task = asyncio.create_task(fetch("B", 2))
first = await first_task
second = await second_task
elapsed = time.perf_counter() - start
print(first, second)
print(f"Elapsed: {elapsed:.2f} seconds")
asyncio.run(main())
A possible execution sequence is:
Starting A
Starting B
Finished A
Finished B
Elapsed: 2.00 seconds
The tasks do not necessarily run in parallel. Instead:
Both waits overlap, so the total time is approximately two seconds.
asyncio.gather()asyncio.gather() runs multiple awaitables concurrently and collects their results in the original input order:
async def main() -> None:
results = await asyncio.gather(
fetch("A", 3),
fetch("B", 1),
fetch("C", 2),
)
print(results)
Output:
Starting A
Starting B
Starting C
Finished B
Finished C
Finished A
['A', 'B', 'C']
Even though B finishes first, the returned results correspond to the original argument order.
gather() is convenient when the operations are related primarily because their results need to be collected together.
asyncio.TaskGroupPython 3.11 introduced asyncio.TaskGroup, which provides structured concurrency.
import asyncio
async def main() -> None:
async with asyncio.TaskGroup() as group:
first_task = group.create_task(fetch("A", 3))
second_task = group.create_task(fetch("B", 1))
third_task = group.create_task(fetch("C", 2))
# All tasks have finished when the block exits.
print(first_task.result())
print(second_task.result())
print(third_task.result())
asyncio.run(main())
A TaskGroup guarantees that its tasks are accounted for before the context manager exits. It also provides coordinated exception and cancellation behaviour when one of its tasks fails.
For new code involving a clearly related group of tasks, TaskGroup often expresses ownership more reliably than manually creating several detached tasks.
See the guide
class Rock:
def __await__(self):
value_sent_in = yield 7
print(f"Rock.__await__ resuming with value: {value_sent_in}.")
return value_sent_in
async def main():
print("Beginning coroutine main().")
rock = Rock()
print("Awaiting rock...")
value_from_rock = await rock
print(f"Coroutine received value: {value_from_rock} from rock.")
return 23
coroutine = main()
intermediate_result = coroutine.send(None)
print(f"Coroutine paused and returned intermediate value: {intermediate_result}.")
print(f"Resuming coroutine and sending in value: 42.")
try:
coroutine.send(42)
except StopIteration as e:
returned_value = e.value
print(f"Coroutine main() finished and provided value: {returned_value}.")
Beginning coroutine main().
Awaiting rock...
Coroutine paused and returned intermediate value: 7.
Resuming coroutine and sending in value: 42.
Rock.__await__ resuming with value: 42.
Coroutine received value: 42 from rock.
Coroutine main() finished and provided value: 23.
await?Suppose a Task is running the following coroutine:
async def worker() -> None:
result = await perform_io()
process(result)
A simplified execution sequence is:
await perform_io().await expression evaluates to the Future’s result.process(result).An awaited object participates in this mechanism through its __await__() method. Internally, yielded values propagate through the coroutine chain until control reaches the Task and event loop.
The event loop does not repeatedly execute every suspended coroutine to ask whether its operation has completed. There is no polling.
Instead, it relies on operating-system mechanisms for detecting I/O readiness.
On selector-based systems, the rough model is:
Depending on the platform, Python may use selector-based readiness mechanisms or Windows I/O completion facilities. The high-level asyncio interface hides most of these platform differences.
asyncio uses cooperative rather than preemptive task scheduling.
A Task continues running until it:
If the coroutine never reaches an effective suspension point, it can occupy the event-loop thread indefinitely. Other tasks do not receive an opportunity to run.
If the task has still synchronous CPU work. While it runs, the event loop cannot process other tasks on the same thread.
Blocking code should not be called directly from the event-loop thread.
Bad example:
import time
async def bad_worker() -> None:
time.sleep(5)
time.sleep(5) blocks the entire thread. During those five seconds, no other Task on that event loop can run.
The asynchronous equivalent is:
async def good_worker() -> None:
await asyncio.sleep(5)
asyncio.sleep() suspends only the current Task and returns control to the event loop.
The same problem occurs with synchronous HTTP clients, synchronous database drivers, blocking file operations, and other traditional blocking APIs.
Whenever possible, use libraries designed for asyncio.
asyncio.to_thread()A blocking synchronous function can be moved to a worker thread using asyncio.to_thread():
import asyncio
import time
def blocking_operation() -> str:
time.sleep(2)
return "finished"
async def main() -> None:
result = await asyncio.to_thread(blocking_operation)
print(result)
asyncio.run(main())
The event-loop thread remains available to process other Tasks while the blocking function runs elsewhere.
This is particularly useful when:
For ordinary CPython code, to_thread() is mainly useful for blocking I/O. Pure Python CPU-heavy work usually requires multiprocessing or another execution strategy to obtain meaningful parallel speed-up, although native extensions that release the GIL may behave differently.
Even in a single-threaded event loop, asynchronous tasks can still create logical race conditions because execution can switch at await points.
counter = 0
async def increment() -> None:
global counter
current = counter
await asyncio.sleep(0)
counter = current + 1
Two tasks could both read the same value before either writes its update.
An asyncio.Lock can protect the critical section:
counter = 0
lock = asyncio.Lock()
async def increment() -> None:
global counter
async with lock:
current = counter
await asyncio.sleep(0)
counter = current + 1
asyncio also provides other coordination primitives, including:
EventConditionSemaphoreBarrierThese primitives coordinate asyncio Tasks and are not replacements for synchronization mechanisms used between operating-system threads.
Where possible, avoid holding a lock across slow external operations. Keeping critical sections small reduces unnecessary contention between tasks.
Tasks can communicate using asyncio.Queue:
import asyncio
async def producer(queue: asyncio.Queue[int]) -> None:
for value in range(5):
await queue.put(value)
await queue.put(-1)
async def consumer(queue: asyncio.Queue[int]) -> None:
while True:
value = await queue.get()
try:
if value == -1:
return
print(f"Received {value}")
finally:
queue.task_done()
async def main() -> None:
queue: asyncio.Queue[int] = asyncio.Queue()
async with asyncio.TaskGroup() as group:
group.create_task(producer(queue))
group.create_task(consumer(queue))
asyncio.run(main())
Queues are useful for producer–consumer systems because they reduce the need for multiple tasks to mutate the same shared state directly.
A bounded queue can also provide backpressure:
queue: asyncio.Queue[bytes] = asyncio.Queue(maxsize=100)
When the queue is full, await queue.put(item) suspends the producer until consumers create more capacity.
An asynchronous context manager uses async with:
async with database.transaction() as transaction:
await transaction.execute(query)
Its conceptual interface is:
class AsyncResource:
async def __aenter__(self):
...
async def __aexit__(self, exc_type, exc, traceback):
...
The actual semantic is not calling aenter and aexit at the start or end of the block.
This is useful when acquiring or releasing the resource may itself require asynchronous operations.
contextlib.AsyncExitStack extends this mechanism to a dynamic number of resources: synchronous and asynchronous context managers or cleanup callbacks can be registered as the program runs, and they are automatically executed in reverse order when the stack exits.
An asynchronous iterator is consumed using async for:
async for message in websocket:
print(message)
Its conceptual interface includes:
class AsyncIterator:
def __aiter__(self):
return self
async def __anext__(self):
...
The iterator may suspend while obtaining the next item. This makes it useful for streaming data from networks, databases, files exposed through asynchronous libraries, and message brokers.
This learning note was developed from my studies and discussions with GPT. The final content was independently reviewed, technically verified, organized, and edited by the author..