Loading the lesson
Loading the lesson
Search AllRounder.ai
Search the courses, subjects, tracks, games and features, or jump straight to a page.
Search the courses, subjects, tracks, games and features, or jump straight to a page.
Learn it your way
A short class with Sarah and her students. The other ways are under More.
Sign up free to listen. Transcripts are free to read.
You're building an app that calls three APIs, but the APIs aren't ready yet. How can you still develop and test the async code?
Pretend to call them? Like a function that waits and then returns fake data?
Exactly, that's a simulated network call. In asyncio we write a coroutine that does await asyncio.sleep() for the delay and then returns some data.
Why not just use time.sleep()?
Because time.sleep() freezes the whole program. await asyncio.sleep() pauses only that coroutine, so the event loop can run the other calls meanwhile, just like real non-blocking I/O.
Our example fetches three sites with random delays between 1 and 3 seconds. If we ran them one after another, how long could it take?
Up to 9 seconds.
And with asyncio.gather()?
About as long as the slowest one, so at most 3 seconds, because they all wait at the same time.
Right. And gather() returns the results in the order you passed the coroutines in, even when they finish in a different order.
Real networks fail. How could our simulation test that?
Make fetch_data() raise an exception now and then, say 30% of the time.
Good. Then use asyncio.as_completed() with try and except around each await, so one failure doesn't stop the others being handled.
Can gather() cope with failures too?
Pass return_exceptions=True. Failed calls then come back as exception objects in the results list, instead of the first error being raised out of gather().
await asyncio.sleep(seconds) stands in for the time a real request spends waiting on the network.
Coroutines that wait at the same time overlap their delays, so the total is close to the slowest call.
asyncio.gather() returns results in the order its arguments were given, not the order they finished.
Raising exceptions at random or adding timeouts tests error handling without a flaky real server.
asyncio.run(main()) starts the event loop, runs main() to completion and then closes the loop.
One after another versus all together: three simulated calls of half a second each take about 1.5 seconds in sequence but about 0.5 seconds with gather():
import asyncio
import time
async def fake_request(name, delay):
await asyncio.sleep(delay) # stands in for waiting on the network
return name + " done"
async def one_by_one():
return [await fake_request("call %d" % i, 0.5) for i in range(3)]
async def all_together():
return await asyncio.gather(*(fake_request("call %d" % i, 0.5) for i in range(3)))
for runner in (one_by_one, all_together):
start = time.perf_counter()
results = asyncio.run(runner())
print(f"{runner.__name__}: {time.perf_counter() - start:.1f} s -> {results}")Simulated failures collected with gather(..., return_exceptions=True); the random seed is fixed so that the same calls fail on every run:
import asyncio
import random
random.seed(12)
async def fetch_data(site_name):
await asyncio.sleep(random.uniform(0.1, 0.5))
if random.random() < 0.3:
raise ConnectionError(f"Network error at {site_name}")
return f"Data from {site_name}"
async def main():
sites = ["api1.example.com", "api2.example.com", "api3.example.com", "api4.example.com"]
results = await asyncio.gather(*(fetch_data(s) for s in sites), return_exceptions=True)
for site, result in zip(sites, results):
if isinstance(result, Exception):
print(f"{site}: FAILED ({result})")
else:
print(f"{site}: {result}")
asyncio.run(main())A timeout for a server that answers too slowly, using asyncio.wait_for():
import asyncio
async def slow_server():
await asyncio.sleep(5) # this server takes far too long
return "finally answered"
async def main():
try:
reply = await asyncio.wait_for(slow_server(), timeout=1.0)
print(reply)
except asyncio.TimeoutError:
print("gave up after 1 second")
asyncio.run(main())Await the sleep and the others run on; call time.sleep and the concurrency's gone.
A tour guide sends three helpers to buy tickets at three counters. She doesn't wait in each queue herself; she waits once for all three to return. Her total wait is the longest queue, and the tickets are handed over in the order she sent the helpers out, just like asyncio.gather().
"Sleep, Gather, Run": asyncio.sleep() fakes the wait, asyncio.gather() runs the calls together, asyncio.run() starts it all.
Term
What stands in for network latency in an asyncio simulation?
Click to flip
Term
Why use `asyncio.sleep()` instead of `time.sleep()` in a coroutine?
Click to flip
Term
In what order does `asyncio.gather()` return its results?
Click to flip
Term
Three concurrent simulated calls take 1, 2 and 3 seconds. How long does `gather()` take?
Click to flip
Term
Which function lets you handle each result as soon as its task finishes?
Click to flip
A function defined with async def that can pause at await and resume later.
The asyncio scheduler that runs coroutines, switching between them whenever one is waiting.
A coroutine that pauses the current task for a number of seconds without blocking other tasks.
Runs several awaitables concurrently and returns their results as a list in argument order.
Returns an iterator of awaitables; awaiting each one gives the next result to finish, so results can be handled in completion order.
The delay between sending a request and receiving its response.
Work that spends most of its time waiting for input or output, such as network requests, rather than computing.
Simulated network calls use asyncio to mimic real-world asynchronous I/O operations like fetching data over a network. This helps test and develop asynchronous programs without needing actual network connections.
By simulating network calls with asyncio.sleep() and asynchronous functions, Python programs can mimic the latency and concurrency of real network requests. This technique is useful for testing concurrency logic, performance, and error handling in an environment without relying on external servers.
A simulated network call is a coroutine that behaves like a request to a server: it waits for a while and then returns some data, without touching the network. The waiting is done with await asyncio.sleep(), which hands control back to the event loop, so other simulated calls can make progress in the meantime, exactly as they would while waiting on real sockets.
Real network calls can be unpredictable, slow, or unavailable during development or testing. Simulating them allows developers to:
await.import asyncio
import random
async def fetch_data(site_name):
print(f"Starting fetch from {site_name}")
await asyncio.sleep(random.uniform(1, 3)) # Simulate network delay
print(f"Completed fetch from {site_name}")
return f"Data from {site_name}"
async def main():
sites = ['api1.example.com', 'api2.example.com', 'api3.example.com']
tasks = [fetch_data(site) for site in sites]
results = await asyncio.gather(*tasks)
print("All data fetched:", results)
asyncio.run(main())Explanation:
fetch_data() mimics a network call with random delay.asyncio.gather().Run it a few times. The three "Starting" lines always come first, because each coroutine runs until its first await. The "Completed" lines appear in whichever order the random delays finish, yet results always lists the data in the order the coroutines were passed to gather(). The whole program takes about as long as the slowest fetch (at most 3 seconds), not the sum of the three delays.
time.sleep() blocks the whole thread, so while one call waits nothing else can run, and three calls take the sum of their delays. await asyncio.sleep() suspends only the current coroutine and lets the event loop run the others, which is how a real non-blocking network call behaves.