本文整理汇总了Python中requests_cache.disabled函数的典型用法代码示例。如果您正苦于以下问题:Python disabled函数的具体用法?Python disabled怎么用?Python disabled使用的例子?那么恭喜您, 这里精选的函数代码示例或许可以为您提供帮助。
在下文中一共展示了disabled函数的20个代码示例,这些例子默认根据受欢迎程度排序。您可以为喜欢或者感觉有用的代码点赞,您的评价将有助于我们的系统推荐出更棒的Python代码示例。
示例1: test_content_and_cookies
def test_content_and_cookies(self):
s = requests.session()
def js(url):
return json.loads(s.get(url).text)
r1 = js('http://httpbin.org/cookies/set/test1/test2')
with requests_cache.disabled():
r2 = js('http://httpbin.org/cookies')
self.assertEqual(r1, r2)
r3 = js('http://httpbin.org/cookies')
with requests_cache.disabled():
r4 = js('http://httpbin.org/cookies/set/test3/test4')
# from cache
self.assertEqual(r3, js('http://httpbin.org/cookies'))
# updated
with requests_cache.disabled():
self.assertEqual(r4, js('http://httpbin.org/cookies'))
开发者ID:dmr,项目名称:requests-cache,代码行数:16,代码来源:test_cache.py
示例2: get_open
def get_open(self):
with requests_cache.disabled():
price_text = requests.get('http://finance.yahoo.com/q?s=%s&ql=1' % self.symbol).text
print self.symbol
start = price_text.find('<span class="time_rtq_ticker">')
end = price_text.find('</span>', start)
self.open_price = float(price_text[start+50+len(self.symbol):end])
开发者ID:mobone,项目名称:AiTrader,代码行数:7,代码来源:monitor2.py
示例3: push_to_ids_at_channel
def push_to_ids_at_channel(self, channel, ids, message):
logger.debug("Pushing {0} to {1}".format(message, channel))
string_ids = ",".join(ids)
payload = {'channel':channel, 'to_ids':string_ids, 'payload':json.dumps({'badge':2, 'sound':'default', 'alert':message})}
url = ACS_URLS['notify'].format(self.key)
with requests_cache.disabled():
r = requests.post(url, data=payload, cookies=self.cookies)
开发者ID:jpgneves,项目名称:t-10_server,代码行数:7,代码来源:teeminus10_helpers.py
示例4: __login
def __login(self):
'''Need to login to appcelerator'''
payload = {'login':self.user, 'password':self.password}
with requests_cache.disabled():
r = requests.post(ACS_URLS['login'].format(self.key), data=payload)
self.cookies = r.cookies
开发者ID:jpgneves,项目名称:t-10_server,代码行数:7,代码来源:teeminus10_helpers.py
示例5: _thread
def _thread(self, item):
title = item[0]
url = item[1]
description = item[2]
self._parse(url, title)
self._parse(url, description)
if url is None or url == "":
return
r = None
headers = { 'User-Agent': "Mozilla/5.0 (compatible; MSIE 10.0; Windows NT 6.1; Trident/6.0)" }
try:
r = requests.get(url, headers=headers, allow_redirects=True, verify=False, timeout=20)
except Exception:
try:
with requests_cache.disabled():
r = requests.get(url, headers=headers, allow_redirects=True, verify=False, timeout=20)
except Exception:
pass
if not r or r.status_code != 200:
self.dlerrors_count += 1
if r:
print("ERROR: %s (%d)" % (url, r.status_code))
else:
print("ERROR: %s" % (url))
return
self._parse(url, r.text)
开发者ID:looran,项目名称:gsdl,代码行数:26,代码来源:gsdl.py
示例6: _api_request
def _api_request(self, query, cache_enabled=True):
request_string = self.API_URL + query
if cache_enabled:
req = requests.get(request_string, headers=self.REQUEST_HEADERS)
else:
with requests_cache.disabled():
req = requests.get(request_string, headers=self.REQUEST_HEADERS)
return req.json()
开发者ID:Darion,项目名称:destipy,代码行数:8,代码来源:__init__.py
示例7: get_projects
def get_projects(project_id_list=None):
global __ENGINE__
with requests_cache.disabled():
if project_id_list is not None:
projects = [Project(__LIMS__,id=pid) for pid in project_id_list]
else:
projects = __LIMS__.get_projects()
return projects
开发者ID:MAKO-Bioinfo,项目名称:mako_pipeline-development_ampliseq,代码行数:8,代码来源:lims.py
示例8: test_content_and_cookies
def test_content_and_cookies(self):
requests_cache.install_cache(CACHE_NAME, CACHE_BACKEND)
s = requests.session()
def js(url):
return json.loads(s.get(url).text)
r1 = js(httpbin('cookies/set/test1/test2'))
with requests_cache.disabled():
r2 = js(httpbin('cookies'))
self.assertEqual(r1, r2)
r3 = js(httpbin('cookies'))
with requests_cache.disabled():
r4 = js(httpbin('cookies/set/test3/test4'))
# from cache
self.assertEqual(r3, js(httpbin('cookies')))
# updated
with requests_cache.disabled():
self.assertEqual(r4, js(httpbin('cookies')))
开发者ID:brbsix,项目名称:requests-cache,代码行数:17,代码来源:test_cache.py
示例9: auth_query
def auth_query(
self,
endpoint,
params=None,
cache_bucket='shortterm',
short_response=True
):
"""
authorized queries here
"""
## Are we actually going to cache?
if not cache_bucket:
use_cache = None
else:
use_cache = True
if cache_bucket:
cache_name = os.path.join('/tmp', cache_bucket)
else:
cache_name = None
cache_timeout = self.constants_data['cache_timeouts'].get(cache_bucket, None)
if not params:
params = {}
import requests_cache
requests_cache.install_cache(cache_name, expire_after=cache_timeout)
from urlparse import urljoin
headers = {"X-API-KEY": API_KEY}
if not SESSION:
sys.stderr.write("Need auth SESSION first\n")
sys.exit(2)
## Make sure this doesn't start with a /
if endpoint[0] == '/':
endpoint = endpoint[1:]
end_url = urljoin(BASE_URL, endpoint)
if not use_cache:
with requests_cache.disabled():
raw_results = SESSION.get(end_url, headers=headers, params=params)
else:
raw_results = SESSION.get(end_url, headers=headers, params=params)
if 'Response' not in raw_results.json():
sys.stderr.write("Error retreiving: %s\n" % end_url)
sys.stderr.write("%s\n" % raw_results.json())
sys.exit(2)
if short_response:
return_result = raw_results.json()['Response']['data']
else:
return_result = raw_results
return return_result
开发者ID:drewstinnett,项目名称:destiny-dragoons,代码行数:58,代码来源:__init__.py
示例10: _fetch
def _fetch(self, uri, cache=True):
if cache:
with requests_cache.enabled():
r = requests.get(uri, headers={'Accept' : 'application/json'})
else:
with requests_cache.disabled():
r = requests.get(uri, headers={'Accept' : 'application/json'})
return r.json
开发者ID:cjseeger,项目名称:aapps,代码行数:9,代码来源:datatank.py
示例11: _batch
def _batch():
if caching_avail and self._cached:
self._cached = False
with requests_cache.disabled():
from_cache, ret = self._get(_url, params=kwargs, verbose=verbose)
self._cached = True
else:
from_cache, ret = self._get(_url, params=kwargs, verbose=verbose)
return ret
开发者ID:SuLab,项目名称:mygene.py,代码行数:9,代码来源:__init__.py
示例12: __do_get
def __do_get(self, url):
with requests_cache.disabled():
r = requests.get(url)
logger.info("Request of weather data to {0}.".format(url))
try:
return json.loads(r.text)
except ValueError:
logger.warning("Could not get weather data from {0}".format(url))
return {} # Something went wrong!
开发者ID:jpgneves,项目名称:t-10_server,代码行数:9,代码来源:teeminus10_helpers.py
示例13: release_history
def release_history(channel, os):
# Query limit is 1000, so to go back far enough we need to query
# each os/channel pair separately.
url = RELEASE_HISTORY_CSV_URL + "?os=%s&channel=%s" % (os, channel)
with requests_cache.disabled():
history_text = requests.get(url).text
lines = history_text.strip('\n').split('\n')
expected_fields = ['os', 'channel', 'version', 'timestamp']
releases = read_csv_lines(lines, expected_fields)
return releases
开发者ID:eseidel,项目名称:cycletimes,代码行数:10,代码来源:cycletimes.py
示例14: subscribe_device
def subscribe_device(self, channel, device_type, device_id):
try:
self.clients[channel].append(device_id)
except KeyError:
self.clients[channel] = [device_id]
finally:
url = ACS_URLS['subscribe'].format(self.key)
payload = {'type':device_type, 'device_id':device_id, 'channel':'channel'}
with requests_cache.disabled():
r = requests.post(url, data=payload, cookies=self.cookies)
开发者ID:jpgneves,项目名称:t-10_server,代码行数:10,代码来源:teeminus10_helpers.py
示例15: fetch_csv_from_url
def fetch_csv_from_url(url):
""" Gets a fresh copy of the Google Forms Response file and treats it like a file object.
In order for this to work, the response sheet must be published to the web as csv and the link must be put in config.py under the variable gforms_url.
"""
#cache avoidance.
with requests_cache.disabled():
r = requests.get(url)
if r.status_code == 200:
return r.iter_lines()
开发者ID:technocake,项目名称:ggj-participants-cards,代码行数:11,代码来源:importing.py
示例16: test_response_history
def test_response_history(self):
r1 = requests.get('http://httpbin.org/redirect/3')
def test_redirect_history(url):
r2 = requests.get(url)
for r11, r22 in zip(r1.history, r2.history):
self.assertEqual(r11.url, r22.url)
test_redirect_history('http://httpbin.org/redirect/3')
test_redirect_history('http://httpbin.org/redirect/2')
with requests_cache.disabled():
r3 = requests.get('http://httpbin.org/redirect/1')
self.assertEqual(len(r3.history), 1)
开发者ID:dmr,项目名称:requests-cache,代码行数:11,代码来源:test_cache.py
示例17: request_json_resource_cacheless
def request_json_resource_cacheless(url, params=None, retry=3, time_between=1):
with requests_cache.disabled():
for i in range(retry):
response = requests.get(url, params=params, headers={'Origin':'http://www.lolesports.com'})
if response.status_code == 200:
return response.json(object_pairs_hook=OrderedDict)
elif response.status_code == 404:
break
else:
time.sleep(time_between)
raise Exception('Unable to retrieve json recourse')
开发者ID:devin-fisher,项目名称:treadstone,代码行数:12,代码来源:http_lol_static.py
示例18: get_cloud_cover
def get_cloud_cover(self, city):
'''Gets cloud cover in % for the given city'''
url = API_URLS['weather']['city_search'].format(city)
logger.info("Requesting cloud cover for {0}.".format(city))
with requests_cache.disabled():
r = requests.get(url)
try:
result = json.loads(r.text)
except ValueError:
logger.warning("Could not get cloud cover for {0}".format(city))
return '0'
return result['data']['current_condition'][0]['cloudcover']
开发者ID:jpgneves,项目名称:t-10_server,代码行数:12,代码来源:teeminus10_helpers.py
示例19: fetch
def fetch(self, uri, cache=True):
if cache:
with requests_cache.enabled():
r = requests.get(uri, headers={'Accept' : 'application/json'})
else:
with requests_cache.disabled():
r = requests.get(uri, headers={'Accept' : 'application/json'})
# last uri-part is dataset name and dictionary key
key = uri.split('/')[-1]
return r.json.get(key)
开发者ID:cjseeger,项目名称:aapps,代码行数:12,代码来源:geojson.py
示例20: fetch_branch_release_times
def fetch_branch_release_times():
release_times = {}
# Always grab the most recent release history.
with requests_cache.disabled():
history_text = requests.get(RELEASE_HISTORY_CSV_URL).text
for line in history_text.strip('\n').split('\n'):
os, channel, version, date_string = line.split(',')
date = parse_datetime_ms(date_string)
branch = version.split('.')[2]
last_date = release_times.get(branch)
if not last_date or last_date > date:
release_times[branch] = date
return release_times
开发者ID:kleopatra999,项目名称:cycletimes,代码行数:13,代码来源:cycletimes.py
注:本文中的requests_cache.disabled函数示例由纯净天空整理自Github/MSDocs等源码及文档管理平台,相关代码片段筛选自各路编程大神贡献的开源项目,源码版权归原作者所有,传播和使用请参考对应项目的License;未经允许,请勿转载。 |
请发表评论