def urlparts(self):
""" The :attr:`url` string as an :class:`urlparse.SplitResult` tuple.
The tuple contains (scheme, host, path, query_string and fragment),
but the fragment is always empty because it is not visible to the
server. """
env = self.environ
http = env.get('HTTP_X_FORWARDED_PROTO') \
or env.get('wsgi.url_scheme', 'http')
host = env.get('HTTP_X_FORWARDED_HOST') or env.get('HTTP_HOST')
if not host:
# HTTP 1.1 requires a Host-header. This is for HTTP/1.0 clients.
host = env.get('SERVER_NAME', '127.0.0.1')
port = env.get('SERVER_PORT')
if port and port != ('80' if http == 'http' else '443'):
host += ':' + port
path = urlquote(self.fullpath)
return UrlSplitResult(http, host, path, env.get('QUERY_STRING'), '')
python类SplitResult()的实例源码
def urlparts(self):
""" The :attr:`url` string as an :class:`urlparse.SplitResult` tuple.
The tuple contains (scheme, host, path, query_string and fragment),
but the fragment is always empty because it is not visible to the
server. """
env = self.environ
http = env.get('HTTP_X_FORWARDED_PROTO') \
or env.get('wsgi.url_scheme', 'http')
host = env.get('HTTP_X_FORWARDED_HOST') or env.get('HTTP_HOST')
if not host:
# HTTP 1.1 requires a Host-header. This is for HTTP/1.0 clients.
host = env.get('SERVER_NAME', '127.0.0.1')
port = env.get('SERVER_PORT')
if port and port != ('80' if http == 'http' else '443'):
host += ':' + port
path = urlquote(self.fullpath)
return UrlSplitResult(http, host, path, env.get('QUERY_STRING'), '')
def urlparts(self):
""" The :attr:`url` string as an :class:`urlparse.SplitResult` tuple.
The tuple contains (scheme, host, path, query_string and fragment),
but the fragment is always empty because it is not visible to the
server. """
env = self.environ
http = env.get('HTTP_X_FORWARDED_PROTO') \
or env.get('wsgi.url_scheme', 'http')
host = env.get('HTTP_X_FORWARDED_HOST') or env.get('HTTP_HOST')
if not host:
# HTTP 1.1 requires a Host-header. This is for HTTP/1.0 clients.
host = env.get('SERVER_NAME', '127.0.0.1')
port = env.get('SERVER_PORT')
if port and port != ('80' if http == 'http' else '443'):
host += ':' + port
path = urlquote(self.fullpath)
return UrlSplitResult(http, host, path, env.get('QUERY_STRING'), '')
def urlparts(self):
""" The :attr:`url` string as an :class:`urlparse.SplitResult` tuple.
The tuple contains (scheme, host, path, query_string and fragment),
but the fragment is always empty because it is not visible to the
server. """
env = self.environ
http = env.get('HTTP_X_FORWARDED_PROTO') \
or env.get('wsgi.url_scheme', 'http')
host = env.get('HTTP_X_FORWARDED_HOST') or env.get('HTTP_HOST')
if not host:
# HTTP 1.1 requires a Host-header. This is for HTTP/1.0 clients.
host = env.get('SERVER_NAME', '127.0.0.1')
port = env.get('SERVER_PORT')
if port and port != ('80' if http == 'http' else '443'):
host += ':' + port
path = urlquote(self.fullpath)
return UrlSplitResult(http, host, path, env.get('QUERY_STRING'), '')
def urlparts(self):
""" The :attr:`url` string as an :class:`urlparse.SplitResult` tuple.
The tuple contains (scheme, host, path, query_string and fragment),
but the fragment is always empty because it is not visible to the
server. """
env = self.environ
http = env.get('HTTP_X_FORWARDED_PROTO') \
or env.get('wsgi.url_scheme', 'http')
host = env.get('HTTP_X_FORWARDED_HOST') or env.get('HTTP_HOST')
if not host:
# HTTP 1.1 requires a Host-header. This is for HTTP/1.0 clients.
host = env.get('SERVER_NAME', '127.0.0.1')
port = env.get('SERVER_PORT')
if port and port != ('80' if http == 'http' else '443'):
host += ':' + port
path = urlquote(self.fullpath)
return UrlSplitResult(http, host, path, env.get('QUERY_STRING'), '')
def urlparts(self):
""" The :attr:`url` string as an :class:`urlparse.SplitResult` tuple.
The tuple contains (scheme, host, path, query_string and fragment),
but the fragment is always empty because it is not visible to the
server. """
env = self.environ
http = env.get('HTTP_X_FORWARDED_PROTO') \
or env.get('wsgi.url_scheme', 'http')
host = env.get('HTTP_X_FORWARDED_HOST') or env.get('HTTP_HOST')
if not host:
# HTTP 1.1 requires a Host-header. This is for HTTP/1.0 clients.
host = env.get('SERVER_NAME', '127.0.0.1')
port = env.get('SERVER_PORT')
if port and port != ('80' if http == 'http' else '443'):
host += ':' + port
path = urlquote(self.fullpath)
return UrlSplitResult(http, host, path, env.get('QUERY_STRING'), '')
def urlparts(self):
""" The :attr:`url` string as an :class:`urlparse.SplitResult` tuple.
The tuple contains (scheme, host, path, query_string and fragment),
but the fragment is always empty because it is not visible to the
server. """
env = self.environ
http = env.get('HTTP_X_FORWARDED_PROTO') \
or env.get('wsgi.url_scheme', 'http')
host = env.get('HTTP_X_FORWARDED_HOST') or env.get('HTTP_HOST')
if not host:
# HTTP 1.1 requires a Host-header. This is for HTTP/1.0 clients.
host = env.get('SERVER_NAME', '127.0.0.1')
port = env.get('SERVER_PORT')
if port and port != ('80' if http == 'http' else '443'):
host += ':' + port
path = urlquote(self.fullpath)
return UrlSplitResult(http, host, path, env.get('QUERY_STRING'), '')
def http_request(method, url, data=None, json=None, headers=None):
body = None
if not headers:
headers = {}
if json and isinstance(json, dict):
headers['Content-Type'] = 'application/json'
body = _json.dumps(json)
elif data and isinstance(data, dict):
headers['Content-Type'] = 'application/x-www-form-urlencoded'
body = urlencode(data)
if method not in ['GET', 'POST', 'PUT', 'PATCH', 'DELETE']:
raise Exception('Method %s is not supported' % method)
try:
parsed_url = urlsplit(url)
assert isinstance(parsed_url, SplitResult)
if parsed_url.scheme == 'https':
connection = httplib.HTTPSConnection
elif parsed_url.scheme == 'http':
connection = httplib.HTTPConnection
conn = connection(parsed_url.netloc)
conn.request(
method,
'{}?{}'.format(parsed_url.path, parsed_url.query),
body,
headers
)
res = conn.getresponse()
res_headers = dict(res.getheaders())
res_code = res.status
res_body = res.read()
except Exception:
logger.exception('Something happened while processing the request')
raise Exception('Http request failed')
return res_body, res_code, res_headers
def test_result_pairs(self):
# Check encoding and decoding between result pairs
result_types = [
urlparse.DefragResult,
urlparse.SplitResult,
urlparse.ParseResult,
]
for result_type in result_types:
self._check_result_type(result_type)
def resolve_adapter(uri):
# type: (AdapterSpec) -> BaseAdapter
"""
Given a URI, returns a properly-configured adapter instance.
"""
if isinstance(uri, BaseAdapter):
return uri
parsed = compat.urllib_parse.urlsplit(uri) # type: SplitResult
if not parsed.scheme:
raise with_context(
exc = InvalidUri(
'URI must begin with "<protocol>://" (e.g., "udp://").',
),
context = {
'parsed': parsed,
'uri': uri,
},
)
try:
adapter_type = adapter_registry[parsed.scheme]
except KeyError:
raise with_context(
exc = InvalidUri('Unrecognized protocol {protocol!r}.'.format(
protocol = parsed.scheme,
)),
context = {
'parsed': parsed,
'uri': uri,
},
)
return adapter_type.configure(parsed)
def configure(cls, parsed):
# type: (Union[Text, SplitResult]) -> HttpAdapter
"""
Creates a new instance using the specified URI.
:param parsed:
Result of :py:func:`urllib.parse.urlsplit`.
"""
return cls(parsed)
def __init__(self, url):
if isinstance(url, ParseResult) or isinstance(url, SplitResult):
self.url_parsed = url
self.url_raw = None
else:
self.url_raw = url
self.url_parsed = None
def __init__(self, uri):
# type: (Union[Text, SplitResult]) -> None
super(HttpAdapter, self).__init__()
if isinstance(uri, text_type):
uri = compat.urllib_parse.urlsplit(uri) # type: SplitResult
if uri.scheme not in self.supported_protocols:
raise with_context(
exc = InvalidUri('Unsupported protocol {protocol!r}.'.format(
protocol = uri.scheme,
)),
context = {
'uri': uri,
},
)
if not uri.hostname:
raise with_context(
exc = InvalidUri(
'Empty hostname in URI {uri!r}.'.format(
uri = uri.geturl(),
),
),
context = {
'uri': uri,
},
)
try:
# noinspection PyStatementEffect
uri.port
except ValueError:
raise with_context(
exc = InvalidUri(
'Non-numeric port in URI {uri!r}.'.format(
uri = uri.geturl(),
),
),
context = {
'uri': uri,
},
)
self.uri = uri