本文整理汇总了Python中requests.models.PreparedRequest类的典型用法代码示例。如果您正苦于以下问题:Python PreparedRequest类的具体用法?Python PreparedRequest怎么用?Python PreparedRequest使用的例子?那么恭喜您, 这里精选的类代码示例或许可以为您提供帮助。
在下文中一共展示了PreparedRequest类的18个代码示例,这些例子默认根据受欢迎程度排序。您可以为喜欢或者感觉有用的代码点赞,您的评价将有助于我们的系统推荐出更棒的Python代码示例。
示例1: deserialize_prepared_request
def deserialize_prepared_request(serialized):
p = PreparedRequest()
p.body = serialized['body']
p.headers = CaseInsensitiveDict(serialized['headers'])
p.method = serialized['method']
p.url = serialized['uri']
return p
开发者ID:pombredanne,项目名称:betamax,代码行数:7,代码来源:cassette.py
示例2: test_prepared_copy
def test_prepared_copy(kwargs):
p = PreparedRequest()
if kwargs:
p.prepare(**kwargs)
copy = p.copy()
for attr in ('method', 'url', 'headers', '_cookies', 'body', 'hooks'):
assert getattr(p, attr) == getattr(copy, attr)
开发者ID:503945930,项目名称:requests,代码行数:7,代码来源:test_requests.py
示例3: check_url
def check_url(url):
prepared_request = PreparedRequest()
try:
prepared_request.prepare_url(url, None)
return True
except requests.exceptions.MissingSchema:
return False
开发者ID:tehnotcpu,项目名称:CirnoBot,代码行数:7,代码来源:utils.py
示例4: test_no_keep_alive_by_default
def test_no_keep_alive_by_default(self):
p = PreparedRequest()
p.prepare(
method='GET',
url='http://www.example.com',
hooks=default_hooks()
)
assert 'Connection' not in p.headers
开发者ID:tychotatitscheff,项目名称:requests,代码行数:8,代码来源:test_requests.py
示例5: test_data_argument_accepts_tuples
def test_data_argument_accepts_tuples(list_of_tuples):
"""
Ensure that the data argument will accept tuples of strings
and properly encode them.
"""
for data in list_of_tuples:
p = PreparedRequest()
p.prepare(method="GET", url="http://www.example.com", data=data, hooks=default_hooks())
assert p.body == urlencode(data)
开发者ID:RRedwards,项目名称:requests,代码行数:9,代码来源:test_requests.py
示例6: test_prepared_request_no_cookies_copy
def test_prepared_request_no_cookies_copy():
p = PreparedRequest()
p.prepare(
method='GET',
url='http://www.example.com',
data='foo=bar',
hooks=default_hooks()
)
assert_copy(p, p.copy())
开发者ID:Hammad04,项目名称:payeezy_direct_API,代码行数:9,代码来源:test_requests.py
示例7: test_data_argument_accepts_tuples
def test_data_argument_accepts_tuples(data):
"""Ensure that the data argument will accept tuples of strings
and properly encode them.
"""
p = PreparedRequest()
p.prepare(
method='GET',
url='http://www.example.com',
data=data,
hooks=default_hooks()
)
assert p.body == urlencode(data)
开发者ID:503945930,项目名称:requests,代码行数:12,代码来源:test_requests.py
示例8: oauth_access_token_url
def oauth_access_token_url(self):
"""Generate the OAuth access token url."""
url = self._access_token_url.format(base_url=self.base_url)
params = [
('client_id', self.credentials.api_key),
('client_secret', self.credentials.secret),
('code', self.credentials.code)
]
parser = PreparedRequest()
parser.prepare_url(url=url, params=params)
return parser.url
开发者ID:RafaAguilar,项目名称:shopify-trois,代码行数:15,代码来源:oauth_engine.py
示例9: _format_args
def _format_args(self):
"""将params/data的数据拼接在url里 """
if not self.record_params:
return ''
args_text = ""
try:
sign = "&" if '?' in self.url else "?"
if self.params:
args_text = sign + PreparedRequest._encode_params(self.params)
elif self.data:
args_text = sign + PreparedRequest._encode_params(self.data)
except:
pass
return args_text
开发者ID:a1401358759,项目名称:my_site,代码行数:16,代码来源:client.py
示例10: oauth_authorize_url
def oauth_authorize_url(self, redirect_to=None):
"""Generates the oauth authorize url.
:param redirect_to: URL shopify will redirect to once authorized.
"""
url = self._authorize_url.format(base_url=self.base_url)
params = [
('client_id', self.credentials.api_key),
('scope', ",".join(self.credentials.scope)),
('redirect_uri', redirect_to)
]
request = PreparedRequest()
request.prepare_url(url=url, params=params)
return request.url
开发者ID:RafaAguilar,项目名称:shopify-trois,代码行数:18,代码来源:oauth_engine.py
示例11: prepare_request
def prepare_request(self, request):
"""Constructs a :class:`PreparedRequest <PreparedRequest>` for
transmission and returns it. The :class:`PreparedRequest` has settings
merged from the :class:`Request <Request>` instance and those of the
:class:`Session`.
:param request: :class:`Request` instance to prepare with this
session's settings.
"""
cookies = request.cookies or {}
# Bootstrap CookieJar.
if not isinstance(cookies, cookielib.CookieJar):
cookies = cookiejar_from_dict(cookies)
# Merge with session cookies
merged_cookies = RequestsCookieJar()
merged_cookies.update(self.cookies)
merged_cookies.update(cookies)
# Set environment's basic authentication if not explicitly set.
auth = request.auth
if self.trust_env and not auth and not self.auth:
auth = get_netrc_auth(request.url)
p = PreparedRequest()
p.prepare(
method=request.method.upper(),
url=request.url,
files=request.files,
data=request.data,
json=request.json,
headers=merge_setting(request.headers, self.headers, dict_class=CaseInsensitiveDict),
params=merge_setting(request.params, self.params),
auth=merge_setting(auth, self.auth),
cookies=merged_cookies,
hooks=merge_hooks(request.hooks, self.hooks),
)
return p
开发者ID:P4ncake,项目名称:weboob,代码行数:40,代码来源:sessions.py
示例12: copy_request
def copy_request(request):
"""Copy a Requests PreparedRequest."""
new_request = PreparedRequest()
new_request.method = request.method
new_request.url = request.url
new_request.body = request.body
new_request.hooks = request.hooks
new_request.headers = request.headers.copy()
return new_request
开发者ID:mdengler,项目名称:requests-ntlm,代码行数:11,代码来源:requests_ntlm.py
示例13: deserialize_prepared_request
def deserialize_prepared_request(serialized):
p = PreparedRequest()
p._cookies = RequestsCookieJar()
body = serialized['body']
if isinstance(body, dict):
original_body = body.get('string')
p.body = original_body or base64.b64decode(
body.get('base64_string', '').encode())
else:
p.body = body
h = [(k, from_list(v)) for k, v in serialized['headers'].items()]
p.headers = CaseInsensitiveDict(h)
p.method = serialized['method']
p.url = serialized['uri']
return p
开发者ID:bboe,项目名称:betamax,代码行数:15,代码来源:util.py
示例14: load_resource
def load_resource(self, cdx, params):
load_url = cdx.get('load_url')
if not load_url:
return None
if params.get('content_type') == VideoLoader.CONTENT_TYPE:
return None
input_req = params['_input_req']
req_headers = input_req.get_req_headers()
dt = timestamp_to_datetime(cdx['timestamp'])
if cdx.get('memento_url'):
req_headers['Accept-Datetime'] = datetime_to_http_date(dt)
method = input_req.get_req_method()
data = input_req.get_req_body()
p = PreparedRequest()
p.prepare_url(load_url, None)
p.prepare_headers(None)
p.prepare_auth(None, load_url)
auth = p.headers.get('Authorization')
if auth:
req_headers['Authorization'] = auth
load_url = p.url
try:
upstream_res = self.pool.urlopen(method=method,
url=load_url,
body=data,
headers=req_headers,
redirect=False,
assert_same_host=False,
preload_content=False,
decode_content=False,
retries=self.num_retries,
timeout=params.get('_timeout'))
except Exception as e:
raise LiveResourceException(load_url)
memento_dt = upstream_res.headers.get('Memento-Datetime')
if memento_dt:
dt = http_date_to_datetime(memento_dt)
cdx['timestamp'] = datetime_to_timestamp(dt)
elif cdx.get('memento_url'):
# if 'memento_url' set and no Memento-Datetime header present
# then its an error
return None
agg_type = upstream_res.headers.get('WebAgg-Type')
if agg_type == 'warc':
cdx['source'] = unquote(upstream_res.headers.get('WebAgg-Source-Coll'))
return None, upstream_res.headers, upstream_res
self.raise_on_self_redirect(params, cdx,
str(upstream_res.status),
upstream_res.headers.get('Location'))
if upstream_res.version == 11:
version = '1.1'
else:
version = '1.0'
status = 'HTTP/{version} {status} {reason}\r\n'
status = status.format(version=version,
status=upstream_res.status,
reason=upstream_res.reason)
http_headers_buff = status
orig_resp = upstream_res._original_response
try: #pragma: no cover
#PY 3
resp_headers = orig_resp.headers._headers
for n, v in resp_headers:
if n.lower() in self.SKIP_HEADERS:
continue
http_headers_buff += n + ': ' + v + '\r\n'
except: #pragma: no cover
#PY 2
resp_headers = orig_resp.msg.headers
for n, v in zip(orig_resp.getheaders(), resp_headers):
if n in self.SKIP_HEADERS:
continue
http_headers_buff += v
http_headers_buff += '\r\n'
http_headers_buff = http_headers_buff.encode('latin-1')
try:
#.........这里部分代码省略.........
开发者ID:ikreymer,项目名称:webrec-platform,代码行数:101,代码来源:responseloader.py
示例15: _request_url
def _request_url(cls, url, params):
pre = PreparedRequest()
pre.prepare_url(url, params)
return pre.url
开发者ID:qq40660,项目名称:ChinaAPI,代码行数:4,代码来源:api.py
示例16: test_prepared_request_empty_copy
def test_prepared_request_empty_copy():
p = PreparedRequest()
assert_copy(p, p.copy())
开发者ID:RRedwards,项目名称:requests,代码行数:3,代码来源:test_requests.py
示例17: test_prepared_request_complete_copy
def test_prepared_request_complete_copy():
p = PreparedRequest()
p.prepare(method="GET", url="http://www.example.com", data="foo=bar", hooks=default_hooks(), cookies={"foo": "bar"})
assert_copy(p, p.copy())
开发者ID:RRedwards,项目名称:requests,代码行数:4,代码来源:test_requests.py
示例18: load_resource
def load_resource(self, cdx, params):
load_url = cdx.get('load_url')
if not load_url:
return None
if params.get('content_type') == VideoLoader.CONTENT_TYPE:
return None
if self.forward_proxy_prefix and not cdx.get('is_live'):
load_url = self.forward_proxy_prefix + load_url
input_req = params['_input_req']
req_headers = input_req.get_req_headers()
dt = timestamp_to_datetime(cdx['timestamp'])
if cdx.get('memento_url'):
req_headers['Accept-Datetime'] = datetime_to_http_date(dt)
method = input_req.get_req_method()
data = input_req.get_req_body()
p = PreparedRequest()
try:
p.prepare_url(load_url, None)
except:
raise LiveResourceException(load_url)
p.prepare_headers(None)
p.prepare_auth(None, load_url)
auth = p.headers.get('Authorization')
if auth:
req_headers['Authorization'] = auth
load_url = p.url
# host is set to the actual host for live loading
# ensure it is set to the load_url host
if not cdx.get('is_live'):
#req_headers.pop('Host', '')
req_headers['Host'] = urlsplit(p.url).netloc
referrer = cdx.get('set_referrer')
if referrer:
req_headers['Referer'] = referrer
upstream_res = self._do_request_with_redir_check(method, load_url,
data, req_headers,
params, cdx)
memento_dt = upstream_res.headers.get('Memento-Datetime')
if memento_dt:
dt = http_date_to_datetime(memento_dt)
cdx['timestamp'] = datetime_to_timestamp(dt)
elif cdx.get('memento_url'):
# if 'memento_url' set and no Memento-Datetime header present
# then its an error
return None
agg_type = upstream_res.headers.get('Warcserver-Type')
if agg_type == 'warc':
cdx['source'] = unquote(upstream_res.headers.get('Warcserver-Source-Coll'))
return None, upstream_res.headers, upstream_res
if upstream_res.version == 11:
version = '1.1'
else:
version = '1.0'
status = 'HTTP/{version} {status} {reason}\r\n'
status = status.format(version=version,
status=upstream_res.status,
reason=upstream_res.reason)
http_headers_buff = status
orig_resp = upstream_res._original_response
try: #pragma: no cover
#PY 3
resp_headers = orig_resp.headers._headers
for n, v in resp_headers:
nl = n.lower()
if nl in self.SKIP_HEADERS:
continue
if nl in self.UNREWRITE_HEADERS:
v = self.unrewrite_header(cdx, v)
http_headers_buff += n + ': ' + v + '\r\n'
http_headers_buff += '\r\n'
try:
# http headers could be encoded as utf-8 (though non-standard)
# first try utf-8 encoding
http_headers_buff = http_headers_buff.encode('utf-8')
except:
# then, fall back to latin-1
#.........这里部分代码省略.........
开发者ID:ikreymer,项目名称:pywb,代码行数:101,代码来源:responseloader.py
注:本文中的requests.models.PreparedRequest类示例由纯净天空整理自Github/MSDocs等源码及文档管理平台,相关代码片段筛选自各路编程大神贡献的开源项目,源码版权归原作者所有,传播和使用请参考对应项目的License;未经允许,请勿转载。 |
请发表评论