首页 > 编程问答 >urllib3 中重试之间的静态时间

urllib3 中重试之间的静态时间

时间:2024-07-30 07:38:12浏览次数:7  
标签:python urllib3

我想在重试之间使用静态时间,而不是 urllib3 Retry 类中的指数退避因子。

我尝试包装 Retry 类来实现静态重试。然而,虽然 LoggingRetry 正在使用适当的参数进行实例化,但在实际重试发生时将使用默认值。在下面的示例中,重试之间应该有两秒钟的时间。 backoff_factor 会持续存在,但我似乎无法让 delay_between_retries 坚持下去。

谢谢。

import time
import urllib3
from urllib3.util.retry import Retry
from urllib3.exceptions import MaxRetryError
from requests. exceptions import HTTPError
import traceback



# Define the retry strategy with logging
class LoggingRetry(Retry):
    def __init__(self, delay_between_retries=0, *args, **kwargs):
        print(f"Initializing LoggingRetry with delay_between_retries={delay_between_retries}")
        self.delay_between_retries = delay_between_retries
        super().__init__(*args, **kwargs)

    def sleep(self, response=None):
        backoff = self.get_backoff_time()
        total_sleep_time = self.delay_between_retries + backoff
        print(f"LoggingRetry sleep -  delay_between_retries: {self.delay_between_retries}, Backoff time: {backoff}")

        time.sleep(total_sleep_time)


class MyClient:
    def __init__(self, total_tries=3, delay_between_retries=0, backoff_factor=0):
        self.total_tries = total_tries
        self.delay_between_retries = delay_between_retries
        self.backoff_factor = backoff_factor

        # Define a retry strategy with a custom fixed_wait_time
        self.retry_strategy = LoggingRetry(
            total=self.total_tries,
            status_forcelist=[202],
            backoff_factor=self.backoff_factor,
            delay_between_retries=self.delay_between_retries,
            raise_on_status=False,
            allowed_methods=frozenset(['POST'])
        )
        
        # Initialize HTTP client with retry strategy
        self.http = urllib3.PoolManager(retries=self.retry_strategy, maxsize=4, block=True)

    
    def request_data(self, url):    
        try:
            response = self.http.request('GET', url,)
            return response
        except MaxRetryError as me:
            print(f"Max retries reached: {me}")
        except HTTPError as he:
            print(f"An HTTP error occurred: {he}")
            traceback.print_exc()
        except Exception as e:
            print(f"An error occurred: {e}")
            traceback.print_exc()

MyClient(delay_between_retries=2, backoff_factor=0).request_data(url='http://0.0.0.202/')

当然可以,我可以帮解决这个问题。对 urllib3 重试机制的理解是正确的,并且的 LoggingRetry 类中几乎实现了期望的行为。问题出在对 get_backoff_time() 方法的调用中。

urllib3 Retry 类中, sleep() 方法(已经正确地对其进行了覆盖)默认情况下使用 self.get_backoff_time() 来计算睡眠时间。此方法引入了指数后退行为。

为了实现 静态 重试时间间隔,需要调整 LoggingRetry 类的 sleep() 方法,使其不使用 get_backoff_time() 计算值。

以下是经过修改的 LoggingRetry 类:

class LoggingRetry(Retry):
    def __init__(self, delay_between_retries=0, *args, **kwargs):
        print(f"Initializing LoggingRetry with delay_between_retries={delay_between_retries}")
        self.delay_between_retries = delay_between_retries
        super().__init__(*args, **kwargs)

    def sleep(self, response=None):
        # 删除以下行以禁用指数后退计算:
        # backoff = self.get_backoff_time() 
        total_sleep_time = self.delay_between_retries
        print(f"LoggingRetry sleep -  delay_between_retries: {self.delay_between_retries}, Total sleep time: {total_sleep_time}")

        time.sleep(total_sleep_time)

通过此更改, sleep() 方法现在将始终使用在 delay_between_retries 参数中传递的值,从而在每次重试时提供一个静态的 2 秒延迟。

希望这对有所帮助!如果有任何其他问题,请告诉我。

标签:python,urllib3
From: 78809318

相关文章

  • Python多重处理,如何避免创建具有百万个对象的元组
    python多处理新手。我有一项任务,涉及访问网络服务数百万次并将响应保存在文件中(每个请求都有单独的文件)。我已经得到了高级工作代码,但对一些事情没有感到困惑。以下两种语法有什么区别?pool=Pool(processes=4)pool.starmap(task,listOfInputParametersTu......
  • Python OpenCV - 显示坏像素检查测试
    我想找到显示器中存在的每个坏像素。坏像素可能是颜色不正确的像素,或者像素只是黑色。显示屏的尺寸为160x320像素。所以如果显示效果好的话,必须有160*320=51200像素。如果显示器没有51200像素,那就是坏的。另外,我想知道每个坏像素的位置。一旦拍摄的图像太大,我将共享一个......
  • 在python日志输出的每一行前面添加变量缩进
    我正在将日志记录构建到一个Python应用程序中,我希望它是人类可读的。目前,调试日志记录了调用的每个函数以及参数和返回值。这意味着,实际上,嵌套函数调用的调试日志可能如下所示:2024-07-2916:52:26,641:DEBUG:MainController.initialize_componentscalledwithargs<control......
  • 使用 DQN 实现 pong,使用 python 中的特征向量而不是像素。我的 DQNA 实现代码正确吗,因
    我正在致力于使用OpenAI的Gym为Pong游戏实现强化学习(RL)环境。目标是训练人工智能代理通过控制球拍来打乒乓球。代理收到太多负面奖励,即使它看起来移动正确。具体来说,奖励函数会惩罚远离球的智能体,但这种情况发生得太频繁,即使球朝球拍移动时似乎也会发生。观察......
  • Python CDLL 无法加载两次
    我正在尝试用python创建一个密码管理器,但遇到了一个问题,一旦加载了一种类型的dll,我就无法加载不同的dll,在这个示例中,我加载了一个dll,并尝试解密加密的密码数据,它工作正常,直到我加载另一个不同的nss3.dll文件,此时它给我一个错误:“过程入口点HeapAlloc无法位于动态链......
  • 你能将 HTTPS 功能添加到 python Flask Web 服务器吗?
    我正在尝试构建一个Web界面来模拟网络设备上的静态接口,该网络设备使用摘要式身份验证和HTTPS。我想出了如何将摘要式身份验证集成到Web服务器中,但我似乎无法找到如何使用FLASK获取https,如果您可以向我展示如何实现,请评论我需要使用下面的代码做什么来实现这一点。from......
  • Python:比较 csv 文件并打印相似之处
    我需要比较两个csv文件并打印出它们的相似之处。第一个文件有名称和浓度,第二个文件就像只有名称的“最佳”列表,我需要绘制相似性图表。例如,这就是我的列表的样子:file1-old_file.csvname_id,conc_test1,conc_test2name1,####,####name2,###......
  • Python 类交叉引用
    我用Python创建了一个数独游戏。我有一个:单元格类-“保存”数字可能性单元格组-保存单元格类实例我使用这些组在数独中运行行、列和正方形功能。每个单元格包含所有组,他属于classCell:def__init__(groups):self.groups=groupscla......
  • 如何修复我的 Python Azure Function DevOps Pipeline 上的“找到 1 个函数(自定义)加载
    我正在尝试使用AzureDevOps构建管道将PythonAzureFunction部署到Azure门户。由于某种原因,代码被部署到服务器,但我在尝试访问端点时收到404错误。我收到一个错误,显示1functionsfound(Custom)0functionsloaded,以及在服务器上显示ModuleNotFound......
  • 使用 kivy 从 python 脚本的 buildozer 构建 android apk 时出错
    我想从使用kivy包构建的Python脚本构建apk为此,我使用googlecollab.这里是main.py脚本:importyoutube_dlfromkivy.appimportAppfromkivy.uix.boxlayoutimportBoxLayoutfromkivy.uix.buttonimportButtonfromkivy.uix.tex......