本文整理汇总了Python中requests_cache.CachedSession类的典型用法代码示例。如果您正苦于以下问题:Python CachedSession类的具体用法?Python CachedSession怎么用?Python CachedSession使用的例子?那么恭喜您, 这里精选的类代码示例或许可以为您提供帮助。
在下文中一共展示了CachedSession类的20个代码示例,这些例子默认根据受欢迎程度排序。您可以为喜欢或者感觉有用的代码点赞,您的评价将有助于我们的系统推荐出更棒的Python代码示例。
示例1: url_get
def url_get(url):
"""
Download an URL using a cache and return the response object
:param url:
:return:
"""
s = None
info = dict()
if 'file://' in url:
s = requests.session()
s.mount('file://', FileAdapter())
else:
s = CachedSession(cache_name="pyff_cache",
backend=config.request_cache_backend,
expire_after=config.request_cache_time,
old_data_on_error=True)
headers = {'User-Agent': "pyFF/{}".format(__version__), 'Accept': '*/*'}
try:
r = s.get(url, headers=headers, verify=False, timeout=config.request_timeout)
except IOError as ex:
s = requests.Session()
r = s.get(url, headers=headers, verify=False, timeout=config.request_timeout)
if six.PY2:
r.encoding = "utf-8"
log.debug("url_get({}) returns {} chrs encoded as {}".format(url, len(r.content), r.encoding))
if config.request_override_encoding is not None:
r.encoding = config.request_override_encoding
return r
开发者ID:leifj,项目名称:pyFF,代码行数:33,代码来源:utils.py
示例2: test_return_old_data_on_error
def test_return_old_data_on_error(self, datetime_mock):
datetime_mock.utcnow.return_value = datetime.utcnow()
expire_after = 100
url = httpbin("get")
s = CachedSession(CACHE_NAME, CACHE_BACKEND, old_data_on_error=True, expire_after=expire_after)
header = "X-Tst"
def get(n):
return s.get(url, headers={header: n}).json()["headers"][header]
get("expired")
self.assertEquals(get("2"), "expired")
datetime_mock.utcnow.return_value = datetime.utcnow() + timedelta(seconds=expire_after * 2)
with mock.patch.object(s.cache, "save_response", side_effect=Exception):
self.assertEquals(get("3"), "expired")
with mock.patch("requests_cache.core.OriginalSession.send") as send_mock:
resp_mock = requests.Response()
request = requests.Request("GET", url)
resp_mock.request = request.prepare()
resp_mock.status_code = 400
resp_mock._content = '{"other": "content"}'
send_mock.return_value = resp_mock
self.assertEquals(get("3"), "expired")
resp_mock.status_code = 200
self.assertIs(s.get(url).content, resp_mock.content)
# default behaviour
datetime_mock.return_value = datetime.utcnow() + timedelta(seconds=expire_after * 2)
s = CachedSession(CACHE_NAME, CACHE_BACKEND, old_data_on_error=False, expire_after=100)
with mock.patch.object(s.cache, "save_response", side_effect=Exception):
with self.assertRaises(Exception):
s.get(url)
开发者ID:brbsix,项目名称:requests-cache,代码行数:35,代码来源:test_cache.py
示例3: test_post_parameters_normalization
def test_post_parameters_normalization(self):
params = {"a": "a", "b": ["1", "2", "3"], "c": "4"}
url = httpbin("post")
s = CachedSession(CACHE_NAME, CACHE_BACKEND,
allowable_methods=('GET', 'POST'))
self.assertFalse(s.post(url, data=params).from_cache)
self.assertTrue(s.post(url, data=params).from_cache)
self.assertTrue(s.post(url, data=sorted(params.items())).from_cache)
self.assertFalse(s.post(url, data=sorted(params.items(), reverse=True)).from_cache)
开发者ID:brbsix,项目名称:requests-cache,代码行数:9,代码来源:test_cache.py
示例4: test_throttle_cache
def test_throttle_cache(self):
url = httpbin('get')
s = CachedSession(CACHE_NAME, backend=CACHE_BACKEND, expire_after=0.06)
s.throttle(url, 0.5) # one every 2 seconds
r = s.get(url)
time.sleep(0.6)
t = time.time()
r = s.get(url)
delta = time.time() - t
self.assertGreaterEqual(delta, 1)
开发者ID:jherre,项目名称:requests-cache,代码行数:10,代码来源:test_cache.py
示例5: test_expire_cache
def test_expire_cache(self):
delay = 1
url = httpbin('delay/%s' % delay)
s = CachedSession(CACHE_NAME, backend=CACHE_BACKEND, expire_after=0.06)
t = time.time()
r = s.get(url)
delta = time.time() - t
self.assertGreaterEqual(delta, delay)
time.sleep(0.5)
t = time.time()
r = s.get(url)
delta = time.time() - t
self.assertGreaterEqual(delta, delay)
开发者ID:brbsix,项目名称:requests-cache,代码行数:13,代码来源:test_cache.py
示例6: test_ignore_parameters_post_raw
def test_ignore_parameters_post_raw(self):
url = httpbin("post")
ignored_param = "ignored"
raw_data = "raw test data"
s = CachedSession(CACHE_NAME, CACHE_BACKEND,
allowable_methods=('POST'),
ignored_parameters=[ignored_param])
self.assertFalse(s.post(url, data=raw_data).from_cache)
self.assertTrue(s.post(url, data=raw_data).from_cache)
raw_data = "new raw data"
self.assertFalse(s.post(url, data=raw_data).from_cache)
开发者ID:brbsix,项目名称:requests-cache,代码行数:14,代码来源:test_cache.py
示例7: api_get
def api_get(endpoint, query, cache=False):
if not cache:
s = CachedSession()
with s.cache_disabled():
r = s.get(
app.config['API_URL'] + endpoint + '?q=' + urllib.parse.quote_plus(json.dumps(query)),
headers=gen_api_header(),
verify=app.config['VERIFY_SSL'])
else:
r = requests.get(
app.config['API_URL'] + endpoint + '?q=' + urllib.parse.quote_plus(json.dumps(query)),
headers=gen_api_header(),
verify=app.config['VERIFY_SSL'])
if r.status_code == 200:
# If created then it returns the object data
return json.loads(r.text).get('objects')
else:
return {}
开发者ID:TorSpider,项目名称:TorSpider-Frontend,代码行数:18,代码来源:helpers.py
示例8: test_ignore_parameters_get
def test_ignore_parameters_get(self):
url = httpbin("get")
ignored_param = "ignored"
usual_param = "some"
params = {ignored_param: "1", usual_param: "1"}
s = CachedSession(CACHE_NAME, CACHE_BACKEND,
ignored_parameters=[ignored_param])
r = s.get(url, params=params)
self.assertIn(ignored_param, r.json()['args'].keys())
self.assertFalse(r.from_cache)
self.assertTrue(s.get(url, params=params).from_cache)
params[ignored_param] = "new"
self.assertTrue(s.get(url, params=params).from_cache)
params[usual_param] = "new"
self.assertFalse(s.get(url, params=params).from_cache)
开发者ID:brbsix,项目名称:requests-cache,代码行数:20,代码来源:test_cache.py
示例9: test_headers_in_get_query
def test_headers_in_get_query(self):
url = httpbin("get")
s = CachedSession(CACHE_NAME, CACHE_BACKEND, include_get_headers=True)
headers = {"Accept": "text/json"}
self.assertFalse(s.get(url, headers=headers).from_cache)
self.assertTrue(s.get(url, headers=headers).from_cache)
headers["Accept"] = "text/xml"
self.assertFalse(s.get(url, headers=headers).from_cache)
self.assertTrue(s.get(url, headers=headers).from_cache)
headers["X-custom-header"] = "custom"
self.assertFalse(s.get(url, headers=headers).from_cache)
self.assertTrue(s.get(url, headers=headers).from_cache)
self.assertFalse(s.get(url).from_cache)
self.assertTrue(s.get(url).from_cache)
开发者ID:brbsix,项目名称:requests-cache,代码行数:17,代码来源:test_cache.py
示例10: test_ignore_parameters_post_json
def test_ignore_parameters_post_json(self):
url = httpbin("post")
ignored_param = "ignored"
usual_param = "some"
d = {ignored_param: "1", usual_param: "1"}
s = CachedSession(CACHE_NAME, CACHE_BACKEND,
allowable_methods=('POST'),
ignored_parameters=[ignored_param])
r = s.post(url, json=d)
self.assertIn(ignored_param, json.loads(r.json()['data']).keys())
self.assertFalse(r.from_cache)
self.assertTrue(s.post(url, json=d).from_cache)
d[ignored_param] = "new"
self.assertTrue(s.post(url, json=d).from_cache)
d[usual_param] = "new"
self.assertFalse(s.post(url, json=d).from_cache)
开发者ID:brbsix,项目名称:requests-cache,代码行数:21,代码来源:test_cache.py
示例11: test_post_params
def test_post_params(self):
# issue #2
self.s = CachedSession(CACHE_NAME, CACHE_BACKEND,
allowable_methods=('GET', 'POST'))
d = {'param1': 'test1'}
for _ in range(2):
self.assertEqual(self.post(d)['form'], d)
d = {'param1': 'test1', 'param3': 'test3'}
self.assertEqual(self.post(d)['form'], d)
self.assertTrue(self.s.post(httpbin('post'), data=d).from_cache)
d.update({'something': 'else'})
self.assertFalse(self.s.post(httpbin('post'), data=d).from_cache)
开发者ID:brbsix,项目名称:requests-cache,代码行数:14,代码来源:test_cache.py
示例12: test_remove_expired_entries
def test_remove_expired_entries(self, datetime_mock, datetime_mock2):
expire_after = timedelta(minutes=10)
start_time = datetime.utcnow().replace(year=2010, minute=0)
datetime_mock.utcnow.return_value = start_time
datetime_mock2.utcnow.return_value = start_time
s = CachedSession(CACHE_NAME, CACHE_BACKEND, expire_after=expire_after)
s.get(httpbin('get'))
s.get(httpbin('relative-redirect/3'))
datetime_mock.utcnow.return_value = start_time + expire_after * 2
datetime_mock2.utcnow.return_value = datetime_mock.utcnow.return_value
ok_url = 'get?x=1'
s.get(httpbin(ok_url))
self.assertEqual(len(s.cache.responses), 3)
self.assertEqual(len(s.cache.keys_map), 3)
s.remove_expired_responses()
self.assertEqual(len(s.cache.responses), 1)
self.assertEqual(len(s.cache.keys_map), 0)
self.assertIn(ok_url, list(s.cache.responses.values())[0][0].url)
开发者ID:brbsix,项目名称:requests-cache,代码行数:20,代码来源:test_cache.py
示例13: __init__
def __init__(self,
url=None,
max_file_size=None,
allowed_content_types=(),
jsfetcher=None,
marker=r'<meta.*name="Generator".*>'):
logger.info('Running downloader for url: %s' % url)
self.url = str(url)
self.max_file_size = max_file_size or settings.RFDOCS.get('MAX_FILE_SIZE')
self.allowed_content_types = allowed_content_types or settings.RFDOCS.get('ALLOWED_CONTENT_TYPES')
self.s = CachedSession()
self.r = None
self.error = {}
self.jsfetcher = jsfetcher
self.marker = marker
开发者ID:Chaser,项目名称:rfdocs,代码行数:16,代码来源:downloader.py
示例14: test_ignore_cgi_parameter
def test_ignore_cgi_parameter(self):
s = CachedSession(CACHE_NAME, backend=CACHE_BACKEND, expire_after=5)
s.ignore_cgi('foo')
url = httpbin('get?foo=1')
r = s.get(url)
self.assertFalse(r.from_cache)
r = s.get(url)
self.assertTrue(r.from_cache)
url = httpbin('get')
r = s.get(url)
self.assertTrue(r.from_cache)
开发者ID:jherre,项目名称:requests-cache,代码行数:11,代码来源:test_cache.py
示例15: test_ignore_parameters
def test_ignore_parameters(self):
url = httpbin("get")
ignored_param = "ignored"
usual_param = "some"
s = CachedSession(CACHE_NAME, CACHE_BACKEND, ignored_parameters=[ignored_param])
params = {ignored_param: "1", usual_param: "1"}
s.get(url, params=params)
self.assertTrue(s.get(url, params=params).from_cache)
params[ignored_param] = "new"
self.assertTrue(s.get(url, params=params).from_cache)
params[usual_param] = "new"
self.assertFalse(s.get(url, params=params).from_cache)
开发者ID:EricLau2018,项目名称:requests-cache,代码行数:15,代码来源:test_cache.py
示例16: test_post_data
def test_post_data(self):
# issue #2, raw payload
self.s = CachedSession(CACHE_NAME, CACHE_BACKEND,
allowable_methods=('GET', 'POST'))
d1 = json.dumps({'param1': 'test1'})
d2 = json.dumps({'param1': 'test1', 'param2': 'test2'})
d3 = str('some unicode data')
if is_py3:
bin_data = bytes('some binary data', 'utf8')
else:
bin_data = bytes('some binary data')
for d in (d1, d2, d3):
self.assertEqual(self.post(d)['data'], d)
r = self.s.post(httpbin('post'), data=d)
self.assert_(hasattr(r, 'from_cache'))
self.assertEqual(self.post(bin_data)['data'],
bin_data.decode('utf8'))
r = self.s.post(httpbin('post'), data=bin_data)
self.assert_(hasattr(r, 'from_cache'))
开发者ID:brbsix,项目名称:requests-cache,代码行数:21,代码来源:test_cache.py
示例17: test_expire_cache_override_value_error
def test_expire_cache_override_value_error(self):
url = httpbin('delay/1')
s = CachedSession(CACHE_NAME, backend=CACHE_BACKEND, expire_after=1)
with self.assertRaises(ValueError):
s.expire_after(url, 2)
开发者ID:jherre,项目名称:requests-cache,代码行数:5,代码来源:test_cache.py
示例18: CachedSession
#!/usr/bin/env python
# -*- coding: utf-8 -*-
import requests
from requests_cache import CachedSession
cs = CachedSession(allowable_methods=('GET', 'POST'))
cs.cache.clear()
for i in range(2):
r = cs.get("http://httpbin.org/get?p1=v1", params={'p2': 'v2', 'p3': 'cyrЯЯ'})
print r
print r.from_cache
开发者ID:5x5x5x5,项目名称:try_git,代码行数:11,代码来源:sandbox.py
示例19: request
def request(self,*args,**kwargs) :
kwargs.update(self._init["kwargs"])
return CachedSession.request(self,*args,**kwargs)
开发者ID:mvtango,项目名称:secscraper,代码行数:3,代码来源:browser.py
示例20: CachedSession
import sys
import time
from requests.packages.urllib3.util.retry import Retry
from requests.adapters import HTTPAdapter
HEADER = ['url', 'link', 'title', 'description', 'content', 'topics', 'organisations']
parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument('input_file', nargs='?', type=argparse.FileType('r'), default=sys.stdin)
parser.add_argument('--environment', '-e', dest='root_url', default='https://www-origin.staging.publishing.service.gov.uk', help='the environment used to query the search API')
parser.add_argument('--skip', '-s', dest='skip', type=int, default=0, help='Number of input rows to skip. Can be used to resume a partially completed import')
parser.add_argument('--skip-redirects', '-r', dest='skip_redirects', action='store_true', help="Don't test URLs on GOV.UK to resolve redirected links.")
parser.add_argument('--wait-time', '-w', dest='wait_time', type=float, default=0.1, help='Time to wait between each link, to work around rate limiting.')
args = parser.parse_args()
session = CachedSession(cache_name='govuk_cache', backend='sqlite')
retries = Retry(total=5, backoff_factor=args.wait_time, status_forcelist=[ 429 ])
session.mount('http://', HTTPAdapter(max_retries=retries))
session.mount('https://', HTTPAdapter(max_retries=retries))
def test_base_path(original_base_path, args):
"""
Given a base path, try and classify it as valid, redirected, or gone,
so that we can fetch data even when the link has been redirected.
If it can't be retrieved, return None, otherwise return the ultimate base path.
We might include the same document multiple times in our analysis, but
this should only happen for a small amount of links and we can strip
out duplicates later.
开发者ID:alphagov,项目名称:govuk-lda-tagger,代码行数:31,代码来源:import_indexable_content.py
注:本文中的requests_cache.CachedSession类示例由纯净天空整理自Github/MSDocs等源码及文档管理平台,相关代码片段筛选自各路编程大神贡献的开源项目,源码版权归原作者所有,传播和使用请参考对应项目的License;未经允许,请勿转载。 |
请发表评论