Mastering asyncio and Background Tasks

Python’s asyncio is a powerful library for writing concurrent code using the async/await syntax. It’s great for I/O-bound operations like network calls, database access, or file I/O—basically anything where your program waits on something.

What is asyncio?

asyncio lets you write concurrent code in Python using the async/await syntax. This is perfect when you want to run I/O-bound tasks like:

• Downloading web pages
• Waiting on network data
• Reading from disk

It helps you run multiple things at once without using threads!

Basics

A coroutine is a special type of subroutine that can pause and resume its execution, allowing for cooperative multitasking.

async and await

import asyncio

async def say_hello():
    print("Hello")
    await asyncio.sleep(1) #is used to simulate a long wait(any task)
    print("World")

# Run the coroutine
asyncio.run(say_hello())

Key Points

• async def defines an asynchronous function.
• await tells Python to pause and wait for the result without blocking other tasks.
• asyncio.sleep()
  • It an async version of time.sleep() that doesn’t block the program.
  • is used to simulate a long wait (any task)
• asyncio.run(): Runs the coroutine inside an event loop.

Hello
<waits 1 second>
World

Running Multiple Coroutines Concurrently

async def main():
    await asyncio.gather(task1(), task2())

asyncio.gather is a function in Python’s asyncio library used to run multiple asynchronous coroutines concurrently. It schedules them to run at the same time and waits for all of them to complete.

asyncio.gather() and asyncio.run()

asyncio.run()

• Starts the event loop.
• Used to run one top-level async def function from sync code.
• You use it once to kick things off.

asyncio.gather()

• Runs multiple async tasks concurrently.
• Used inside an async def function.
• Returns results in the order of tasks passed.

import asyncio

async def task(name, delay):
    await asyncio.sleep(delay)
    print(f"Task {name} done after {delay}s")

async def main():
    await asyncio.gather(task("A", 2),task("B", 1),task("C", 3))

asyncio.run(main())  

Create a task and let it run in the background

Example 1

import asyncio

async def background_task(name, delay):
    print(f"Start task {name}")
    await asyncio.sleep(delay)
    print(f"{name} finished after {delay}s")

async def main():
    task1 = asyncio.create_task(background_task("Task1", 2))
    task2 = asyncio.create_task(background_task("Task2", 1))

    print("Both tasks started")
    await task1
    await task2

    #or
    #await asyncio.gather(t1, t2)


asyncio.run(main())

Output

Both tasks started
Start task Task1
Start task Task2
Task2 finished after 1s
Task1 finished after 2s

Key Points

• create_task() schedules a coroutine to run in the background.
• It returns a Task object, which you can await later to get results.
• This is useful when you want to start something now and wait for it later.

print(“Both tasks started”) executed even before the task 1 and task2 completes, this is a power of asyncio

Play with async

In above code

Case 1 Comment both await task1 and await task 2

#await task1
#await task2


#output
Both tasks started
Start task Task1
Start task Task2

Program closed with waitaing to finish the task

Case 2: Comment await task2

await task1
#await task2


#Output 
Both tasks started
Start task Task1
Start task Task2
Task2 finished after 1s

Even If task 2 is comented, we got it output; Because we are waiting for task1 and task 1 took longer time than t2

Case 3 : Comment await task 1

#await task1
await task2
    
#Output
Both tasks started
Start task Task1
Start task Task2
Task2 finished after 1s

Unlink case 2, here we have not got result from task 1, as task2 finishes before task1 and we are not waiting for task 1

So always use awiting

Synchronous Vs asynchronous

Example 1

Synchronous Code

import time

def download_file(filename):
    print(f"Starting download: {filename}")
    time.sleep(2)  # Simulate a 2-second blocking I/O operation
    print(f"Finished download: {filename}")

def main():
    download_file("file1.txt")
    download_file("file2.txt")

main()

Output

Starting download: file1.txt
Finished download: file1.txt
Starting download: file2.txt
Finished download: file2.txt

Total time: ~4 seconds (2 seconds per file, sequentially)

Asynchronous Code Example

import asyncio

async def download_file(filename):
    print(f"Starting download: {filename}")
    await asyncio.sleep(2)  # Simulate a 2-second non-blocking I/O operation
    print(f"Finished download: {filename}")

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

Starting download: file1.txt
Starting download: file2.txt
Finished download: file1.txt
Finished download: file2.txt

Total time: ~2 seconds (both downloads happen concurrently)

Key Points

• Synchronous code blocks the program until the task is done.
• Asynchronous code allows other tasks to run while waiting (e.g., during I/O).

Example 2 : HTTP requests

• requests (synchronous)
• aiohttp (asynchronous)

Using requests (Synchronous)

import requests
import time

def fetch(url):
    print(f"Fetching {url}")
    response = requests.get(url)
    print(f"Done: {url} -> Status: {response.status_code}")

def main():
    urls = [
        "https://httpbin.org/delay/2",
        "https://httpbin.org/delay/2",
        "https://httpbin.org/delay/2"
    ]
    start = time.time()
    for url in urls:
        fetch(url)
    print(f"Total time: {time.time() - start:.2f} seconds")

main()

Each URL has a 2-second delay → total time will be around 6 seconds.

Using aiohttp (Asynchronous)

import aiohttp
import asyncio
import time

async def fetch(session, url):
    print(f"Fetching {url}")
    async with session.get(url) as response:
        print(f"Done: {url} -> Status: {response.status}")

async def main():
    urls = [
        "https://httpbin.org/delay/2",
        "https://httpbin.org/delay/2",
        "https://httpbin.org/delay/2"
    ]
    start = time.time()
    async with aiohttp.ClientSession() as session:
        tasks = [fetch(session, url) for url in urls]
        await asyncio.gather(*tasks)
    print(f"Total time: {time.time() - start:.2f} seconds")

asyncio.run(main())

Even though each request has a 2-second delay, they happen concurrently, so total time is around 2 seconds.

Is it possible for a regular (non-async) function to call an asynchronous (async) function

Yes, but with a catch — a normal (sync) function can’t directly await an async function, because await only works inside an async def function.

Problem Example (❌ Wrong)

async def say_hello():
    print("Hello")

def normal_func():
    # ❌ This will cause a syntax error!
    await say_hello()

This gives an error:

SyntaxError: 'await' outside async function

Correct Way: Run it Using asyncio.run() (✅)

import asyncio

async def say_hello():
    print("Hello")

def normal_func():
    asyncio.run(say_hello())  # This works!

normal_func()

Note: asyncio.run() is used only once — usually at the top-level (like if name == “main“). Don’t call it repeatedly inside deeply nested sync code.

Communicate and coordinate

Way 1 :Shared Variables (with care)

You can use shared objects like lists, dicts, or classes — just like in sync code — but you have to avoid race conditions.

shared_data = []

async def producer():
    shared_data.append("data")

async def consumer():
    while not shared_data:
        await asyncio.sleep(0.1)
    print("Got:", shared_data.pop())

Way 2: asyncio.Queue (Best way to communicate between tasks)

Built-in, thread-safe queue for async use — perfect for producer-consumer models.

import asyncio

async def producer(queue):
    for i in range(3):
        await queue.put(f"item {i}")
        print(f"Produced item {i}")
        await asyncio.sleep(1)

async def consumer(queue):
    while True:
        item = await queue.get()
        print(f"Consumed {item}")
        queue.task_done()

async def main():
    queue = asyncio.Queue()
    await asyncio.gather(
        producer(queue),
        consumer(queue)
    )

asyncio.run(main())

Way 3: asyncio.Event

Used to signal between coroutines, like a flag.

import asyncio

event = asyncio.Event()

async def waiter():
    print("Waiting for event...")
    await event.wait()
    print("Event received!")

async def setter():
    await asyncio.sleep(2)
    print("Setting event.")
    event.set()

asyncio.run(asyncio.gather(waiter(), setter()))

Way 4: asyncio.Lock / Semaphore

Used when async tasks share resources and must prevent race conditions.

Real-life Long-Running Async Tasks

Here are some common ones where async shines:

If your “long task” is CPU-bound (like image recognition or large data processing), async is not always the best fit — you’d use multi-threading or multi-processing instead.

Conclusion

Python’s asyncio module empowers developers to write highly concurrent code using the async/await syntax, making it easier to manage I/O-bound operations efficiently. Whether you’re building web servers, network clients, or handling a large number of simultaneous tasks, asyncio provides a solid foundation for scalable and performant applications.

By understanding the core concepts like event loops, coroutines, tasks, and futures, you can take full advantage of asynchronous programming in Python. While there may be a learning curve, especially when transitioning from traditional synchronous code, the performance benefits and cleaner structure are well worth the effort.

As Python continues to evolve, asynchronous programming is becoming an essential tool in every developer’s toolkit — and asyncio is at the heart of it

Resource

• Thread and async