Использование urllib3 для выполнения HTTP запросов в Python

Раздел: Сетевое программирование -> HTTP-запросы

Основные возможности 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 для решения различных задач сетевого программирования.

Использование urllib3 для запросов - comments

En
Python urllib3 request (python)