编写装饰器实现python请求错误重试功能

来源:http://www.sh-fengwen.com 作者:家常菜谱 人气:140 发布时间:2019-11-12
摘要:在做接口自动化测试的时候,总会遇到,因连接超时等错误导致,接口脚本失败。 利用python写一些网络服务的时候,当网络状况不好,或者资源占用过多,任务拥塞的情况下,总会抛出

在做接口自动化测试的时候,总会遇到,因连接超时等错误导致,接口脚本失败。

利用python 写一些网络服务的时候,当网络状况不好,或者资源占用过多,任务拥塞的情况下,总会抛出一些异常,当前任务就被终止了,可以很好的利用@装饰器,写一个重试的装饰器,这样比较python!
执行结果:

官方给出的方法:

复制代码 代码如下:

max_retries=5 出错重试5次
注意的是,这个只对DNS,连接错误进行重试。

    from requests.adapters import HTTPAdapter
    s = requests.Session()
    s.mount('http://',HTTPAdapter(max_retries=5))
    s.mount('https://',HTTPAdapter(max_retries=5))
    s.get('https://www.baidu.com')

WARNING:root:timed out, Retrying in 3 seconds...
WARNING:root:timed out, Retrying in 6 seconds...
WARNING:root:timed out, Retrying in 12 seconds...

 

复制代码 代码如下:

 

#!/usr/bin/env python
# -*- coding: utf-8 -*-
#  [email protected]
#  2014/01/27 10:36
import time
import logging
import socket
from functools import wraps

自编写装饰器一

logging.basicConfig(level=logging.DEBUG)

from requests.exceptions import ConnectionError
import requests
def retry(**kw):
    def war(func):
        def w(*args,**kwargs):
            try:
                ret = func(*args,**kwargs)
            except ConnectionError:
                kw['reNum'] = int(kw['reNum']) - 1
                if kw['reNum'] >=0:
                    print kw['reNum']
                    ret = w(*args,**kwargs)
                else:
                    ret = ConnectionError
            return ret
        return w
    return war

def retry(MyException, tries=4, delay=3, backoff=2, logger=None):
    def deco_retry(f):
        @wraps(f)
        def f_retry(*args, **kwargs):
            mtries, mdelay = tries, delay
            while mtries > 1:
                try:
                    return f(*args, **kwargs)
                except MyException as ex:
                    msg = "%s, Retrying in %d seconds..." % (str(ex), mdelay)
                    if logger:
                        logger.warning(msg)
                    else:
                        print msg
                    time.sleep(mdelay)
                    mtries -= 1
                    mdelay *= backoff
            return str(ex)

 

        return f_retry

自编写装饰器二

    return deco_retry

from requests.exceptions import ConnectionError

def retry(**kw):
    def wrapper(func):
        def _wrapper(*args,**kwargs):
            raise_ex = None
            for _ in range(kw['reNum']):
                print _
                try:
                    return func(*args,**kwargs)
                except ConnectionError as ex:
                    raise_ex = ex
            #raise raise_ex
        return _wrapper
    return wrapper

@retry(Exception, logger=logging)
def check():
    sk = socket.socket()
    sk.settimeout(5)
    sk.connect(('6.6.6.6', 80))

 

if __name__ == "__main__":
    check()

使用方法:reNum = 5 代表,出现ConnectionError时最多可重试5次。

写一些网络服务的时候,当网络状况不好,或者资源占用过多,任务拥塞的情况下,总会抛出一些异常,当前任务就被终止了,可...

@retry(reNum=5)
def demo():
    raise ConnectionError

本文由美高梅游戏平台网站发布于家常菜谱,转载请注明出处:编写装饰器实现python请求错误重试功能

关键词:

上一篇:python进阶(四) windows下虚拟环境使用

下一篇:没有了

最火资讯