Skip to content

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:

blocking_example.py
import asyncio
import 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

async_fix.py
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() → use asyncio.sleep()
  • requests.get() → use aiohttp or httpx
  • open() for files → use aiofiles or asyncio.to_thread()
  • Any CPU-bound work → use asyncio.to_thread() or run_in_executor()

Mistake #2: Calling a Coroutine Directly

I defined an async function and called it:

direct_call.py
async def main():
print("Hello, asyncio!")
main() # Called it directly

What 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()

asyncio_run.py
import asyncio
async def main():
print("Hello, asyncio!")
asyncio.run(main()) # This actually runs it

Output:

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:

unawaited_task.py
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 finished

No 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

await_task.py
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 1
Fetched user 2
Fetched user 3
Results: ['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:

await_outside.py
def process():
await asyncio.sleep(1) # Thought this would work
process()

What I got:

SyntaxError: 'await' outside async function

await can only be used inside async def functions. This makes sense—awaiting requires a coroutine context.

The Fix: Make the Function Async

await_inside.py
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:

sequential_iteration.py
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 0
Processing item 0
Fetched item 1
Processing item 1
Fetched item 2
Processing item 2
Fetched item 3
Processing item 3
Fetched item 4
Processing item 4

Sequential. 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

concurrent_iteration.py
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 0
Fetched item 1
Fetched item 2
Fetched item 3
Fetched item 4
All items fetched: [0, 1, 2, 3, 4]
Processing item 0
Processing item 1
Processing item 2
Processing item 3
Processing item 4
All 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:

blocking_io.py
import asyncio
import 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

async_io.py
import asyncio
import 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 LibraryAsync Alternative
requestsaiohttp, httpx
open() / read()aiofiles
time.sleep()asyncio.sleep()
sqlite3aiosqlite
psycopg2asyncpg

The Mental Model That Clicked

I used to think async def automatically made things concurrent. It doesn’t.

What async actually does:

  1. Allows the function to use await
  2. Allows the event loop to switch to other tasks at await points
  3. 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:

  1. Check for blocking calls: Search for time.sleep, requests, open() in async functions
  2. Verify all tasks are awaited: Look for create_task() without corresponding await
  3. Ensure coroutines are scheduled: Did I use asyncio.run() or forget it?
  4. Test with prints: Add print statements before and after awaits to see execution order

Summary of Fixes

correct_async.py
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 loop

Final 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