我正在运行以下代码来发出异步“获取”请求。所述CustomSession类用于添加用于定时每个请求支持。
如果发生异常或请求运行正常，我希望能够访问test_id添加到futures列表的URL以及要请求的URL。换句话说，当请求运行或引发异常时，我想找到test_id与调用关联的对象session.get。
from datetime import datetime
from requests_futures.sessions import FuturesSession
class CustomSession(FuturesSession):
def __init__(self, *args, **kwargs):
super(CustomSession, self).__init__(*args, **kwargs)
self.timing = {}
def request(self, method, url, *args, **kwargs):
background_callback = kwargs.pop('background_callback', None)
test_id = kwargs.pop('test_id', None)
# start counting
self.timing[test_id] = datetime.now()
def time_it(sess, resp):
# here if you want to time the server stuff only
self.timing[test_id] = datetime.now() - self.timing[test_id]
if background_callback:
background_callback(sess, resp)
# here if you want to include any time in the callback
return super(CustomSession, self).request(method, url, *args,
background_callback=time_it,
**kwargs)
session = CustomSession()
futures = []
for url in ('http://httpbin.org/get?key=val',
'http://httpasdfasfsadfasdfasdfbin.org/get?key2=val2'):
futures.append(session.get(url, test_id=1))
for future in futures:
try:
r = future.result()
print(r.status_code)
except Exception as e:
print(e)
我为将来对象的result（）函数创建了一个装饰器：
def mark_exception(fn, id, url):
def new_fn(*args, **kwargs):
try:
return fn(*args, **kwargs)
except:
raise Exception("test id %d with url %s threw exception" % (id, url))
return new_fn
并将其应用于CustomSession.request（）函数的末尾，替换原始的return语句：
future = super(CustomSession, self).request(method, url, *args,
background_callback=time_it,
**kwargs)
future.result = mark_exception(future.result, test_id, url)
return future
输出：
200
test id 1 with url http://httpasdfasfsadfasdfasdfbin.org/get?key2=val2 threw exception
我希望这有帮助。
编辑：
如果您想获取每个将来的测试ID，可以通过以下两种方法进行：
futures = []
for url in ('http://httpbin.org/get?key=val',
'http://httpasdfasfsadfasdfasdfbin.org/get?key2=val2'):
tid = 1
future = session.get(url, test_id=tid)
# option 1: set test_id as an attrib of the future object
future.test_id = tid
# option 2: put test_id and future object in a tuple before appending to the list
futures.append((tid, future))
for tid, future in futures:
try:
r = future.result()
print("tracked test_id is %d" % tid) #option 2
print("status for test_id %d is %d" % (future.test_id, r.status_code)) #option 1
except Exception as e:
print(e)
本文收集自互联网，转载请注明来源。
如有侵权，请联系 [email protected] 删除。
我来说两句