53 lines
1.8 KiB
Python
53 lines
1.8 KiB
Python
import time
|
|
import logging
|
|
import httpx
|
|
import asyncio
|
|
from typing import Any
|
|
|
|
logging.basicConfig(
|
|
level=logging.INFO, format="%(asctime)s [%(levelname)s] %(message)s"
|
|
)
|
|
|
|
logger = logging.getLogger(__name__)
|
|
logging.getLogger("httpx").setLevel(logging.CRITICAL)
|
|
|
|
|
|
async def fetch_user(client: httpx.AsyncClient, id: int) -> dict[str, Any]:
|
|
try:
|
|
response = await client.get(f"https://jsonplaceholder.typicode.com/users/{id}")
|
|
response.raise_for_status()
|
|
return response.json()
|
|
except httpx.HTTPStatusError:
|
|
logger.error(f"User with id {id} not found.")
|
|
return {}
|
|
|
|
|
|
# toClaude: VSC подсвечивает красным fetch_all_async, потому что возвращаемый list[dict] частично Unknown
|
|
async def fetch_all_async(ids: list[int]) -> list[dict[str, Any]]:
|
|
start_time = time.time()
|
|
async with httpx.AsyncClient() as client:
|
|
tasks = [fetch_user(client, id) for id in ids]
|
|
result = await asyncio.gather(*tasks)
|
|
logger.info(f"Async completed for {time.time() - start_time:.2f}s")
|
|
return result
|
|
|
|
|
|
def fetch_all_sync(ids: list[int]) -> list[dict[str, Any]]:
|
|
start_time = time.time()
|
|
result: list[dict[str, Any]] = []
|
|
with httpx.Client() as client:
|
|
for id in ids:
|
|
response = client.get(f"https://jsonplaceholder.typicode.com/users/{id}")
|
|
response.raise_for_status()
|
|
result.append(response.json())
|
|
logger.info(f"Sync completed for {time.time() - start_time:.2f}s")
|
|
return result
|
|
|
|
|
|
if __name__ == "__main__":
|
|
user_ids = [1, 2, 3, 4, 5, 6, 7, 8, 9, 10]
|
|
async_result = asyncio.run(fetch_all_async(user_ids))
|
|
sync_result = fetch_all_sync(user_ids)
|
|
|
|
print("Async:", len(async_result), "records")
|
|
print("Sync:", len(sync_result), "records")
|