Выполнение HTTP-запросов в Python: подробное руководство

Расширенные примеры запросов веб-страниц

Ниже приведены подробные примеры с кодом и выводом, демонстрирующие разные сценарии.

1. Библиотека requests: сессия, заголовки, куки, прокси

import requests

# Создание сессии (сохраняет куки, заголовки, пул соединений)
session = requests.Session()
session.headers.update({'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36'})
session.cookies.set('sessionid', 'abc123', domain='httpbin.org')

# GET-запрос с прокси (если требуется через HTTPS прокси)
proxies = {'http': 'http://user:pass@proxy:8080', 'https': 'http://user:pass@proxy:8080'}
try:
    response = session.get('https://httpbin.org/cookies', proxies=proxies, timeout=10)
    response.raise_for_status()
    print(response.json())
except requests.exceptions.ProxyError as e:
    print(f"Ошибка прокси: {e}")

{"cookies": {"sessionid": "abc123"}}

Пояснение: сессия автоматически отправляет установленные куки. Прокси задаются словарём. Если прокси не нужны, параметр опускается.

2. urllib.request: кастомные заголовки и обработка перенаправлений

import urllib.request

url = 'https://httpbin.org/redirect/3'  # сервер делает 3 редиректа
req = urllib.request.Request(url, headers={'User-Agent': 'CustomAgent'})
# Отключить автоматическое следование редиректам (по умолчанию включено)
# Для этого создаём свой обработчик (opener)
opener = urllib.request.build_opener()
opener.add_handler = urllib.request.HTTPRedirectHandler()
opener.add_handler.max_redirections = 0  # не следовать

try:
    with opener.open(req, timeout=5) as response:
        print(f"Код ответа: {response.status}")
        print(response.read().decode()[:100])
except urllib.error.HTTPError as e:
    print(f"Ошибка: {e.code} {e.reason}")

Код ответа: 302
Ошибка: 302 Found

Пояснение: HTTPRedirectHandler управляет редиректами. Установив max_redirections=0, мы не следуем перенаправлениям и получаем ответ с кодом 302.

3. httpx: асинхронные запросы с повторными попытками

import asyncio
import httpx

async def fetch_with_retry(url, retries=3):
    async with httpx.AsyncClient(http2=True, timeout=10) as client:
        for attempt in range(retries):
            try:
                resp = await client.get(url)
                resp.raise_for_status()
                return resp.text[:150]
            except httpx.HTTPStatusError as e:
                if attempt == retries - 1:
                    raise
                await asyncio.sleep(2 ** attempt)  # экспоненциальная задержка

async def main():
    result = await fetch_with_retry('https://httpbin.org/status/503')
    print(result)

asyncio.run(main())

(При успехе вернётся текст, при трёх неудачах - исключение)

Пояснение: httpx.AsyncClient используется в асинхронном контексте. Механизм повторных попыток с задержкой помогает при временных ошибках сервера.

4. aiohttp: параллельный сбор данных с ограничением числа одновременных соединений

import asyncio
import aiohttp
from aiohttp import ClientTimeout

async def fetch(session, url, semaphore):
    async with semaphore:
        try:
            async with session.get(url, timeout=ClientTimeout(total=10)) as resp:
                return await resp.text()
        except (aiohttp.ClientError, asyncio.TimeoutError) as e:
            return f"Ошибка: {e}"

async def main():
    urls = [f'https://httpbin.org/delay/{i}' for i in range(1, 6)]
    semaphore = asyncio.Semaphore(3)  # не более 3 одновременных запросов
    async with aiohttp.ClientSession() as session:
        tasks = [fetch(session, url, semaphore) for url in urls]
        results = await asyncio.gather(*tasks, return_exceptions=False)
        for r in results:
            print(len(r))

asyncio.run(main())

(Выводятся длины ответов, от 200 до 2000 символов)

Пояснение: семафор (asyncio.Semaphore) ограничивает количество одновременных запросов для предотвращения перегрузки сети или сервера.

5. subprocess + curl: передача тела POST-запроса и обработка заголовков

import subprocess
import json

url = 'https://httpbin.org/post'
data = json.dumps({'key': 'value'})
cmd = ['curl', '-s', '-X', 'POST', url, '-H', 'Content-Type: application/json', '-d', data]
result = subprocess.run(cmd, capture_output=True, text=True, timeout=10)
if result.returncode == 0:
    resp = json.loads(result.stdout)
    print(resp['json'])
else:
    print(f"Ошибка: {result.stderr}")

{'key': 'value'}

Пояснение: аргументы curl передаются в виде списка. Флаг -d отправляет тело POST-запроса. Вывод разбирается как JSON.