Отправка запросов по HTTP с помощью Python
Отправка HTTP-запросов на Python: обзор методов
Как наиболее просто и эффективно отправить HTTP-запрос на Python?
Библиотека requests
Для выполнения HTTP-запросов в Python самой популярной и удобной является библиотека requests. Она предоставляет высокоуровневый интерфейс для работы с HTTP, автоматически обрабатывает кодировку, поддерживает сессии, cookies, таймауты и многое другое.
Установка:
pip install requestsPython http запрос (отправка http-запроса на python)
Пример GET-запроса:
import requests
response = requests.get('https://httpbin.org/get')
print(response.status_code)
print(response.text)Результат:
200
{
"args": {},
"headers": {
"Accept": "*/*",
"Accept-Encoding": "gzip, deflate",
"Host": "httpbin.org",
"User-Agent": "python-requests/2.31.0",
"X-Amzn-Trace-Id": "..."
},
"origin": "...",
"url": "https://httpbin.org/get"
}Для отправки POST-запроса с JSON-данными:
import requests
url = 'https://httpbin.org/post'
data = {'key': 'value'}
response = requests.post(url, json=data)
print(response.json())Результат:
{'args': {}, 'data': '{"key": "value"}', 'files': {}, 'form': {}, 'headers': {...}, 'json': {'key': 'value'}, 'url': 'https://httpbin.org/post'}Пояснения: метод get() возвращает объект Response, из которого можно получить статус, заголовки, тело. Параметр json в POST автоматически сериализует словарь в JSON и устанавливает заголовок Content-Type. Библиотека также поддерживает передачу параметров через params, заголовков через headers, таймаут через timeout.
Типичные проблемы и их решение:
- ImportError: No module named 'requests' - библиотека не установлена. Решение: выполнить
pip install requests. - SSLError - ошибка проверки SSL-сертификата. Можно отключить проверку параметром
verify=False, но это небезопасно. Лучше установить корневые сертификаты:pip install certifi. - Timeout - превышение времени ожидания. Установите явно таймаут:
timeout=5(секунд). - Кодировка - если текст ответа содержит не-ASCII символы, requests автоматически определяет кодировку, но можно принудительно установить
response.encoding = 'utf-8'.
Как выполнить HTTP-запрос с помощью стандартной библиотеки, не устанавливая внешних пакетов?
Модуль urllib.request
В стандартной библиотеке Python есть модуль urllib.request, который позволяет отправлять HTTP-запросы без дополнительных зависимостей. Он более низкоуровневый и требует ручной обработки многих аспектов.
Пример GET-запроса:
from urllib.request import urlopen
from urllib.error import URLError, HTTPError
try:
with urlopen('https://httpbin.org/get') as response:
print(response.status)
print(response.read().decode('utf-8'))
except HTTPError as e:
print(f'HTTP Error: {e.code} - {e.reason}')
except URLError as e:
print(f'URL Error: {e.reason}')Результат:
200
{
"args": {},
...
}POST-запрос с передачей данных в теле:
from urllib.request import Request, urlopen
from urllib.parse import urlencode
url = 'https://httpbin.org/post'
data = urlencode({'key': 'value'}).encode('utf-8')
req = Request(url, data=data, method='POST')
req.add_header('Content-Type', 'application/x-www-form-urlencoded')
with urlopen(req) as response:
print(response.read().decode('utf-8'))Результат:
{"args": {}, "data": "", "form": {"key": "value"}, ...}Пояснения: Используется urlopen для открытия соединения. Для POST необходимо создать объект Request с указанием данных и метода. Данные нужно предварительно закодировать с помощью urlencode и преобразовать в байты. Обработка ошибок выполняется через исключения HTTPError и URLError.
Типичные проблемы:
- Байтовый ответ - ответ приходит в виде байтов, требуется декодирование. Всегда используйте
.decode()с правильной кодировкой. - Сложность с JSON - чтобы получить JSON, нужно вызвать
json.loads(response.read().decode()). - Обработка редиректов - по умолчанию
urlopenне следует редиректам, можно использоватьHTTPRedirectHandler. - Таймаут - не устанавливается по умолчанию, может привести к зависанию. Всегда передавайте параметр
timeoutвurlopen.
Какой современный HTTP-клиент поддерживает HTTP/2 и асинхронность, оставаясь удобным?
Библиотека httpx
httpx - это современная библиотека, которая предоставляет синхронный и асинхронный API, поддерживает HTTP/1.1 и HTTP/2, а также множество возможностей, аналогичных requests, но с большей производительностью.
Установка:
pip install httpxСинхронный GET-запрос:
import httpx
response = httpx.get('https://httpbin.org/get')
print(response.status_code)
print(response.text)Асинхронный GET-запрос:
import asyncio
import httpx
async def fetch():
async with httpx.AsyncClient() as client:
response = await client.get('https://httpbin.org/get')
return response
response = asyncio.run(fetch())
print(response.text)Пояснения: Для асинхронного использования нужен контекстный менеджер AsyncClient. Запрос выполняется через await. httpx также поддерживает передачу JSON с помощью параметра json, как и requests.
Типичные проблемы:
- Установка httpx с поддержкой HTTP/2 - по умолчанию HTTP/2 не включается. Для его активации установите
pip install httpx[http2]. - Ошибка при асинхронном запуске вне цикла событий - используйте
asyncio.run()или работайте внутри асинхронной функции. - Проблемы с прокси - httpx использует переменные окружения, но можно указать явно:
proxies={'https://': 'http://localhost:8080'}.
Как выполнить множество HTTP-запросов параллельно с максимальной производительностью?
Библиотека aiohttp
aiohttp - это асинхронная библиотека для HTTP-клиента и сервера. Она специально разработана для высоконагруженных приложений, где требуется много параллельных запросов.
Установка:
pip install aiohttpПример одиночного GET-запроса:
import aiohttp
import asyncio
async def main():
async with aiohttp.ClientSession() as session:
async with session.get('https://httpbin.org/get') as response:
print(response.status)
data = await response.text()
print(data)
asyncio.run(main())Пример параллельных запросов:
import aiohttp
import asyncio
async def fetch(session, url):
async with session.get(url) as response:
return await response.text()
async def main():
urls = ['https://httpbin.org/get'] * 5
async with aiohttp.ClientSession() as session:
tasks = [fetch(session, url) for url in urls]
results = await asyncio.gather(*tasks)
for res in results:
print(len(res))
asyncio.run(main())Пояснения: Создается сессия ClientSession, которая управляет соединениями. Каждый запрос выполняется в контексте session.get(). Для параллельного выполнения используется asyncio.gather. Сессию следует переиспользовать для нескольких запросов, чтобы не открывать новые соединения.
Типичные проблемы:
- Необходимость управлять сессией - сессия должна быть создана один раз и закрыта после использования. Используйте
async with. - Ограничение на количество одновременных соединений - по умолчанию
ClientSessionимеет лимит 100 соединений. Можно увеличить черезconnector = aiohttp.TCPConnector(limit=500). - Ошибки при отсутствии asyncio - код должен выполняться в асинхронном контексте. Нельзя вызывать асинхронные функции из синхронного кода без
asyncio.run().
Расширенные примеры и нестандартные сценарии
Далее приведены более сложные варианты использования HTTP-клиентов на Python, включая управление сессиями, загрузку файлов, работу через прокси, повторные попытки и асинхронные ограничения.
Как использовать сессию requests для сохранения cookies и заголовков?
import requests
session = requests.Session()
session.headers.update({'User-Agent': 'my-app/1.0'})
# Первый запрос (устанавливает cookie)
response = session.get('https://httpbin.org/cookies/set?name=value')
print(response.cookies)
# Второй запрос (cookie отправляется автоматически)
response = session.get('https://httpbin.org/cookies')
print(response.json()){'cookies': {'name': 'value'}}
Сессия автоматически сохраняет и отправляет cookies, а также управляет пулом соединений. Это полезно для имитации поведения браузера.
Как отправить файл через multipart/form-data с помощью requests?
import requests
url = 'https://httpbin.org/post'
files = {'file': ('test.txt', b'Hello World', 'text/plain')}
response = requests.post(url, files=files)
print(response.json()['files']){'file': 'Hello World'}Метод files позволяет передавать один или несколько файлов. Кортеж содержит имя файла, содержимое и MIME-тип. Поддерживаются также открытые файловые объекты.
Как настроить прокси с аутентификацией в requests?
import requests
proxies = {
'http': 'http://user:password@proxy.example.com:8080',
'https': 'http://user:password@proxy.example.com:8080'
}
response = requests.get('https://httpbin.org/ip', proxies=proxies)
print(response.json()){'origin': 'IP через прокси'}Прокси указываются в словаре с протоколами. Если требуется аутентификация, логин и пароль включаются в URL.
Как реализовать автоматические повторные попытки при ошибках сервера?
import requests
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry
session = requests.Session()
retries = Retry(total=3, backoff_factor=0.5, status_forcelist=[500, 502, 503, 504])
session.mount('https://', HTTPAdapter(max_retries=retries))
try:
response = session.get('https://httpbin.org/status/500')
response.raise_for_status()
except requests.exceptions.RetryError as e:
print(f'Ошибка после всех попыток: {e}')Ошибка после всех попыток: ...
Механизм retry использует стратегию с экспоненциальной задержкой (backoff_factor). Можно настроить, какие коды статуса считать ошибкой и сколько попыток делать.
Как ограничить количество параллельных запросов в асинхронном httpx?
import asyncio
import httpx
from asyncio import Semaphore
sem = Semaphore(3)
async def fetch(client, url):
async with sem:
response = await client.get(url)
return response.status_code
async def main():
urls = ['https://httpbin.org/delay/1'] * 10
async with httpx.AsyncClient() as client:
tasks = [fetch(client, url) for url in urls]
results = await asyncio.gather(*tasks)
print(results)
asyncio.run(main())[200, 200, 200, 200, 200, 200, 200, 200, 200, 200]
Использование Semaphore позволяет контролировать степень параллелизма. В данном примере одновременно выполняется не более 3 запросов, что предотвращает перегрузку сервера.
Как настроить лимит соединений в aiohttp с помощью TCPConnector?
import aiohttp
import asyncio
connector = aiohttp.TCPConnector(limit=10, limit_per_host=5, ttl_dns_cache=300)
async def main():
async with aiohttp.ClientSession(connector=connector) as session:
async with session.get('https://httpbin.org/get') as resp:
print(await resp.text())
asyncio.run(main()){...} (стандартный ответ)Параметры коннектора позволяют ограничить общее количество соединений (limit) и количество соединений на один хост (limit_per_host), а также настроить кэширование DNS.