Async Python

What Is Async Python?

Imagine your application needs to call three APIs. Each API takes two seconds to respond.

A normal sequential program might do this:

API 1
  ↓
wait 2 seconds
  ↓
API 2
  ↓
wait 2 seconds
  ↓
API 3
  ↓
wait 2 seconds

Total ≈ 6 seconds

Async code can start the requests and allow them to wait at the same time.

API 1 ──┐
API 2 ──┼── waiting together
API 3 ──┘
   ↓
results

Total ≈ 2 seconds

The exact timing depends on the application and network, but the important idea is that the waiting periods can overlap.

Synchronous vs Asynchronous

Synchronous code generally performs one operation and waits for it to finish before moving to the next.

result1 = get_data()

result2 = get_data()

result3 = get_data()

The second call starts after the first call finishes.

Async code can allow multiple operations to make progress while they are waiting.

Synchronous

Task 1 → wait → finish
                  ↓
Task 2 → wait → finish
                  ↓
Task 3 → wait → finish


Asynchronous

Task 1 → wait ──────┐
Task 2 → wait ──────┼→ results
Task 3 → wait ──────┘

async and await

Python uses two important keywords for asynchronous programming:

• async
• await

An asynchronous function is created with async def.

async def say_hello():

    print("Hello")

The await keyword tells Python that the function may need to wait for an asynchronous operation.

async def get_data():

    result = await some_operation()

    return result

The asyncio Module

Python provides the built-in asyncio module for asynchronous programming.

import asyncio


async def main():

    print("Hello from async Python")


asyncio.run(main())

Hello from async Python

asyncio.run() starts the asynchronous program.

asyncio.sleep()

A simple way to understand async behavior is with asyncio.sleep().

import asyncio


async def main():

    print("Starting")

    await asyncio.sleep(2)

    print("Finished")


asyncio.run(main())

The program waits for two seconds, but the important point is that this is an asynchronous wait.

Running Tasks Together

Now consider two operations that each wait for two seconds.

import asyncio


async def task_one():

    print("Task one started")

    await asyncio.sleep(2)

    print("Task one finished")


async def task_two():

    print("Task two started")

    await asyncio.sleep(2)

    print("Task two finished")


async def main():

    await asyncio.gather(
        task_one(),
        task_two()
    )


asyncio.run(main())

asyncio.gather() allows the asynchronous operations to run concurrently.

Task one started
Task two started
Task one finished
Task two finished

Both tasks can spend their waiting time at the same time instead of waiting for one task to completely finish before starting the other.

Creating Async Tasks

You can explicitly create tasks with asyncio.create_task().

import asyncio


async def download_file(name):

    print(f"Downloading {name}")

    await asyncio.sleep(2)

    print(f"{name} downloaded")


async def main():

    task1 = asyncio.create_task(
        download_file("file1.txt")
    )

    task2 = asyncio.create_task(
        download_file("file2.txt")
    )

    await task1
    await task2


asyncio.run(main())

The two download operations can make progress while waiting.

Why Async Is Useful for APIs

This is where async becomes important for AI applications.

Suppose your application needs information from three different services:

Your AI application
       │
       ├── Weather API
       │
       ├── Product API
       │
       └── AI API

These requests may spend most of their time waiting for network responses.

Async programming can allow those waits to overlap.

Simulating API Requests

Before using a real API, we can simulate network requests with asyncio.sleep().

import asyncio


async def get_weather():

    print("Getting weather...")

    await asyncio.sleep(2)

    return "Sunny"


async def get_products():

    print("Getting products...")

    await asyncio.sleep(2)

    return ["Laptop", "Phone"]


async def main():

    weather, products = await asyncio.gather(
        get_weather(),
        get_products()
    )

    print(weather)
    print(products)


asyncio.run(main())

Getting weather...
Getting products...
Sunny
['Laptop', 'Phone']

Both operations can wait at the same time.

Async in an AI Application

Imagine an AI application needs to collect information before generating an answer.

User question
      ↓
 ┌───────────────┐
 │ Weather API   │
 │ Product API   │
 │ Search API    │
 └───────────────┘
      ↓
Combine results
      ↓
Send to AI model
      ↓
Generate answer

The first three operations can potentially be performed concurrently if they are independent.

import asyncio


async def get_weather():
    await asyncio.sleep(1)
    return "Sunny"


async def search_products():
    await asyncio.sleep(1)
    return ["Laptop", "Phone"]


async def get_news():
    await asyncio.sleep(1)
    return ["AI news"]


async def main():

    weather, products, news = await asyncio.gather(
        get_weather(),
        search_products(),
        get_news()
    )

    print(weather)
    print(products)
    print(news)


asyncio.run(main())

Returning Values from Async Functions

Async functions can return values just like normal functions.

import asyncio


async def get_name():

    await asyncio.sleep(1)

    return "Raj"


async def main():

    name = await get_name()

    print(name)


asyncio.run(main())

Raj

The important difference is that calling an async function normally gives you a coroutine. You use await to get its result.

What Is a Coroutine?

An async function produces a coroutine object when it is called.

async def get_data():

    return "Data"


result = get_data()

print(result)

You have not actually received the final string yet. You have a coroutine that represents the asynchronous operation.

Inside another async function, you can use:

result = await get_data()

This is why async and await normally appear together.

Error Handling in Async Code

You can use normal try and except blocks with async functions.

import asyncio


async def get_data():

    await asyncio.sleep(1)

    raise ValueError(
        "Something went wrong"
    )


async def main():

    try:

        await get_data()

    except ValueError as error:

        print(f"Error: {error}")


asyncio.run(main())

Error: Something went wrong

Async Timeouts

API calls can sometimes take too long. Async Python provides tools for handling timeouts.

import asyncio


async def slow_operation():

    await asyncio.sleep(10)

    return "Finished"


async def main():

    try:

        result = await asyncio.wait_for(
            slow_operation(),
            timeout=3
        )

        print(result)

    except asyncio.TimeoutError:

        print("Operation timed out")


asyncio.run(main())

Operation timed out

This is useful when working with external APIs where waiting forever is unacceptable.

When Should You Use Async?

Async is especially useful for I/O-bound work.

• API requests
• Network operations
• Database operations
• File operations
• WebSocket connections
• Multiple external services

But async is not automatically faster for everything.

CPU-heavy work such as large mathematical calculations or model training requires a different approach.

Async Is Not the Same as Parallel Processing

This distinction is important.

Async
→ Efficiently handles waiting
→ Excellent for I/O-bound work


Parallel processing
→ Multiple CPU operations at the same time
→ Useful for CPU-heavy work

For example, making 100 API requests can be a good async use case.

Training a large machine learning model is not made faster simply by converting the code to async.

Async + Logging

The Logging lesson becomes useful here because async applications can contain many operations happening around the same time.

import asyncio
import logging


logging.basicConfig(
    level=logging.INFO
)


logger = logging.getLogger(__name__)


async def fetch_data():

    logger.info("Fetching data")

    await asyncio.sleep(2)

    logger.info("Data received")

    return "Data"


async def main():

    result = await fetch_data()

    logger.info(
        f"Result: {result}"
    )


asyncio.run(main())

Logging makes it easier to understand what each asynchronous operation is doing.

Mini Project — Async AI Data Collector

Build a small program that collects information from three simulated services at the same time.

import asyncio
import logging


logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s - %(levelname)s - %(message)s"
)


logger = logging.getLogger(__name__)


async def get_weather():

    logger.info("Getting weather")

    await asyncio.sleep(2)

    return "Sunny"


async def get_products():

    logger.info("Getting products")

    await asyncio.sleep(2)

    return [
        "Laptop",
        "Phone"
    ]


async def get_news():

    logger.info("Getting news")

    await asyncio.sleep(2)

    return [
        "AI is growing"
    ]


async def main():

    logger.info(
        "Starting data collection"
    )

    weather, products, news = await asyncio.gather(
        get_weather(),
        get_products(),
        get_news()
    )

    logger.info(
        "Data collection completed"
    )

    print("Weather:", weather)
    print("Products:", products)
    print("News:", news)


asyncio.run(main())

Weather: Sunny
Products: ['Laptop', 'Phone']
News: ['AI is growing']

This is a simplified version of a pattern used by applications that collect data from multiple external services before sending that data to an AI model.

What You Learned

• What asynchronous programming means
• Synchronous vs asynchronous execution
• async def
• await
• The asyncio module
• asyncio.run()
• asyncio.gather()
• asyncio.create_task()
• Coroutines
• Async error handling
• Async timeouts
• Using async for API operations
• Using async in AI applications
• Why async is mainly useful for I/O-bound work
• Why async is not the same as parallel CPU processing