Python自动重试HTTP连接装饰器

yipeiwu_com6年前Python基础

有时候我们要去别的接口取数据,可能因为网络原因偶尔失败,为了能自动重试,写了这么一个装饰器。
这个是python2.7x 的版本,python3.x可以用 nonlocal 来重写。

#-*- coding: utf-8 -*-  
#all decorators in this tool file 
#author: orangleliu 
 
############################################################ 
#http连接有问题时候,自动重连 
def conn_try_again(function): 
  RETRIES = 0 
  #重试的次数 
  count = {"num": RETRIES} 
  def wrapped(*args, **kwargs): 
    try: 
      return function(*args, **kwargs) 
    except Exception, err: 
      if count['num'] < 2: 
        count['num'] += 1 
        return wrapped(*args, **kwargs)          
      else: 
        raise Exception(err) 
  return wrapped 

用法很的简单,下面是一个程序片段。

@conn_try_again 
def post_query_bandwidth_for_bandwidth(self, contract_no, data_month, product_code): 
  #根据webluker接口情况获取计费数据   
  try: 
    post_data = {'contract':contract_no, 'month': data_month, 'code':product_code} 
    params = urllib.urlencode(post_data) 
    response = urllib2.urlopen(WEBLUKER_BANDWITH_API + "?" +params) 
    billdata = {} 
    billdata = response.read() 
    if not billdata: 
      billdata = {} 
    return billdata 
  except Exception, err: 
    err = u'与webluker接口间通信异常' 
    raise Exception(err) 

如果try块中有异常,就会自动重试2次。

相关文章

解决pyttsx3无法封装的问题

解决pyttsx3无法封装的问题

一、python3下面的pyttsx3正常可以使用,类似下面: import pyttsx3 engine = pyttsx3.init() rate = engine.getPro...

Python Threading 线程/互斥锁/死锁/GIL锁

导入线程包 import threading 准备函数线程,传参数 t1 = threading.Thread(target=func,args=(args,)) 类继承线程,创建线程对...

pytorch 修改预训练model实例

我就废话不多说了,直接上代码吧! class Net(nn.Module): def __init__(self , model): super(Net, self)._...

基于python实现的百度新歌榜、热歌榜下载器(附代码)

基于python实现的百度新歌榜、热歌榜下载器(附代码)

前言 首先声明,本工具仅仅为学习之用,不涉及版权问题,因为百度音乐里面的歌曲本身是可以下载的,而且现在百度也提供了”百度音乐播放器”,可以通过这个工具进行批量下载。 我当时做这个工具的时...

分享8个非常流行的 Python 可视化工具包

分享8个非常流行的 Python 可视化工具包

喜欢用 Python 做项目的小伙伴不免会遇到这种情况:做图表时,用哪种好看又实用的可视化工具包呢?之前文章里出现过漂亮的图表时,也总有读者在后台留言问该图表时用什么工具做的。下面,作者...