如何重试特定异常的请求

Question

我目前一直在使用我自己的“重试”功能，我想重试直到请求有效。在某些情况下，如果我达到任何 5xx 状态，我应该在很长的延迟后重试。

如果我点击特定的状态码，例如200 或 404，它不应该引发状态码，否则引发它。

所以我做了这样的事情：

import time

import requests
from bs4 import BeautifulSoup
from requests import (
    RequestException,
    Timeout
)


def do_request():
    try:
        # There is some scenarios where I would use my own proxies by doing
        # requests.get("https://www.bbc.com/", timeout=0.1, proxies={'https': 'xxx.xxxx.xxx.xx'))
        while (response := requests.get("https://www.bbc.com/", timeout=0.1)).status_code >= 500:
            print("sleeping")
            time.sleep(20)

        if response.status_code not in (200, 404):
            response.raise_for_status()

        print("Successful requests!")

        soup = BeautifulSoup(response.text, 'html.parser')

        for link in soup.find_all("a", {"class": "media__link"}):
            yield link.get('href')

    except Timeout as err:
        print(f"Retry due to timed out: {err}")

    except RequestException as err:
        raise RequestException("Unexpected request error")


# ----------------------------------------------------#

if __name__ == '__main__':
    for found_links in do_request():
        print(found_links)

现在对我来说的问题是我故意将超时设置为 0.1 以触发异常 Timeout 发生，我希望它在这里发生的是它应该在遇到请求后再次重试。

目前它停在那里，我想知道如果它遇到我不引发错误的超时，我应该怎么做才能再次重试请求？

Answer 1

你可以在你的情况下递归调用函数本身，但要小心意外的边缘情况：

def do_request(retry: int = 3):
    try:
        # There is some scenarios where I would use my own proxies by doing
        # requests.get("https://www.bbc.com/", timeout=0.1, proxies={'https': 'xxx.xxxx.xxx.xx'))
        while (response := requests.get("https://www.bbc.com/", timeout=0.1)).status_code >= 500:
            print("sleeping")
            time.sleep(20)

        if response.status_code not in (200, 404):
            response.raise_for_status()

        print("Successful requests!")

        soup = BeautifulSoup(response.text, 'html.parser')

        for link in soup.find_all("a", {"class": "media__link"}):
            yield link.get('href')

    except Timeout as err:
        if retry:
            print(f"Retry due to timed out: {err}")
            yield from do_request(retry=retry - 1)
        else:
            raise

    except RequestException as err:
        raise RequestException("Unexpected request error")

这将尝试 3 次（或您在参数中设置的次数），直到 retry 等于 0 或直到遇到另一个错误

Answer 2

我会把它放在一个while循环中，当动作完成时打破循环。

示例：

def do_request():
    while True:
        try:
            # There is some scenarios where I would use my own proxies by doing
            # requests.get("https://www.bbc.com/", timeout=0.1, proxies={'https': 'xxx.xxxx.xxx.xx'))
            while (response := requests.get("https://www.bbc.com/", timeout=0.1)).status_code >= 500:
                print("sleeping")
                time.sleep(20)

            if response.status_code not in (200, 404):
                response.raise_for_status()

            print("Successful requests!")

            soup = BeautifulSoup(response.text, 'html.parser')

            for link in soup.find_all("a", {"class": "media__link"}):
                yield link.get('href')
            break
        except Timeout as err:
            print(f"Retry due to timed out: {err}")

        except RequestException as err:
            raise RequestException("Unexpected request error")

您还可以在每次试用之间添加time.sleep(0.1)。

Answer 3

tenacity 包优雅地解决了各种重试问题。

对于您的问题，只需添加这样的装饰器：

@retry(retry=retry_if_exception_type(Timeout))
def do_request():
    while (response := requests.get("https://www.bbc.com/", timeout=0.1)).status_code >= 500:
        print("sleeping")
        time.sleep(20)

    if response.status_code not in (200, 404):
        response.raise_for_status()

    print("Successful requests!")

    soup = BeautifulSoup(response.text, 'html.parser')

    for link in soup.find_all("a", {"class": "media__link"}):
        yield link.get('href')