Использование requests в Python — тайм-ауты, повторы, хуки

HTTP-библиотека Python requests - это, вероятно, моя любимая HTTP-утилита во всех языках, на которых я программирую. Она проста, интуитивно понятна и повсеместно распространена в сообществе Python. Большинство программ, взаимодействующих с HTTP, используют либо requests, либо urllib3 из стандартной библиотеки.

Хотя благодаря простому API можно сразу же начать продуктивно работать с запросами, библиотека также предлагает расширяемость для продвинутых случаев использования. Если вы пишете API-клиент или веб-скребок, вам наверняка понадобится устойчивость к сетевым сбоям, полезные трассировки отладки и синтаксический сахар

Ниже приводится краткое описание возможностей, которые я нашел полезными в запросах при написании инструментов веб-скрейпинга или программ, которые широко используют JSON API.

• Крючки запросов
• Установка базовых URL
• Установка таймаутов по умолчанию
• Возврат при неудаче
  • Комбинирование тайм-аутов и повторных попыток
• Отладка HTTP-запросов
  • Печать HTTP-заголовков
  • Печать всего
  • .
• Тестирование и издевательство над запросами
• Имитация поведения браузера

Хуки для запросов

Часто при использовании стороннего API вы хотите убедиться, что возвращаемый ответ действительно действителен. Requests предлагает сокращенный помощник raise_for_status(), который утверждает, что код состояния HTTP ответа не является 4xx или 5xx, т.е. что запрос не привел к ошибке клиента или сервера.

Например

response = requests.get('https://api.github.com/user/repos?page=1')
# Assert that there were no errors
response.raise_for_status()

Это может стать повторяющимся, если вам нужно запрашивать raise_for_status() для каждого вызова. К счастью, библиотека requests предлагает интерфейс 'hooks', с помощью которого вы можете подключить обратные вызовы на определенные части процесса запроса.

Мы можем использовать хуки, чтобы обеспечить вызов raise_for_status() для каждого объекта ответа.

# Create a custom requests object, modifying the global module throws an error
http = requests.Session()

assert_status_hook = lambda response, *args, **kwargs: response.raise_for_status()
http.hooks["response"] = [assert_status_hook]

http.get("https://api.github.com/user/repos?page=1")

> HTTPError: 401 Client Error: Unauthorized for url: https://api.github.com/user/repos?page=1

Установка базовых URL

Предположим, что вы используете только один API, размещенный на api.org. В итоге вы будете повторять протокол и домен для каждого вызова http:

requests.get('https://api.org/list/')
requests.get('https://api.org/list/3/item')

Вы можете сэкономить на вводе текста, используя BaseUrlSession. Это позволит вам указать базовый url для HTTP клиента и указать только путь к ресурсу во время запроса.

from requests_toolbelt import sessions
http = sessions.BaseUrlSession(base_url="https://api.org")
http.get("/list")
http.get("/list/item")

Обратите внимание, что набор инструментов requests toolbelt не включен в стандартную установку requests, поэтому вам придется установить его отдельно.

Установка тайм-аутов по умолчанию

Документация по requests рекомендует устанавливать таймауты для всего производственного кода. Если вы забудете установить тайм-ауты, неправильно работающий сервер может привести к зависанию вашего приложения, особенно учитывая, что большинство кода Python является синхронным.

requests.get('https://github.com/', timeout=0.001)

Используя Транспортные адаптеры, мы можем установить таймаут по умолчанию для всех HTTP-вызовов. Это гарантирует, что будет установлен разумный тайм-аут, даже если разработчик забудет добавить параметр timeout=1 в свой отдельный вызов, но позволяет отменить его на основе каждого вызова.

Ниже приведен пример пользовательского транспортного адаптера с таймаутами по умолчанию, вдохновленный этим комментарием на Github. Мы переопределяем конструктор для предоставления таймаута по умолчанию при создании http-клиента и метод send() для обеспечения использования таймаута по умолчанию, если не предоставлен аргумент таймаута.

from requests.adapters import HTTPAdapter

DEFAULT_TIMEOUT = 5 # seconds

class TimeoutHTTPAdapter(HTTPAdapter):
    def __init__(self, *args, **kwargs):
        self.timeout = DEFAULT_TIMEOUT
        if "timeout" in kwargs:
            self.timeout = kwargs["timeout"]
            del kwargs["timeout"]
        super().__init__(*args, **kwargs)

    def send(self, request, **kwargs):
        timeout = kwargs.get("timeout")
        if timeout is None:
            kwargs["timeout"] = self.timeout
        return super().send(request, **kwargs)

Мы можем использовать его следующим образом:

import requests

http = requests.Session()

# Mount it for both http and https usage
adapter = TimeoutHTTPAdapter(timeout=2.5)
http.mount("https://", adapter)
http.mount("http://", adapter)

# Use the default 2.5s timeout
response = http.get("https://api.twilio.com/")

# Override the timeout as usual for specific requests
response = http.get("https://api.twilio.com/", timeout=10)

Причина отказа

Сетевые соединения несут потери, перегружены, а серверы выходят из строя. Если мы хотим создать действительно надежную программу, мы должны учитывать отказы и иметь стратегию повторных попыток.

Добавить стратегию повторной попытки в HTTP-клиент очень просто. Мы создаем HTTPAdapter и передаем нашу стратегию адаптеру.

from requests.adapters import HTTPAdapter
from requests.packages.urllib3.util.retry import Retry

retry_strategy = Retry(
    total=3,
    status_forcelist=[429, 500, 502, 503, 504],
    method_whitelist=["HEAD", "GET", "OPTIONS"]
)
adapter = HTTPAdapter(max_retries=retry_strategy)
http = requests.Session()
http.mount("https://", adapter)
http.mount("http://", adapter)

response = http.get("https://en.wikipedia.org/w/api.php")

Класс Retry по умолчанию предлагает разумные параметры по умолчанию, но в значительной степени настраивается, поэтому здесь приводится сводка наиболее часто используемых мною параметров.

Приведенные ниже параметры включают параметры по умолчанию, используемые библиотекой requests.

total=10

Общее количество повторных попыток. Если количество неудачных запросов или перенаправлений превысит это число, клиент выбросит исключение urllib3.exceptions.MaxRetryError. Я варьирую этот параметр в зависимости от API, с которым работаю, но обычно устанавливаю его меньше 10, обычно достаточно 3 повторных попыток.

status_forcelist=[413, 429, 503]

Коды ответов HTTP для повторных попыток. Скорее всего, вы захотите повторить попытку при обычных ошибках сервера (500, 502, 503, 504), поскольку серверы и обратные прокси не всегда придерживаются спецификации HTTP. Всегда повторяйте попытку при превышении лимита скорости 429, потому что библиотека urllib по умолчанию должна инкрементально отступать при неудачных запросах.

method_whitelist=["HEAD", "GET", "PUT", "DELETE", "OPTIONS", "TRACE"]

Методы HTTP для повторных попыток. По умолчанию сюда включены все HTTP-методы, кроме POST, поскольку POST может привести к новой вставке.  Измените этот параметр так, чтобы он включал POST, потому что большинство API, с которыми я имею дело, не возвращают код ошибки и не выполняют вставку в одном и том же вызове. А если они это делают, вам, вероятно, следует опубликовать сообщение об ошибке.

backoff_factor=0

Это интересный параметр. Он позволяет изменять время сна процессов между неудачными запросами. Алгоритм следующий:

{backoff factor} * (2 ** ({number of total retries} - 1))

Например, если коэффициент отката установлен на:

• 1 секунда последующие сны будут 0.5, 1, 2, 4, 8, 16, 32, 64, 128, 256.
• 2 секунды - 1, 2, 4, 8, 16, 32, 64, 128, 256, 512
• 10 секунд - 5, 10, 20, 40, 80, 160, 320, 640, 1280, 2560

Значение экспоненциально возрастает, что является разумной реализацией по умолчанию для стратегий возврата.

По умолчанию это значение равно 0, что означает, что экспоненциальный откат не будет установлен и повторные попытки будут выполняться немедленно.  Обязательно установите значение 1, чтобы избежать забивания ваших серверов!.

Полная документация по модулю retry находится здесь.

Комбинирование тайм-аутов и повторных попыток

Поскольку HTTPAdapter является сравнимым, мы можем объединить повторные попытки и таймауты следующим образом:

retries = Retry(total=3, backoff_factor=1, status_forcelist=[429, 500, 502, 503, 504])
http.mount("https://", TimeoutHTTPAdapter(max_retries=retries))

Отладка HTTP-запросов

Иногда запросы не работают, и вы не можете понять, почему. Логирование запроса и ответа может дать вам понимание причины неудачи. Есть два способа сделать это - либо используя встроенные настройки отладочного протоколирования, либо используя крючки запросов.

Печать HTTP-заголовков

При изменении уровня отладки логирования больше 0 будут записываться HTTP-заголовки ответа. Это самый простой вариант, но он не позволяет вам увидеть HTTP-запрос или тело ответа. Это полезно, если вы имеете дело с API, который возвращает большое тело полезной нагрузки, не подходящее для протоколирования, или содержит двоичное содержимое.

Любое значение, которое больше 0, включит ведение отладочного журнала.

import requests
import http

http.client.HTTPConnection.debuglevel = 1

requests.get("https://www.google.com/")

# Output
send: b'GET / HTTP/1.1\r\nHost: www.google.com\r\nUser-Agent: python-requests/2.22.0\r\nAccept-Encoding: gzip, deflate\r\nAccept: */*\r\nConnection: keep-alive\r\n\r\n'
reply: 'HTTP/1.1 200 OK\r\n'
header: Date: Fri, 28 Feb 2020 12:13:26 GMT
header: Expires: -1
header: Cache-Control: private, max-age=0

Печать всего

Если вы хотите регистрировать весь жизненный цикл HTTP, включая текстовое представление запроса и ответа, вы можете использовать крючки запроса и утилиты дампа из requests_toolbelt.

Я предпочитаю этот вариант всегда, когда имею дело с API на основе REST, который не возвращает очень большие ответы.

import requests
from requests_toolbelt.utils import dump

def logging_hook(response, *args, **kwargs):
    data = dump.dump_all(response)
    print(data.decode('utf-8'))

http = requests.Session()
http.hooks["response"] = [logging_hook]

http.get("https://api.openaq.org/v1/cities", params={"country": "BA"})

# Output
< GET /v1/cities?country=BA HTTP/1.1
< Host: api.openaq.org

> HTTP/1.1 200 OK
> Content-Type: application/json; charset=utf-8
> Transfer-Encoding: chunked
> Connection: keep-alive
>
{
   "meta":{
      "name":"openaq-api",
      "license":"CC BY 4.0",
      "website":"https://docs.openaq.org/",
      "page":1,
      "limit":100,
      "found":1
   },
   "results":[
      {
         "country":"BA",
         "name":"Goražde",
         "city":"Goražde",
         "count":70797,
         "locations":1
      }
   ]
}

Смотрите https://toolbelt.readthedocs.io/en/latest/dumputils.html

Тестирование и подражание запросам

Использование сторонних API создает болевую точку в разработке - их трудно тестировать. Инженеры из Sentry облегчили некоторые из этих проблем, написав библиотеку для имитации запросов во время разработки.

Вместо отправки HTTP-ответа на сервер getsentry/responses перехватывает HTTP-запрос и возвращает заранее заданный ответ, который вы добавили во время тестирования.

Лучше показать на примере.

import unittest
import requests
import responses


class TestAPI(unittest.TestCase):
    @responses.activate  # intercept HTTP calls within this method
    def test_simple(self):
        response_data = {
                "id": "ch_1GH8so2eZvKYlo2CSMeAfRqt",
                "object": "charge",
                "customer": {"id": "cu_1GGwoc2eZvKYlo2CL2m31GRn", "object": "customer"},
            }
        # mock the Stripe API
        responses.add(
            responses.GET,
            "https://api.stripe.com/v1/charges",
            json=response_data,
        )

        response = requests.get("https://api.stripe.com/v1/charges")
        self.assertEqual(response.json(), response_data)

Если выполняется HTTP-запрос, который не соответствует имитируемым ответам, возникает ошибка ConnectionError.

class TestAPI(unittest.TestCase):
    @responses.activate
    def test_simple(self):
        responses.add(responses.GET, "https://api.stripe.com/v1/charges")
        response = requests.get("https://invalid-request.com")

Выход

requests.exceptions.ConnectionError: Connection refused by Responses - the call doesn't match any registered mock.

Request:
- GET https://invalid-request.com/

Available matches:
- GET https://api.stripe.com/v1/charges

Имитация поведения браузера

Если вы написали достаточно кода для веб-скрапера, вы заметите, что некоторые сайты возвращают разный HTML в зависимости от того, используете ли вы браузер или обращаетесь к сайту программно. Иногда это является мерой защиты от скраппинга, но обычно серверы используют User-Agent sniffing, чтобы выяснить, какой контент лучше всего подходит для конкретного устройства (например, настольного или мобильного)

Если вы хотите вернуть то же содержимое, которое отображает браузер, вы можете переопределить наборы заголовков User-Agent запросов с тем, что послал бы Firefox или Chrome.

import requests
http = requests.Session()
http.headers.update({
    "User-Agent": "Mozilla/5.0 (X11; Ubuntu; Linux x86_64; rv:68.0) Gecko/20100101 Firefox/68.0"
})

Вернуться на верх