condition.wait_for is a coroutine method for waiting until a supplied predicate becomes true. Used inside an acquired asyncio.Condition, it releases the condition lock while waiting and reacquires it before returning. section 11.5
Use it when progress depends on shared state rather than a single notification:
- Acquire the condition with async with. section 11.5
- Call await condition.wait_for(predicate) to suspend until the predicate succeeds. section 11.5
- Let the condition manage releasing the lock during the wait and reacquiring it before continuing. section 11.5
Wait for a connection state
A connection can protect its state with an asyncio.Condition. An execute coroutine acquires the condition and waits with await self._condition.wait_for(self._is_initialized). Initialization changes the state and notifies waiting tasks; execute resumes only after the predicate confirms that the connection is initialized. section 11.5
Listing 11.15 Using conditions to wait for specific states
import asyncio
from enum import Enum
class ConnectionState(Enum):
WAIT_INIT = 0
INITIALIZING = 1
INITIALIZED = 2
class Connection:
def __init__(self):
self._state = ConnectionState.WAIT_INIT
self._condition = asyncio.Condition()
async def initialize(self):
await self._change_state(ConnectionState.INITIALIZING)
print('initialize: Initializing connection...')
await asyncio.sleep(3) # simulate connection startup time
print('initialize: Finished initializing connection')
await self._change_state(ConnectionState.INITIALIZED)
async def execute(self, query: str):
async with self._condition:
print('execute: Waiting for connection to initialize')
await self._condition.wait_for(self._is_initialized)
print(f'execute: Running {query}!!!')
await asyncio.sleep(3) # simulate a long query
async def _change_state(self, state: ConnectionState):
async with self._condition:
print(f'change_state: State changing from {self._state} to {state}')
self._state = state
self._condition.notify_all()
def _is_initialized(self):
if self._state is not ConnectionState.INITIALIZED:
print(f'_is_initialized: Connection not finished initializing, state is {self._state}')
return False
print(f'_is_initialized: Connection is initialized!')
return True
async def main():
connection = Connection()
query_one = asyncio.create_task(connection.execute('select * from table'))
query_two = asyncio.create_task(connection.execute('select * from other_table'))
asyncio.create_task(connection.initialize())
await query_one
await query_two
asyncio.run(main())
Runnable state-based example showing wait_for coordinating connection initialization and query execution. section 11.5
