Asynchronous Programming in Python

In asynchronous programming, waiting is not a pause — it’s an opportunity.

In this blog, we will cover a basic introduction to Asynchronous Programming, hands-on asynchronous programming, and related tasks.

Synchronous Programming vs Asynchronous Programming:

In synchronous programming, tasks are executed one after another. Each task must be completed before starting the next task. If one task takes a long time, the entire program stops and waits until that task finishes.

In asynchronous programming, multiple tasks can run without blocking each other. When one task is waiting, the program can start another task instead of staying idle.

How Asynchronous Programming is achieved in Python?

Asynchronous Programming is achieved in Python using the asyncio library with async and await keywords.

What are coroutines?

Coroutines are special functions that can pause and resume their execution. In Python, coroutines are the building blocks of asynchronous programming. They allow a program to handle multiple tasks efficiently without blocking execution.

Async and Await Keywords:

async is used to define a coroutine, and await is used to pause a coroutine until the awaited task is completed.

Installation

To use the asyncio library, we first need to install it using the below command:

pip install asyncio 

Usage

After Installation, we need to import it using the below command:

import asyncio

Hands-on:

1. Async sleep demo:

import time
import asyncio
#  async first function
async def first():
    print("Started function 1")
    await asyncio.sleep(2)
    print("Ended function 1")

#  async second function
async def second():
    print("Started function 2")
    await asyncio.sleep(1)
    print("Ended function 2")
    
# async main function
async def main():
    await asyncio.gather(first(),second())

# starting async event loop
asyncio.run(main())

2. Async file reading:

Get Jaspinder kaur Walia’s stories in your inbox

Join Medium for free to get updates from this writer.

Remember me for faster sign in

For asynchronous file reading, we need to use the aiofiles library.

import asyncio
import aiofiles

# async function
async def read_file(file_name):
    async with aiofiles.open(file_name,'r') as file:
        contents=await file.read()  
        print(contents)
    
# starting async event loop
asyncio.run(read_file('asynchronous.txt'))

3. Fetch multiple APIs asynchronously:

For asynchronous requests, we use the aiohttp library in Python. To use this library, we need to first install it and then need to import it in the program.

import asyncio
import aiohttp

#async function
async def fetch_data(session, url):
    print(f"Starting task: {url}")
    try:
        async with session.get(url) as response:
            data = await response.json()
            print(f"Completed task: {url}")
            return data
    except aiohttp.ClientError:
        print("Fetching error from api")
    
# async main function   
async def main():
    urls=['https://api.potterdb.com/','https://api.dictionaryapi.dev/api/v2/entries/en/digital','http://api.football-data.org/v4/competitions/']

    async with aiohttp.ClientSession() as session:
        tasks = [fetch_data(session, url) for url in urls]
        results = await asyncio.gather(*tasks)
        for result in results:
            print(result)
    
# starting async event loop
asyncio.run(main())

Mini Task:

Build an async script that hits 5 APIs concurrently and logs response times

import time
import asyncio
import aiofiles
import aiohttp

# async function
async def fetch_read(session,url):
    #setting start time
    start_time=time.time()
    async with session.get(url) as response:
        # getting response
        data =await response.json()
        # end time
        end_time=time.time()
        # finding duration
        duration=end_time-start_time
        print(f"{url},Duration: {duration}")
    
# async main function
async def main():
        urls=['https://api.potterdb.com/','https://api.dictionaryapi.dev/api/v2/entries/en/digital','http://api.football-data.org/v4/competitions/','https://www.affirmations.dev/','https://softwium.com/api/books']
        try:
            async with aiohttp.ClientSession() as session:
                tasks=[fetch_read(session,url) for url in urls ]
                result=await asyncio.gather(*tasks)
        except:
            print("fetching error from api")

# starting async event loop
asyncio.run(main())