Использование urllib3 для выполнения HTTP запросов в Python
Основные возможности urllib3
Библиотека urllib3 предоставляет мощный и удобный интерфейс для работы с HTTP в Python. Основной класс для управления соединениями - PoolManager. Он автоматически управляет пулом TCP-соединений, поддерживает keep-alive, повторные попытки и множество других возможностей. Ниже представлен базовый пример выполнения GET-запроса:
import urllib3
http = urllib3.PoolManager()
response = http.request('GET', 'https://httpbin.org/get')
print(response.status)
print(response.data.decode('utf-8'))
запрос страницы python (запрос веб-страницы в python)
В данном примере создается менеджер пула, выполняется запрос, выводится статус и тело ответа. Для POST-запроса с данными:
import urllib3
http = urllib3.PoolManager()
response = http.request('POST', 'https://httpbin.org/post', fields={'key': 'value'})
print(response.json())
Python requests (библиотека requests в python)
Класс PoolManager можно использовать как контекстный менеджер для гарантированного закрытия соединений.
Типичные проблемы:
- Утечка соединений - если не закрывать PoolManager вручную. Решение: использовать with или вызывать .clear().
- Ошибка SSL - при самоподписанных сертификатах. Решение: создать свой SSLContext или передать cert_reqs='CERT_NONE'.
- Кодировка ответа - данные могут быть сжаты (gzip). urllib3 автоматически декодирует, если заголовок Accept-Encoding установлен.
Как выполнить GET запрос с параметрами строки запроса?
Параметры передаются в аргументе fields. urllib3 автоматически кодирует их:
import urllib3
http = urllib3.PoolManager()
params = {'q': 'python urllib3', 'page': 1}
response = http.request('GET', 'https://httpbin.org/get', fields=params)
print(response.data.decode('utf-8'))
Python urllib3 request (использование urllib3 для запросов)
Проблема:
Если параметры содержат специальные символы, они будут корректно закодированы. Однако некорректный URL может вызвать ошибку. Рекомендуется проверять URL перед передачей.
Как отправить данные в формате JSON?
Для отправки JSON используется аргумент body с явной установкой заголовка Content-Type:
import urllib3
import json
http = urllib3.PoolManager()
data = {'name': 'Alice', 'age': 30}
encoded_data = json.dumps(data).encode('utf-8')
response = http.request('POST', 'https://httpbin.org/post',
body=encoded_data,
headers={'Content-Type': 'application/json'})
print(response.json())
Http error requests python (обработка http ошибок в requests)
Типичная ошибка:
Забывают указать заголовок Content-Type - сервер может неправильно интерпретировать тело запроса. Другая ошибка - передача строки вместо байтов.
Как обработать ошибки соединения и статус коды HTTP?
Для обработки ошибок сети используется блок try/except с исключениями urllib3.exceptions. Для проверки статуса кода - свойство status:
import urllib3
http = urllib3.PoolManager()
try:
response = http.request('GET', 'https://httpbin.org/status/404')
if response.status == 200:
print('Успех')
else:
print(f'Статус: {response.status}')
except urllib3.exceptions.HTTPError as e:
print(f'Ошибка HTTP: {e}')
except urllib3.exceptions.MaxRetryError as e:
print(f'Превышено количество попыток: {e}')
Make request python (выполнение http-запроса)
Проблема:
Некоторые ошибки, такие как таймаут, могут не быть явно обработаны. Следует всегда ожидать исключения ConnectTimeoutError и ReadTimeoutError.
Как автоматически повторять запросы при временных ошибках?
urllib3 имеет встроенный механизм повторных попыток через класс Retry:
import urllib3
from urllib3.util.retry import Retry
retry_strategy = Retry(
total=3,
backoff_factor=0.5,
status_forcelist=[500, 502, 503, 504]
)
http = urllib3.PoolManager(retries=retry_strategy)
response = http.request('GET', 'https://httpbin.org/status/500')
print(response.status)
Python request files file name (отправка файла через requests)
Этот код автоматически повторит запрос до 3 раз с возрастающей задержкой при получении серверных ошибок.
Проблема:
Неправильный backoff_factor может привести к слишком длительному ожиданию. Рекомендуется выбирать значение от 0.1 до 1.
Как отправить файл через multipart/form-data?
Для отправки файлов используется urllib3.filepost.encode_multipart_formdata:
import urllib3
from urllib3.filepost import encode_multipart_formdata
http = urllib3.PoolManager()
with open('test.txt', 'r') as f:
data = f.read()
fields = {
'file': ('test.txt', data, 'text/plain'),
'description': 'Пример файла'
}
body, content_type = encode_multipart_formdata(fields)
response = http.request('POST', 'https://httpbin.org/post',
body=body,
headers={'Content-Type': content_type})
print(response.json())
Python request params (параметры запроса)
Типичная ошибка:
Неправильный формат кортежа для файла. Должен быть кортеж из (имя файла, данные, content-type). Если content-type не указан, может быть использовано 'application/octet-stream'.
Как выполнить запрос через HTTP/HTTPS прокси?
Для работы через прокси используется класс ProxyManager:
import urllib3
proxy_url = 'http://user:pass@proxy.example.com:8080'
proxy = urllib3.ProxyManager(proxy_url, cert_reqs='CERT_NONE')
response = proxy.request('GET', 'https://httpbin.org/get')
print(response.data.decode('utf-8'))
Python request download (скачивание файла через requests)
Проблема:
Прокси может требовать аутентификации. В таком случае данные передаются в URL. Для HTTPS прокси необходимо указать полный URL.
Как установить таймауты на соединение и чтение?
Таймауты задаются через класс Timeout:
import urllib3
from urllib3.util.timeout import Timeout
timeout = Timeout(connect=2.0, read=7.0)
http = urllib3.PoolManager(timeout=timeout)
response = http.request('GET', 'https://httpbin.org/delay/6')
print(response.status)
Если запрос не успевает за указанное время, будет выброшено исключение ReadTimeoutError.
Проблема:
Неправильно указанные таймауты могут привести к ложным срабатываниям. Рекомендуется устанавливать таймауты с запасом для медленных сетей.
Расширенные примеры использования urllib3
Потоковая загрузка большого файла (stream=True)
Для работы с большими файлами без загрузки всего содержимого в память используется параметр preload_content=False и чтение потока:
import urllib3
http = urllib3.PoolManager()
response = http.request('GET', 'https://httpbin.org/bytes/1024', preload_content=False)
# Читаем данные порциями
with open('output.bin', 'wb') as f:
for chunk in response.stream(32):
f.write(chunk)
response.release_conn()
print('Файл скачан')
Результат: Файл output.bin содержит 1024 байта.
Файл скачан
Кастомный User-Agent и другие заголовки
Можно задавать пользовательские заголовки при создании менеджера или в каждом запросе:
import urllib3
http = urllib3.PoolManager(headers={'User-Agent': 'MyApp/1.0'})
response = http.request('GET', 'https://httpbin.org/headers', headers={'X-Custom': 'value'})
print(response.json()['headers']['User-Agent'])
print(response.json()['headers']['X-Custom'])
MyApp/1.0 value
Отправка chunked запроса (Transfer-Encoding: chunked)
Для отправки больших данных частями используется генератор в качестве тела запроса:
import urllib3
def generate_chunks():
yield b'Hello, '
yield b'world!'
http = urllib3.PoolManager()
response = http.request('POST', 'https://httpbin.org/post',
body=generate_chunks(),
headers={'Transfer-Encoding': 'chunked'})
print(response.json()['data'])
"Hello, world!"
Параллельные запросы с пулом соединений
Используя concurrent.futures можно выполнить несколько запросов одновременно, переиспользуя один PoolManager:
import urllib3
from concurrent.futures import ThreadPoolExecutor, as_completed
http = urllib3.PoolManager(maxsize=10)
urls = [f'https://httpbin.org/delay/{i}' for i in range(1, 4)]
def fetch(url):
response = http.request('GET', url)
return response.status, url
with ThreadPoolExecutor(max_workers=5) as executor:
futures = {executor.submit(fetch, url): url for url in urls}
for future in as_completed(futures):
status, url = future.result()
print(f'{url}: {status}')
https://httpbin.org/delay/1: 200 https://httpbin.org/delay/2: 200 https://httpbin.org/delay/3: 200
Верификация SSL с собственным сертификатом
Для использования собственного CA-сертификата создается SSLContext:
import urllib3
import ssl
cert = '/path/to/custom-ca.pem'
ctx = ssl.create_default_context(cafile=cert)
http = urllib3.PoolManager(ssl_context=ctx)
response = http.request('GET', 'https://internal.example.com/api')
print(response.status)
Проблема:
Если сертификат не загружен, возникнет SSLError. Следует проверять правильность пути и формата PEM.
Работа с keep-alive соединениями
urllib3 автоматически переиспользует соединения. Для тонкой настройки используется параметр socket_options и connection_pool_kw:
import urllib3
http = urllib3.PoolManager(
socket_options=urllib3.connection.SSLConnection.socket_options + [
(urllib3.util.connection.SOL_SOCKET, urllib3.util.connection.SO_KEEPALIVE, 1)
]
)
response = http.request('GET', 'https://httpbin.org/get')
print(f'Соединение переиспользовано: {response.connection is not None}')
Соединение переиспользовано: True
Эти примеры демонстрируют гибкость urllib3 для решения различных задач сетевого программирования.