Рубрики
Без рубрики

Расширенное использование запросов Python – Timeouts, Retries, крючки

Запросы на библиотеку HTTP Python, вероятно, моя любимая утилита HTTP на всех языках I программируется … Помечено с Python, учебным пособием, WebDev, сегодня предшествующим.

Библиотека Python HTTP Запросы Вероятно, моя любимая http-утилита на всех языках I программирует. Это просто, интуитивно понятно и повсеместно в сообществе Python. Большинство программ, которые взаимодействуют с HTTP используют либо запросы или URLLIB3 из стандартной библиотеки.

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

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

Оглавление

  • Запросить крючки
  • Настройка базовых URL-адресов
  • Настройка тайм-аута по умолчанию
  • Повторите попытку отказа
  • Отладка HTTP-запросов
  • Тестирование и издевательства запросов
  • Повышение поведения браузера

Запросить крючки

Часто при использовании API третьей стороны вы хотите убедиться, что возвращенный ответ действительно действителен. Запросы предлагают помощник состязания ROING_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()

Это может получить повторение, если вам нужно ROING_FOR_STATUS () для каждого вызова. К счастью, библиотека запросов предлагает интерфейс «крючки», где вы можете прикрепить обратные вызовы на определенных частях процесса запроса.

Мы можем использовать крючки для обеспечения ROING_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 Call:

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")

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

Настройка тайм-аута по умолчанию

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

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

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

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

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

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-клиенту просто. Мы создаем htttpadapter и передаем нашу стратегию к адаптеру.

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")

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

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

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 Поскольку большинство 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

Значение экспоненциально увеличивается, что является внедрением по умолчанию Sane для Стратегии повторных попыток Отказ

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

Полная документация по модулю повторной попытки – здесь Отказ

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

Поскольку Htttraadapter сопоставим, мы можем комбинировать повторные попытки и тайм-ауты, которые так:

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 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 Lifecycle, в том числе как текстовое представление запроса и ответа, вы можете использовать крючки запроса и дамп Utils из запросов_toolbelt.

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

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-ответ на сервер Getentry/Ответы Перехватывает 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 в зависимости от того, если вы используете браузер или доступ к сайту программно. Иногда это анти-соскобная мера, но обычно серверы участвуют в пользовательском агенте, чтобы узнать, какой контент лучше всего подходит для устройства (E.G Desktop или Mobile).

Если вы хотите вернуть тот же контент, что и дисплеи браузера, вы можете переопределить запросы заголовка пользователя, наборы с чем-то 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"
})

Оригинал: “https://dev.to/danihodovic/advanced-usage-of-python-requests-timeouts-retries-hooks-kok”

Читайте ещё по теме: