Mastering Concurrency: A Practical Python Async Await Tutorial with Examples

If you’ve been writing Python for a while, you’ve probably hit that frustrating wall where your script takes hours to run because it’s waiting on network requests, file I/O, or database queries. I remember staring at my terminal years ago, watching a web scraping script crawl through thousands of pages one by one, thinking there had to be a better way.

That better way is asynchronous programming.

Welcome to this comprehensive python async await tutorial with examples. By the end of this guide, you won’t just understand how async and await work under the hood; you’ll be confidently writing concurrent Python code that runs blazingly fast. We will cover the core concepts, build practical, copy-paste-ready projects, and dismantle the common pitfalls that catch almost every developer when they first make the jump to async.

Introduction to Asynchronous Programming in Python

To understand why asynchronous programming is a game-changer, we first need to talk about I/O bound tasks versus CPU bound tasks.

When your code is doing heavy math, training a machine learning model, or processing massive arrays, it is CPU bound. The processor is doing the heavy lifting.

However, when your code is fetching data from a REST API, reading a file from a hard drive, or waiting for a PostgreSQL database to return rows, it is I/O bound. During these operations, your CPU is essentially sitting idle, tapping its fingers on the desk, waiting for the external resource to respond.

In traditional synchronous Python, your code runs sequentially. If Function A takes 5 seconds to fetch data from an API, Function B has to wait 5 seconds before it can even start. Asynchronous programming flips this paradigm. It allows your program to say, “Hey, I’m waiting for this API to respond. I’m going to pause Function A, go do some other work (like starting Function B), and come back to Function A when the data is ready.”

This is handled by the Event Loop, the heart of Python’s asyncio library.

Prerequisites for This Tutorial

Before we dive into the code, make sure you have the following set up:

1. Python 3.11 or higher: We will be using modern Python syntax. As of Python 3.11 (and continuing into 3.12, 3.13, and the upcoming 3.14/3.15 releases), the asyncio API has been significantly streamlined, particularly with the introduction of Task Groups. If you’re using an older version like 3.8 or 3.9, some of the structural code in this tutorial will throw syntax errors.
2. Basic Python Knowledge: You should understand functions, decorators, and how to run Python scripts in your terminal.
3. A Code Editor: VS Code, PyCharm, Neovim, or any editor with Python syntax highlighting.
4. An HTTP Client: We will use aiohttp for our real-world API example. You can install it via pip:
   bash
pip install aiohttp

Step-by-Step Guide to Async/Await Syntax

Let’s break down the syntax. There are two main keywords you need to know: async and await.

Defining an Async Function (Coroutines)

When you put the async keyword before def, you are no longer creating a standard function. You are creating a coroutine.

async def fetch_data():
    print("Starting to fetch data...")
    return {"data": "success"}

If you try to call this function like a normal Python function (fetch_data()), it won’t execute the print statement. Instead, it returns a coroutine object.

>>> fetch_data()
<coroutine object fetch_data at 0x10a1b2c40>

To actually run the code inside a coroutine, it must be awaited or scheduled on the event loop.

The await Keyword

The await keyword tells the event loop, “Pause this coroutine right here until the awaited operation finishes, and go run other coroutines in the meantime.” You can only use await inside a function that has been defined with async def.

Let’s look at a basic example using asyncio.sleep(), which is the async equivalent of time.sleep().

import asyncio
import time

async def make_coffee(name):
    print(f"Starting to brew {name}...")
    # Simulate the 2 seconds it takes for the coffee machine to brew
    await asyncio.sleep(2) 
    print(f"{name} is ready!")
    return name

async def main():
    start_time = time.time()

    # We await the coroutine
    await make_coffee("Espresso")

    print(f"Total time taken: {time.time() - start_time:.2f} seconds")

# The modern entry point for an async script
if __name__ == "__main__":
    asyncio.run(main())

If you run this, the output will be:

Starting to brew Espresso...
Espresso is ready!
Total time taken: 2.00 seconds

Running Multiple Tasks Concurrently

Awaiting a single coroutine doesn’t save us any time. The magic happens when we have multiple tasks. Let’s say we want to brew an Espresso and a Latte. In a synchronous world, this would take 4 seconds (2 seconds per drink). With asyncio, we can do both at the same time.

In modern Python (3.11+), the best way to run multiple coroutines concurrently is using Task Groups (asyncio.TaskGroup).

import asyncio
import time

async def make_coffee(name, brew_time):
    print(f"Starting to brew {name}...")
    await asyncio.sleep(brew_time)
    print(f"{name} is ready!")
    return name

async def main():
    start_time = time.time()

    # Using a Task Group to run coroutines concurrently
    async with asyncio.TaskGroup() as tg:
        task1 = tg.create_task(make_coffee("Espresso", 2))
        task2 = tg.create_task(make_coffee("Latte", 3))

    # The code automatically waits for all tasks in the group to finish 
    # before exiting the 'async with' block.

    print(f"Espresso result: {task1.result()}")
    print(f"Latte result: {task2.result()}")
    print(f"Total time taken: {time.time() - start_time:.2f} seconds")

if __name__ == "__main__":
    asyncio.run(main())

Output:

Starting to brew Espresso...
Starting to brew Latte...
Espresso is ready!
Latte is ready!
Espresso result: Espresso
Latte result: Latte
Total time taken: 3.00 seconds

Notice what happened? Even though the Espresso takes 2 seconds and the Latte takes 3 seconds, the total time was only 3 seconds. Because we fired them off concurrently, the event loop handled both waits simultaneously.

Real-World Use Cases for Python Async/Await

Now that we have the basics down, let’s look at how you actually use this in production code.

Use Case 1: High-Performance Web Scraping

One of the most common reasons developers seek out a python async await tutorial with examples is to speed up network requests. Synchronous libraries like requests block the main thread while waiting for the server to respond. The aiohttp library allows us to fetch dozens or hundreds of URLs concurrently.

Let’s build a quick script to fetch user data from a public dummy API (JSONPlaceholder).

import asyncio
import aiohttp
import time

async def fetch_user(session, user_id):
    """Fetches a single user from the API."""
    url = f"https://jsonplaceholder.typicode.com/users/{user_id}"

    # The 'async with' statement ensures resources are cleaned up properly
    async with session.get(url) as response:
        # We await the json() method because reading the body is also I/O
        data = await response.json()
        print(f"Fetched user: {data.get('name')}")
        return data

async def main():
    start_time = time.time()
    user_ids = range(1, 11) # Fetch users 1 through 10

    # aiohttp.ClientSession should be reused for multiple requests
    async with aiohttp.ClientSession() as session:
        async with asyncio.TaskGroup() as tg:
            for uid in user_ids:
                tg.create_task(fetch_user(session, uid))

    print(f"\nFetched 10 users in {time.time() - start_time:.2f} seconds")

if __name__ == "__main__":
    asyncio.run(main())

If you run this script synchronously using the requests library, fetching 10 users would take about 3 to 5 seconds depending on your internet latency. With aiohttp and asyncio, it happens almost instantaneously (often under 0.5 seconds).

Use Case 2: Async Database Queries with SQLAlchemy

In 2026, modern web frameworks like FastAPI, Starlette, and the latest versions of Django heavily utilize asynchronous database drivers. If you use an ORM like SQLAlchemy 2.0, you can make your database queries non-blocking.

Here is a conceptual example using asyncpg (an asynchronous PostgreSQL driver) and SQLAlchemy.

import asyncio
from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession
from sqlalchemy.orm import sessionmaker
from sqlalchemy import select, text

# Using an async PostgreSQL driver
DATABASE_URL = "postgresql+asyncpg://user:password@localhost/mydatabase"

# Create the async engine
engine = create_async_engine(DATABASE_URL)
AsyncSessionLocal = sessionmaker(engine, class_=AsyncSession, expire_on_commit=False)

async def get_users():
    """Fetches users from a database without blocking the event loop."""
    async with AsyncSessionLocal() as session:
        # Await the database execution
        result = await session.execute(text("SELECT * FROM users LIMIT 5"))
        users = result.fetchall()

        for user in users:
            print(user)

        return users

async def main():
    users = await get_users()

if __name__ == "__main__":
    asyncio.run(main())

Using async database queries ensures that your web server can handle hundreds of incoming HTTP requests while waiting for the database to return data for previous requests.

Common Pitfalls and How to Avoid Them

Async programming is incredibly powerful, but it has sharp edges. Here are the most common mistakes developers make and how to avoid them.

Pitfall 1: Blocking the Event Loop

This is the cardinal sin of asynchronous programming. If you introduce a blocking, synchronous operation inside an async def function, you freeze the entire event loop. Nothing else can run until that blocking operation finishes.

Bad:

import asyncio
import time

async def bad_blocking_task():
    print("Starting blocking task...")
    # THIS IS BAD! time.sleep() blocks the whole thread.
    time.sleep(3) 
    print("Finished blocking task.")

async def fast_task():
    print("Fast task trying to run...")
    await asyncio.sleep(0.5)
    print("Fast task finished!")

async def main():
    async with asyncio.TaskGroup() as tg:
        tg.create_task(bad_blocking_task())
        tg.create_task(fast_task())

if __name__ == "__main__":
    asyncio.run(main())

Because time.sleep(3) is synchronous, fast_task() won’t even start until the 3 seconds have passed.

The Fix:
Always use await asyncio.sleep().
If you must run a heavy synchronous CPU-bound function or use a legacy synchronous library (like Python’s standard requests), you must push it to a background thread or process using asyncio.to_thread() (int