33 lines
1 KiB
Python
33 lines
1 KiB
Python
import asyncio
|
|
from datetime import datetime
|
|
import random
|
|
import httpx
|
|
|
|
|
|
async def fetch(url: str, client: httpx.AsyncClient, sem: asyncio.Semaphore) -> dict:
|
|
async with sem:
|
|
try:
|
|
print(f"Fetching of {url} started at {datetime.today()}")
|
|
r = await client.get(url)
|
|
data = r.json()
|
|
print(f"Fetching of {url} ended at {datetime.today()}")
|
|
return data
|
|
except httpx.HTTPStatusError as e:
|
|
print(f"HTTP error: {e}")
|
|
except httpx.RequestError as e:
|
|
print(f"Request failed: {e}")
|
|
|
|
async def main():
|
|
urls = [f"https://jsonplaceholder.typicode.com/posts/{n}" for n in range(1, 15)]
|
|
sem = asyncio.Semaphore(5)
|
|
tasks: list[asyncio.Task] = []
|
|
|
|
async with httpx.AsyncClient(timeout=10) as client:
|
|
async with asyncio.TaskGroup() as tg:
|
|
for url in urls:
|
|
tasks.append(tg.create_task(fetch(url, client, sem)))
|
|
|
|
print([task.result() for task in tasks])
|
|
|
|
if __name__ == "__main__":
|
|
asyncio.run(main())
|