Pythoner
Фотография
✈️ Асинхронное использование httpx. В Python httpx предоставляет мощный и удобный асинхронный интерфейс для работы с HTTP-запросами. Асинхронный режим позволяет эффективно обрабатывать множество запросов параллельно, не блокируя основной поток выполнения, что особенно полезно при взаимодействии с медленными или перегруженными серверами.
➡️ Применение асинхронного httpx идеально подходит для ситуаций, где требуется высокая производительность при работе с сетью, например, при веб-скрапинге, взаимодействии с API, загрузке файлов и обработке потоковых данных. Асинхронный подход позволяет максимально использовать ресурсы системы и сократить время ожидания ответов от сервера.
➡️ Вот примеры, демонстрирующие асинхронную работу с httpx:
import asyncio
import httpx
async def fetch_page(client: httpx.AsyncClient, url: str):
try:
response = await client.get(url, timeout=10.0)
response.raise_for_status()
return {"url": url, "status_code": response.status_code, "content": response.text}
except httpx.HTTPError as exc:
return {"url": url, "error": f"HTTP Error: {exc}"}
except httpx.TimeoutException:
return {"url": url, "error": "Request timed out"}
except Exception as e:
return {"url": url, "error": f"An unexpected error occurred: {e}"}
async def main():
urls = [
"https://www.example.com",
"https://www.google.com",
"https://www.python.org",
"https://this-is-an-invalid-url.com", # Пример некорректного URL
"https://jsonplaceholder.typicode.com/todos/1" # Пример API
]
async with httpx.AsyncClient(http2=True) as client: # Использование HTTP/2
tasks = [fetch_page(client, url) for url in urls]
results = await asyncio.gather(*tasks)
for result in results:
print(result)
async def make_requests_with_limits(urls, max_concurrent=5):
semaphore = asyncio.Semaphore(max_concurrent)
async def fetch_with_semaphore(url):
async with semaphore:
return await fetch_page(httpx.AsyncClient(), url)
async with httpx.AsyncClient() as client:
tasks = [fetch_with_semaphore(url) for url in urls]
results = await asyncio.gather(*tasks)
for result in results:
print(result)
if __name__ == "__main__":
asyncio.run(main())
# Пример с ограничением числа одновременных запросов
asyncio.run(make_requests_with_limits(["https://www.example.com"] * 10))
⬆️ В этом примере fetch_page теперь принимает httpx.AsyncClient в качестве аргумента, что позволяет использовать один клиент для всех запросов внутри main. Это способствует более эффективному использованию соединений. Добавлен пример с некорректным URL и API-запросом для демонстрации обработки ошибок и различных типов ответов. Также показано, как использовать HTTP/2 с помощью httpx.AsyncClient(http2=True). Дополнительно, представлен пример make_requests_with_limits с использованием asyncio.Semaphore для ограничения количества одновременных запросов, что позволяет контролировать нагрузку на сервер и предотвращать ошибки, связанные с превышением лимитов.
🐍 Pythoner
33 · 1.9K ·