How to avoid common Python asyncio mistakes that block your event loop
I wrote an async function and called time.sleep() inside it, expecting it to behave asynchronously. Instead, my entire program froze—nothing else ran for those 2 seconds. The event loop was blocked.
This was my first asyncio mistake. I discovered there are several common pitfalls that can break async code or make it perform worse than synchronous alternatives.
Mistake #1: Using Blocking Calls Inside Async Functions
I had this code:
import asyncioimport time
async def fetch_data(): print("Fetching data...") time.sleep(2) # I thought this was fine in async print("Data fetched!") return "data"
async def process_data(): print("Processing data...")
async def main(): # Expected both to run concurrently await asyncio.gather(fetch_data(), process_data())
asyncio.run(main())What I expected:
- Both functions run concurrently
- “Processing data…” prints immediately
What actually happened:
Fetching data...(2 seconds of silence - nothing happens)Data fetched!Processing data...The time.sleep(2) blocked the entire event loop. During those 2 seconds, no other coroutine could run—not even process_data().
The Fix: Use Async Equivalents
import asyncio
async def fetch_data(): print("Fetching data...") await asyncio.sleep(2) # Yields control to event loop print("Data fetched!") return "data"
async def process_data(): print("Processing data...")
async def main(): await asyncio.gather(fetch_data(), process_data())
asyncio.run(main())Output:
Fetching data...Processing data... # Prints immediately now!Data fetched!Why this works: asyncio.sleep() is a coroutine that yields control back to the event loop. Other coroutines can run during the wait. time.sleep() is synchronous—it blocks everything.
The Rule
Never use blocking operations inside async def functions:
time.sleep()→ useasyncio.sleep()requests.get()→ useaiohttporhttpxopen()for files → useaiofilesorasyncio.to_thread()- Any CPU-bound work → use
asyncio.to_thread()orrun_in_executor()
Mistake #2: Calling a Coroutine Directly
I defined an async function and called it:
async def main(): print("Hello, asyncio!")
main() # Called it directlyWhat I expected: The function runs and prints “Hello, asyncio!”
What actually happened: Nothing printed. No error either.
>>> async def main():... print("Hello, asyncio!")...>>> main()<coroutine object main at 0x1027a6150>Calling main() doesn’t run it. It returns a coroutine object that sits there doing nothing.
The Fix: Use asyncio.run()
import asyncio
async def main(): print("Hello, asyncio!")
asyncio.run(main()) # This actually runs itOutput:
Hello, asyncio!Why: Coroutines must be scheduled on an event loop. asyncio.run() creates a loop, schedules the coroutine, runs it, and closes the loop.
Mistake #3: Creating Tasks but Never Awaiting Them
I wanted to run multiple coroutines concurrently:
import asyncio
async def fetch_user(user_id): await asyncio.sleep(1) print(f"Fetched user {user_id}") return f"user_{user_id}"
async def main(): # Created tasks to run concurrently asyncio.create_task(fetch_user(1)) asyncio.create_task(fetch_user(2)) asyncio.create_task(fetch_user(3))
print("Main finished")
asyncio.run(main())What I expected: All three users fetched concurrently.
What actually happened:
Main finishedNo users were fetched. The tasks were created but never ran.
Why This Happens
When main() finishes, the event loop stops. Any tasks created with create_task() that haven’t completed get cancelled.
The Fix: Await or Gather
import asyncio
async def fetch_user(user_id): await asyncio.sleep(1) print(f"Fetched user {user_id}") return f"user_{user_id}"
async def main(): task1 = asyncio.create_task(fetch_user(1)) task2 = asyncio.create_task(fetch_user(2)) task3 = asyncio.create_task(fetch_user(3))
# Must await them results = await asyncio.gather(task1, task2, task3) print(f"Results: {results}")
asyncio.run(main())Output:
Fetched user 1Fetched user 2Fetched user 3Results: ['user_1', 'user_2', 'user_3']Key insight: create_task() schedules a coroutine for execution, but you must still wait for it to complete—either with await task or await asyncio.gather().
Mistake #4: Using await Outside Async Functions
I tried to use await in a regular function:
def process(): await asyncio.sleep(1) # Thought this would work
process()What I got:
SyntaxError: 'await' outside async functionawait can only be used inside async def functions. This makes sense—awaiting requires a coroutine context.
The Fix: Make the Function Async
import asyncio
async def process(): # Now it's async await asyncio.sleep(1) # Works! print("Processed")
asyncio.run(process())Mistake #5: Thinking async for Creates Concurrent Iteration
I had an async generator and expected concurrent iteration:
import asyncio
async def fetch_items(count): for i in range(count): await asyncio.sleep(0.5) # Simulate async fetch print(f"Fetched item {i}") yield i
async def process_items(): # I thought this would process items concurrently async for item in fetch_items(5): print(f"Processing item {item}")
asyncio.run(process_items())What I expected: Items processed as they arrive, possibly overlapping.
What actually happened:
Fetched item 0Processing item 0Fetched item 1Processing item 1Fetched item 2Processing item 2Fetched item 3Processing item 3Fetched item 4Processing item 4Sequential. Each item is fully processed before the next is fetched.
Why
async for allows the event loop to run other tasks between iterations, but each iteration is still sequential. It doesn’t make the loop concurrent by itself.
The Fix: Use gather() for True Concurrency
import asyncio
async def fetch_item(item_id): await asyncio.sleep(0.5) # Simulate async fetch print(f"Fetched item {item_id}") return item_id
async def process_item(item): print(f"Processing item {item}") return f"processed_{item}"
async def main(): # Create tasks for all items fetch_tasks = [fetch_item(i) for i in range(5)]
# Fetch all concurrently items = await asyncio.gather(*fetch_tasks) print(f"All items fetched: {items}")
# Process all concurrently process_tasks = [process_item(item) for item in items] results = await asyncio.gather(*process_tasks) print(f"All processed: {results}")
asyncio.run(main())Output:
Fetched item 0Fetched item 1Fetched item 2Fetched item 3Fetched item 4All items fetched: [0, 1, 2, 3, 4]Processing item 0Processing item 1Processing item 2Processing item 3Processing item 4All processed: ['processed_0', 'processed_1', 'processed_2', 'processed_3', 'processed_4']All items fetched in ~0.5 seconds total (not 2.5 seconds), then all processed.
Mistake #6: Mixing Blocking I/O with Async Code
I used requests inside an async function:
import asyncioimport requests # Synchronous library
async def fetch_api(): # This blocks the entire event loop! response = requests.get("https://api.example.com/data") return response.json()
async def other_task(): print("Running other task...")
async def main(): await asyncio.gather(fetch_api(), other_task())
asyncio.run(main())During the HTTP request, other_task() can’t run. The event loop is blocked.
The Fix: Use Async Libraries
import asyncioimport aiohttp # Async library
async def fetch_api(): async with aiohttp.ClientSession() as session: async with session.get("https://api.example.com/data") as response: return await response.json()
async def other_task(): print("Running other task...")
async def main(): await asyncio.gather(fetch_api(), other_task())
asyncio.run(main())Now other_task() runs while the HTTP request is in flight.
Quick Reference: Synchronous → Async Libraries
| Blocking Library | Async Alternative |
|---|---|
requests | aiohttp, httpx |
open() / read() | aiofiles |
time.sleep() | asyncio.sleep() |
sqlite3 | aiosqlite |
psycopg2 | asyncpg |
The Mental Model That Clicked
I used to think async def automatically made things concurrent. It doesn’t.
What async actually does:
- Allows the function to use
await - Allows the event loop to switch to other tasks at
awaitpoints - Nothing more
Concurrency requires explicit effort:
- Use
asyncio.gather()to run multiple coroutines - Use
asyncio.create_task()and await the results - Use async libraries instead of blocking ones
The event loop is single-threaded. When you block it, everything stops. The whole point of async is yielding control at strategic points.
A Debugging Workflow
When my async code doesn’t work as expected:
- Check for blocking calls: Search for
time.sleep,requests,open()in async functions - Verify all tasks are awaited: Look for
create_task()without correspondingawait - Ensure coroutines are scheduled: Did I use
asyncio.run()or forget it? - Test with prints: Add print statements before and after awaits to see execution order
Summary of Fixes
import asyncio
async def main(): # GOOD: Async sleep await asyncio.sleep(1)
# GOOD: Create and await tasks tasks = [asyncio.create_task(work(i)) for i in range(3)] results = await asyncio.gather(*tasks)
# GOOD: Use async libraries async with aiohttp.ClientSession() as session: async with session.get(url) as response: data = await response.json()
# NEVER: time.sleep(), requests.get(), open() # in async functions without run_in_executor
asyncio.run(main()) # Required to start event loopFinal Words + More Resources
My intention with this article was to help others share my knowledge and experience. If you want to contact me, you can contact by email: Email me
Here are also the most important links from this article along with some further resources that will help you in this scope:
Oh, and if you found these resources useful, don’t forget to support me by starring the repo on GitHub!
Comments