[Python] Python asyncio not able to run the tasks

I am trying to test python asyncio and aiohttp. Idea is to fetch the data from API parallely and store the .html file in a local drive. Below is my code.

import asyncio
import aiohttp
import time
import os

url_i = "<some_urls>-"
file_path = "<local_drive>\\asynciotest"

async def download_pep(pep_number: int) -> bytes:
url = url + f"{pep_number}/"
print(f"Begin downloading {url}")
async with aiohttp.ClientSession() as session:
async with session.get(url) as resp:
content = await resp.read()
print(f"Finished downloading {url}")
return content

async def write_to_file(pep_number: int, content: bytes) -> None:
with open(os.path.join(file_path,f"{pep_number}"+'-async.html'), "wb") as pep_file:
print(f"{pep_number}_Begin writing ")
pep_file.write(content)
print(f"Finished writing")

async def web_scrape_task(pep_number: int) -> None:
content = await download_pep(pep_number)
await write_to_file(pep_number, content)

async def main() -> None:
tasks = []
for i in range(8010, 8016):
tasks.append(web_scrape_task(i))
await asyncio.wait(tasks)

if __name__ == "__main__":
s = time.perf_counter()
asyncio.run(main())
elapsed = time.perf_counter() - s
print(f"Execution time: {elapsed:0.2f} seconds.")


The above code is throwing an error

TypeError: Passing coroutines is forbidden, use tasks explicitly.
sys:1: RuntimeWarning: coroutine 'web_scrape_task' was never awaited


I am completely novish in asyncio hence not getting any clue. I have followed documentation but have not got any clue.

Am I missing here things?

Continue reading...