本文整理汇总了Python中werkzeug.urls.url_decode_stream函数的典型用法代码示例。如果您正苦于以下问题:Python url_decode_stream函数的具体用法?Python url_decode_stream怎么用?Python url_decode_stream使用的例子?那么恭喜您, 这里精选的函数代码示例或许可以为您提供帮助。
在下文中一共展示了url_decode_stream函数的13个代码示例,这些例子默认根据受欢迎程度排序。您可以为喜欢或者感觉有用的代码点赞,您的评价将有助于我们的系统推荐出更棒的Python代码示例。
示例1: _perform_method_overloading
def _perform_method_overloading(self):
"""
Perform method and content type overloading.
Provides support for browser PUT, PATCH, DELETE & other requests,
by specifing a '_method' form field.
Also provides support for browser non-form requests (eg JSON),
by specifing '_content' and '_content_type' form fields.
"""
self._method = super(APIRequest, self).method
self._stream = super(APIRequest, self).stream
self._content_type = self.headers.get("Content-Type")
self._content_length = get_content_length(self.environ)
if self._method == "POST" and self._content_type == "application/x-www-form-urlencoded":
# Read the request data, then push it back onto the stream again.
body = self.get_data()
data = url_decode_stream(io.BytesIO(body))
self._stream = io.BytesIO(body)
if "_method" in data:
# Support browser forms with PUT, PATCH, DELETE & other methods.
self._method = data["_method"]
if "_content" in data and "_content_type" in data:
# Support browser forms with non-form data, such as JSON.
body = data["_content"].encode("utf8")
self._stream = io.BytesIO(body)
self._content_type = data["_content_type"]
self._content_length = len(body)
开发者ID:jrgrafton,项目名称:tweet-debate,代码行数:29,代码来源:request.py
示例2: _parse_urlencoded
def _parse_urlencoded(self, stream, mimetype, content_length, options):
if self.max_form_memory_size is not None and \
content_length > self.max_form_memory_size:
raise RequestEntityTooLarge()
form = url_decode_stream(stream, self.charset,
errors=self.errors, cls=self.cls)
return _empty_stream, form, self.cls()
开发者ID:AnIrishDuck,项目名称:werkzeug,代码行数:7,代码来源:formparser.py
示例3: authenticate
def authenticate(self,
state,
requested_redirect_url: str,
wsgi_environ: Mapping[str, object]) -> Identity:
logger = logging.getLogger(__name__ + '.StashTeam.authenticate')
logger.debug('state = %r', state)
try:
oauth_token, oauth_token_secret = state
except ValueError:
raise AuthenticationError()
req = Request(wsgi_environ, populate_request=False, shallow=True)
args = cast(ImmutableMultiDict, req.args)
logger.debug('req.args = %r', args)
if args.get('oauth_token') != oauth_token:
raise AuthenticationError()
response = self.request(
'POST', self.ACCESS_TOKEN_URL.format(self),
resource_owner_key=oauth_token,
resource_owner_secret=oauth_token_secret
)
access_token = url_decode_stream(response)
logger.debug('access_token = %r', access_token)
response.close()
response = self.request(
'GET', self.USER_URL.format(self),
resource_owner_key=access_token['oauth_token'],
resource_owner_secret=access_token['oauth_token_secret']
)
whoami = response.read().decode('utf-8')
return Identity(
type(self),
self.USER_PROFILE_URL.format(self, whoami),
(access_token['oauth_token'], access_token['oauth_token_secret'])
)
开发者ID:dahlia,项目名称:geofront,代码行数:34,代码来源:stash.py
示例4: authenticate
def authenticate(
self,
state,
requested_redirect_url: str,
wsgi_environ: Mapping[str, object]
) -> Identity:
logger = self.logger.getChild('authenticate')
req = Request(wsgi_environ, populate_request=False, shallow=True)
args = cast(ImmutableMultiDict, req.args)
try:
code = args['code']
if args['state'] != state:
raise AuthenticationError()
except KeyError:
raise AuthenticationError()
data = url_encode({
'client_id': self.client_id,
'client_secret': self.client_secret,
'code': code,
'redirect_uri': requested_redirect_url,
'grant_type': 'authorization_code',
}).encode()
try:
response = urllib.request.urlopen(self.access_token_url, data)
except urllib.error.HTTPError as e:
logger.debug('Response of POST %s (with/ %r): %s\n%s',
self.access_token_url, data, e.code, e.read())
raise
assert isinstance(response, http.client.HTTPResponse), \
'isinstance(response, {0.__module__}.{0.__qualname__})'.format(
type(response))
headers = getattr(response, 'headers') # workaround mypy
content_type = headers['Content-Type']
mimetype, options = parse_options_header(content_type)
if mimetype == 'application/x-www-form-urlencoded':
token_data = url_decode_stream(response)
elif mimetype == 'application/json':
charset = options.get('charset', 'utf-8')
token_data = json.load(
io.TextIOWrapper(cast(IO[bytes], response), encoding=charset)
)
else:
response.close()
raise AuthenticationError(
'{} sent unsupported content type: {}'.format(
self.access_token_url,
content_type
)
)
response.close()
identity = self.determine_identity(token_data['access_token'])
if self.authorize(identity):
return identity
raise AuthenticationError(
self.unauthorized_identity_message_format.format(
identity=identity, team=self
)
)
开发者ID:spoqa,项目名称:geofront,代码行数:58,代码来源:oauth.py
示例5: test_streamed_url_decoding
def test_streamed_url_decoding(self):
item1 = "a" * 100000
item2 = "b" * 400
string = "a=%s&b=%s&c=%s" % (item1, item2, item2)
gen = urls.url_decode_stream(StringIO(string), limit=len(string), return_iterator=True)
self.assert_equal(gen.next(), ("a", item1))
self.assert_equal(gen.next(), ("b", item2))
self.assert_equal(gen.next(), ("c", item2))
self.assert_raises(StopIteration, gen.next)
开发者ID:FakeSherlock,项目名称:Report,代码行数:9,代码来源:urls.py
示例6: test_streamed_url_decoding
def test_streamed_url_decoding(self):
item1 = u'a' * 100000
item2 = u'b' * 400
string = ('a=%s&b=%s&c=%s' % (item1, item2, item2)).encode('ascii')
gen = urls.url_decode_stream(BytesIO(string), limit=len(string),
return_iterator=True)
self.assert_strict_equal(next(gen), ('a', item1))
self.assert_strict_equal(next(gen), ('b', item2))
self.assert_strict_equal(next(gen), ('c', item2))
self.assert_raises(StopIteration, lambda: next(gen))
开发者ID:ArslanRafique,项目名称:werkzeug,代码行数:10,代码来源:urls.py
示例7: test_streamed_url_decoding
def test_streamed_url_decoding(self):
item1 = 'a' * 100000
item2 = 'b' * 400
string = 'a=%s&b=%s&c=%s' % (item1, item2, item2)
gen = urls.url_decode_stream(StringIO(string), limit=len(string),
return_iterator=True)
self.assert_equal(gen.next(), ('a', item1))
self.assert_equal(gen.next(), ('b', item2))
self.assert_equal(gen.next(), ('c', item2))
self.assert_raises(StopIteration, gen.next)
开发者ID:Mirabis,项目名称:usntssearch,代码行数:10,代码来源:urls.py
示例8: parse
def parse(self, stream, content_type, content_length, context=None):
"""Parse the `stream` as a URL encoded form.
:param stream: the stream to be parsed.
:param content_type: the content type of the request payload.
:param content_length: the content length of the request payload.
:param context: a dictionary containing extra context data
that can be useful for parsing.
"""
data = url_decode_stream(stream)
return data.to_dict()
开发者ID:RussellLuo,项目名称:restart,代码行数:11,代码来源:parsers.py
示例9: test_streamed_url_decoding
def test_streamed_url_decoding():
item1 = u"a" * 100000
item2 = u"b" * 400
string = ("a=%s&b=%s&c=%s" % (item1, item2, item2)).encode("ascii")
gen = urls.url_decode_stream(
BytesIO(string), limit=len(string), return_iterator=True
)
strict_eq(next(gen), ("a", item1))
strict_eq(next(gen), ("b", item2))
strict_eq(next(gen), ("c", item2))
pytest.raises(StopIteration, lambda: next(gen))
开发者ID:pallets,项目名称:werkzeug,代码行数:11,代码来源:test_urls.py
示例10: request_authentication
def request_authentication(
self, redirect_url: str
) -> AuthenticationContinuation:
response = self.request('POST', self.REQUEST_TOKEN_URL.format(self))
request_token = url_decode_stream(response)
response.close()
return AuthenticationContinuation(
self.AUTHORIZE_URL.format(self) + '?' + url_encode({
'oauth_token': request_token['oauth_token'],
'oauth_callback': redirect_url
}),
(request_token['oauth_token'], request_token['oauth_token_secret'])
)
开发者ID:dahlia,项目名称:geofront,代码行数:13,代码来源:stash.py
示例11: authenticate
def authenticate(self,
auth_nonce: str,
requested_redirect_url: str,
wsgi_environ: collections.abc.Mapping) -> Identity:
req = Request(wsgi_environ, populate_request=False, shallow=True)
try:
code = req.args['code']
if req.args['state'] != auth_nonce:
raise AuthenticationError()
except KeyError:
raise AuthenticationError()
data = url_encode({
'client_id': self.client_id,
'client_secret': self.client_secret,
'code': code,
'redirect_uri': requested_redirect_url
}).encode()
response = urllib.request.urlopen(self.ACCESS_TOKEN_URL, data)
content_type = response.headers['Content-Type']
mimetype, options = parse_options_header(content_type)
if mimetype == 'application/x-www-form-urlencoded':
token_data = url_decode_stream(response)
elif mimetype == 'application/json':
charset = options.get('charset')
token_data = json.load(
io.TextIOWrapper(response, encoding=charset)
)
else:
response.close()
raise AuthenticationError(
'{} sent unsupported content type: {}'.format(
self.ACCESS_TOKEN_URL,
content_type
)
)
response.close()
user_data = request(token_data['access_token'], self.USER_URL)
identity = Identity(
type(self),
user_data['login'],
token_data['access_token']
)
if self.authorize(identity):
return identity
raise AuthenticationError(
'@{} user is not a member of @{} organization'.format(
user_data['login'],
self.org_login
)
)
开发者ID:kimdwkimdw,项目名称:geofront,代码行数:50,代码来源:github.py
示例12: parser
def parser(stream, **context):
return url_decode_stream(stream)
开发者ID:gitter-badger,项目名称:api-star,代码行数:2,代码来源:parsers.py
示例13: parse
def parse(self, stream, media_type, **options):
return url_decode_stream(stream)
开发者ID:CoffeeBeanSoftware,项目名称:flask-api,代码行数:2,代码来源:parsers.py
注:本文中的werkzeug.urls.url_decode_stream函数示例由纯净天空整理自Github/MSDocs等源码及文档管理平台,相关代码片段筛选自各路编程大神贡献的开源项目,源码版权归原作者所有,传播和使用请参考对应项目的License;未经允许,请勿转载。 |
请发表评论