Python · Syntax · Advanced
Async Python
Write concurrent programs using `async def`, `await`, and `asyncio`. Covers coroutines, `asyncio.gather`, tasks, `asyncio.Queue`, and async iterators.
Quick topic start and explanations before exercises (exercises below):
Tasks, async for/with, timeout, concurrent HTTP pattern
#Common pitfalls, asyncio vs threads vs multiprocessing, quick ref
#Exercises:
First coroutine with asyncio.run
#Write an `async def greet(name: str)` coroutine that prints `'Hello, <name>!'`, waits 0.1 seconds with `asyncio.sleep`, then prints `'Goodbye, <name>!'`. Run it with `asyncio.run(greet('Alice'))`.
import asyncio
async def greet(name: str) -> None:
pass
asyncio.run(greet('Alice'))
# Hello, Alice!
# Goodbye, Alice!
Solution
import asyncio
async def greet(name: str) -> None:
print(f'Hello, {name}!')
await asyncio.sleep(0.1)
print(f'Goodbye, {name}!')
asyncio.run(greet('Alice'))
# Hello, Alice!
# Goodbye, Alice!
Concurrent tasks with asyncio.gather
#Write a coroutine `fetch(url: str, delay: float) -> str` that simulates an HTTP request by sleeping `delay` seconds and returning `f'Response from {url}'`. Use `asyncio.gather` to run three fetches concurrently: `fetch('A', 0.3)`, `fetch('B', 0.1)`, `fetch('C', 0.2)`. Print all results. The total time should be about 0.3s, not 0.6s.
import asyncio
async def fetch(url: str, delay: float) -> str:
pass
async def main() -> None:
results = await asyncio.gather(
fetch('A', 0.3),
fetch('B', 0.1),
fetch('C', 0.2),
)
for r in results:
print(r)
asyncio.run(main())
Solution
import asyncio
async def fetch(url: str, delay: float) -> str:
await asyncio.sleep(delay)
return f'Response from {url}'
async def main() -> None:
results = await asyncio.gather(
fetch('A', 0.3),
fetch('B', 0.1),
fetch('C', 0.2),
)
for r in results:
print(r)
asyncio.run(main())
# Response from A
# Response from B
# Response from C
Background tasks with create_task
#Write a `main()` coroutine that creates two background tasks using `asyncio.create_task`: one that counts down from 3 to 1 (printing each number with 0.1s delay), and one that prints `'Working...'` three times with 0.15s delay. Await both tasks at the end. Observe that the outputs interleave.
import asyncio
async def countdown() -> None:
for i in range(3, 0, -1):
print(f'Countdown: {i}')
await asyncio.sleep(0.1)
async def worker() -> None:
for _ in range(3):
print('Working...')
await asyncio.sleep(0.15)
async def main() -> None:
# create tasks and await them
pass
asyncio.run(main())
Solution
import asyncio
async def countdown() -> None:
for i in range(3, 0, -1):
print(f'Countdown: {i}')
await asyncio.sleep(0.1)
async def worker() -> None:
for _ in range(3):
print('Working...')
await asyncio.sleep(0.15)
async def main() -> None:
task1 = asyncio.create_task(countdown())
task2 = asyncio.create_task(worker())
await task1
await task2
asyncio.run(main())
Producer-consumer with asyncio.Queue
#Implement a producer-consumer pattern using `asyncio.Queue`. The `producer` coroutine puts numbers 1–5 into the queue (with 0.05s delay each), then puts `None` as a sentinel. The `consumer` coroutine reads items from the queue until it sees `None`, printing `'Consuming: <n>'` for each number.
import asyncio
async def producer(queue: asyncio.Queue) -> None:
pass
async def consumer(queue: asyncio.Queue) -> None:
pass
async def main() -> None:
queue: asyncio.Queue = asyncio.Queue()
await asyncio.gather(producer(queue), consumer(queue))
asyncio.run(main())
# Consuming: 1
# Consuming: 2
# ...
Solution
import asyncio
async def producer(queue: asyncio.Queue) -> None:
for i in range(1, 6):
await queue.put(i)
await asyncio.sleep(0.05)
await queue.put(None) # sentinel
async def consumer(queue: asyncio.Queue) -> None:
while True:
item = await queue.get()
if item is None:
break
print(f'Consuming: {item}')
async def main() -> None:
queue: asyncio.Queue = asyncio.Queue()
await asyncio.gather(producer(queue), consumer(queue))
asyncio.run(main())
Timeout with asyncio.wait_for
#Write a coroutine `slow_op(seconds: float) -> str` that sleeps for `seconds` and returns `'done'`. Use `asyncio.wait_for(slow_op(5), timeout=0.2)` to attempt the operation with a 0.2-second timeout. Catch `asyncio.TimeoutError` and print `'Operation timed out!'`.
import asyncio
async def slow_op(seconds: float) -> str:
await asyncio.sleep(seconds)
return 'done'
async def main() -> None:
try:
result = await asyncio.wait_for(slow_op(5), timeout=0.2)
print(result)
except asyncio.TimeoutError:
pass # print the timeout message here
asyncio.run(main())
# Operation timed out!
Solution
import asyncio
async def slow_op(seconds: float) -> str:
await asyncio.sleep(seconds)
return 'done'
async def main() -> None:
try:
result = await asyncio.wait_for(slow_op(5), timeout=0.2)
print(result)
except asyncio.TimeoutError:
print('Operation timed out!')
asyncio.run(main())
# Operation timed out!
Async generator
#Write an async generator `async_range(start: int, stop: int, delay: float)` that yields integers from `start` to `stop-1`, pausing `delay` seconds between each yield. Use `async for` to consume it and print each value.
import asyncio
async def async_range(start: int, stop: int, delay: float):
pass
async def main() -> None:
async for value in async_range(0, 4, 0.05):
print(value)
asyncio.run(main())
# 0
# 1
# 2
# 3
Solution
import asyncio
async def async_range(start: int, stop: int, delay: float):
for i in range(start, stop):
yield i
await asyncio.sleep(delay)
async def main() -> None:
async for value in async_range(0, 4, 0.05):
print(value)
asyncio.run(main())
# 0
# 1
# 2
# 3
Async context manager
#Create a class `AsyncTimer` that works as an async context manager. `__aenter__` records the start time and prints `'Timer started'`. `__aexit__` computes elapsed time and prints `f'Elapsed: {elapsed:.2f}s'`. Use it with `async with AsyncTimer():`.
import asyncio
import time
class AsyncTimer:
async def __aenter__(self):
pass
async def __aexit__(self, *args):
pass
async def main() -> None:
async with AsyncTimer():
await asyncio.sleep(0.1)
asyncio.run(main())
# Timer started
# Elapsed: 0.10s
Solution
import asyncio
import time
class AsyncTimer:
async def __aenter__(self):
self._start = time.perf_counter()
print('Timer started')
return self
async def __aexit__(self, *args):
elapsed = time.perf_counter() - self._start
print(f'Elapsed: {elapsed:.2f}s')
async def main() -> None:
async with AsyncTimer():
await asyncio.sleep(0.1)
asyncio.run(main())
# Timer started
# Elapsed: 0.10s
Limit concurrency with asyncio.Semaphore
#You have 8 tasks but want at most 3 running concurrently. Use `asyncio.Semaphore(3)` to limit concurrency. Each task should print `'Task N started'`, sleep 0.1s, then print `'Task N done'`. Create all 8 tasks with `asyncio.gather`.
import asyncio
async def task(n: int, sem: asyncio.Semaphore) -> None:
async with sem:
print(f'Task {n} started')
await asyncio.sleep(0.1)
print(f'Task {n} done')
async def main() -> None:
sem = asyncio.Semaphore(3)
# create and gather 8 tasks (n from 1 to 8)
pass
asyncio.run(main())
Solution
import asyncio
async def task(n: int, sem: asyncio.Semaphore) -> None:
async with sem:
print(f'Task {n} started')
await asyncio.sleep(0.1)
print(f'Task {n} done')
async def main() -> None:
sem = asyncio.Semaphore(3)
await asyncio.gather(*(task(n, sem) for n in range(1, 9)))
asyncio.run(main())
Run blocking code with run_in_executor
#The function `blocking_sum(n)` uses a CPU-bound loop to compute the sum of 1..n (simulating slow blocking work). Wrap it in an async function `async_sum(n)` using `loop.run_in_executor(None, blocking_sum, n)` so it doesn't block the event loop. Run two `async_sum` calls concurrently with `asyncio.gather`.
import asyncio
def blocking_sum(n: int) -> int:
return sum(range(n + 1))
async def async_sum(n: int) -> int:
loop = asyncio.get_running_loop()
# run blocking_sum in a thread pool
pass
async def main() -> None:
results = await asyncio.gather(async_sum(1000), async_sum(2000))
print(results) # [500500, 2001000]
asyncio.run(main())
Solution
import asyncio
def blocking_sum(n: int) -> int:
return sum(range(n + 1))
async def async_sum(n: int) -> int:
loop = asyncio.get_running_loop()
return await loop.run_in_executor(None, blocking_sum, n)
async def main() -> None:
results = await asyncio.gather(async_sum(1000), async_sum(2000))
print(results) # [500500, 2001000]
asyncio.run(main())
Coordination with asyncio.Event
#Use `asyncio.Event` to coordinate two coroutines. The `setter` coroutine waits 0.15s, prints `'Setting event'`, then sets the event. The `waiter` coroutine prints `'Waiting for event...'`, waits with `await event.wait()`, then prints `'Event received!'`. Run both concurrently with `asyncio.gather`.
import asyncio
async def setter(event: asyncio.Event) -> None:
await asyncio.sleep(0.15)
print('Setting event')
event.set()
async def waiter(event: asyncio.Event) -> None:
pass # wait for the event, then print 'Event received!'
async def main() -> None:
event = asyncio.Event()
await asyncio.gather(setter(event), waiter(event))
asyncio.run(main())
# Waiting for event...
# Setting event
# Event received!
Solution
import asyncio
async def setter(event: asyncio.Event) -> None:
await asyncio.sleep(0.15)
print('Setting event')
event.set()
async def waiter(event: asyncio.Event) -> None:
print('Waiting for event...')
await event.wait()
print('Event received!')
async def main() -> None:
event = asyncio.Event()
await asyncio.gather(setter(event), waiter(event))
asyncio.run(main())
# Waiting for event...
# Setting event
# Event received!