Python asyncio in der Lage, die Aufgaben nicht auszuführen ⇐ Python

[Guest]

Ich versuche, Python asyncio und aiohttp zu testen. Die Idee besteht darin, die Daten von API parallel zu holen und die .html -Datei in einem lokalen Laufwerk zu speichern. Unten ist mein Code. < /P>

Code: Select all

import asyncio
import aiohttp
import time
import os

url_i = "-"
file_path = "\\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.")
< /code>
Der obige Code wirft einen Fehler < /p>
TypeError: Passing coroutines is forbidden, use tasks explicitly.
sys:1: RuntimeWarning: coroutine 'web_scrape_task' was never awaited

Ich bin in Asyncio daher keine Ahnung. Ich habe Dokumentation verfolgt, habe aber keine Ahnung bekommen.
Fehlt mir hier Dinge?