#!/usr/bin/env python
# -*- coding: utf-8 -*-
"""
urlfetch
~~~~~~~~~~
An easy to use HTTP client based on httplib.
:copyright: (c) 2011-2026 by Yue Du.
:license: BSD 2-clause License, see LICENSE for more details.
"""
__version__ = "3.0.1"
__author__ = "Yue Du <ifduyue@gmail.com>"
__url__ = "https://github.com/ifduyue/urlfetch"
__license__ = "BSD 2-Clause License"
import zlib
import os, sys, base64, codecs, uuid, stat, time, socket
import ssl
import mimetypes
from os.path import basename, dirname, abspath, join as pathjoin
from functools import partial
from io import BytesIO
import re
import ipaddress
import json
json_dumps = json.dumps
json_loads = json.loads
from http.client import HTTPConnection, HTTPSConnection
from urllib.parse import parse_qs, urlencode, urlsplit, urljoin
import http.cookies as Cookie
import http.cookiejar as cookiejar
__all__ = (
"request",
"fetch",
"Session",
"get",
"head",
"put",
"post",
"delete",
"options",
"trace",
"patch",
"UrlfetchException",
"ContentLimitExceeded",
"URLError",
"ContentDecodingError",
"TooManyRedirects",
"Timeout",
"HTTPError",
)
GET = "GET"
POST = "POST"
PUT = "PUT"
DELETE = "DELETE"
HEAD = "HEAD"
OPTIONS = "OPTIONS"
TRACE = "TRACE"
PATCH = "PATCH"
[docs]
class UrlfetchException(IOError):
"Base exception. All exceptions and errors will subclass from this."
[docs]
class ContentLimitExceeded(UrlfetchException):
"Content length is beyond the limit."
[docs]
class URLError(UrlfetchException, ValueError):
"Error parsing or handling the URL."
[docs]
class ContentDecodingError(UrlfetchException):
"Failed to decode the content."
[docs]
class TooManyRedirects(UrlfetchException):
"""Too many redirects."""
[docs]
class Timeout(UrlfetchException):
"""Request timed out."""
[docs]
class HTTPError(UrlfetchException):
"""HTTP status code is 400 or higher."""
def __init__(self, response):
self.response = response
self.status = self.status_code = response.status
self.reason = response.reason
UrlfetchException.__init__(
self, "%s %s" % (response.status, response.reason)
)
class cached_property:
"""Cached property.
A property that is only computed once per instance and then replaces
itself with an ordinary attribute. Deleting the attribute resets the
property.
"""
def __init__(self, fget, fset=None, fdel=None, doc=None):
self.__get = fget
self.__set = fset
self.__del = fdel
self.__doc__ = doc or fget.__doc__
self.__name__ = fget.__name__
self.__module__ = fget.__module__
def __get__(self, instance, owner):
if instance is None:
# attribute is accessed through the owner class
return self
try:
return instance.__dict__[self.__name__]
except KeyError:
value = instance.__dict__[self.__name__] = self.__get(instance)
return value
def __set__(self, instance, value):
if instance is None:
return self
if self.__set is not None:
value = self.__set(instance, value)
instance.__dict__[self.__name__] = value
def __delete__(self, instance):
if instance is None:
return self
try:
value = instance.__dict__.pop(self.__name__)
except KeyError:
pass
else:
if self.__del is not None:
self.__del(instance, value)
def setter(self, fset):
return self.__class__(self.__get, fset, self.__del)
def deleter(self, fdel):
return self.__class__(self.__get, self.__set, fdel)
##############################################################################
# Core Methods and Classes ####################################################
##############################################################################
[docs]
class Response:
"""A Response object.
>>> import urlfetch
>>> response = urlfetch.get("http://docs.python.org/")
>>> response.total_time
0.033042049407959
>>> response.status, response.reason, response.version
(200, 'OK', 10)
>>> type(response.body), len(response.body)
(<type 'str'>, 8719)
>>> type(response.text), len(response.text)
(<type 'unicode'>, 8719)
>>> response.getheader('server')
'Apache/2.2.16 (Debian)'
>>> response.getheaders()
[
('content-length', '8719'),
('x-cache', 'MISS from localhost'),
('accept-ranges', 'bytes'),
('vary', 'Accept-Encoding'),
('server', 'Apache/2.2.16 (Debian)'),
('last-modified', 'Tue, 26 Jun 2012 19:23:18 GMT'),
('connection', 'close'),
('etag', '"13cc5e4-220f-4c36507ded580"'),
('date', 'Wed, 27 Jun 2012 06:50:30 GMT'),
('content-type', 'text/html'),
('x-cache-lookup', 'MISS from localhost:8080')
]
>>> response.headers
{
'content-length': '8719',
'x-cache': 'MISS from localhost',
'accept-ranges': 'bytes',
'vary': 'Accept-Encoding',
'server': 'Apache/2.2.16 (Debian)',
'last-modified': 'Tue, 26 Jun 2012 19:23:18 GMT',
'connection': 'close',
'etag': '"13cc5e4-220f-4c36507ded580"',
'date': 'Wed, 27 Jun 2012 06:50:30 GMT',
'content-type': 'text/html',
'x-cache-lookup': 'MISS from localhost:8080'
}
:raises: :class:`ContentLimitExceeded`
"""
def __init__(self, r, **kwargs):
self._conn = kwargs.pop("connection", None)
for k in kwargs:
setattr(self, k, kwargs[k])
self._r = r # httplib.HTTPResponse
self.msg = r.msg
#: Status code returned by server.
self.status = r.status
#: An alias of :attr:`status`.
self.status_code = r.status
#: Reason phrase returned by server.
self.reason = r.reason
#: HTTP protocol version used by server.
#: 10 for HTTP/1.0, 11 for HTTP/1.1.
self.version = r.version
#: total time
self.total_time = kwargs.pop("total_time", None)
self.getheader = r.getheader
self.getheaders = r.getheaders
self._content_encoding = self.getheader("content-encoding", None)
self._decoder = None
try:
self.length_limit = int(kwargs.get("length_limit") or 0) or None
except (TypeError, ValueError):
self.length_limit = None
# if content (length) size is more than length_limit, skip
try:
content_length = int(self.getheader("Content-Length", 0) or 0)
except (TypeError, ValueError):
content_length = 0
if self.length_limit and content_length > self.length_limit:
self.close()
raise ContentLimitExceeded(
"Content length is more than %d bytes" % self.length_limit
)
@property
def ok(self):
"""``True`` if :attr:`status` is less than 400."""
return self.status < 400
[docs]
def raise_for_status(self):
"""Raise :class:`HTTPError` if :attr:`status` is 400 or higher."""
if self.status >= 400:
raise HTTPError(self)
[docs]
def read(self, chunk_size=65536):
"""Read content (for streaming and large files)
:arg int chunk_size: size of chunk, default is 65536, i.e. 64KiB.
"""
return self._r.read(chunk_size)
def __iter__(self):
return self
def __next__(self):
chunk = self.read()
if not chunk:
if self._decoder:
chunk = self._decoder.flush()
self._decoder = None
return chunk
else:
raise StopIteration
else:
ce = self._content_encoding
if ce in ("gzip", "deflate"):
if not self._decoder:
if ce == "gzip":
self._decoder = zlib.decompressobj(16 + zlib.MAX_WBITS)
else:
self._decoder = zlib.decompressobj()
try:
return self._decoder.decompress(chunk)
except zlib.error:
self._decoder = zlib.decompressobj(-zlib.MAX_WBITS)
try:
return self._decoder.decompress(chunk)
except (IOError, zlib.error) as e:
self.close()
raise ContentDecodingError(e)
if ce:
self.close()
raise ContentDecodingError("Unknown encoding: %s" % ce)
return chunk
[docs]
def iter_lines(self, delimiter=None):
"""Yield decoded chunks split on ``delimiter`` (default ``\\n``)."""
pending = b""
sep = delimiter if delimiter is not None else b"\n"
for chunk in self:
pending += chunk
lines = pending.split(sep)
pending = lines.pop()
for line in lines:
yield line
if pending:
yield pending
def __enter__(self):
return self
def __exit__(self, exc_type, exc_val, exc_tb):
self.close()
return False
[docs]
@classmethod
def from_httplib(cls, r, **kwargs):
"""Make an :class:`~urlfetch.Response` object from a httplib response
object."""
return cls(r, **kwargs)
[docs]
@cached_property
def body(self):
"""Response body.
:raises: :class:`ContentLimitExceeded`, :class:`ContentDecodingError`
"""
content = []
length = 0
for chunk in self:
content.append(chunk)
length += len(chunk)
if self.length_limit and length > self.length_limit:
self.close()
raise ContentLimitExceeded(
"Content length is more than %d " "bytes" % self.length_limit
)
return b"".join(content)
# compatible with requests
#: An alias of :attr:`body`.
@property
def content(self):
return self.body
[docs]
@cached_property
def encoding(self):
"""Charset from ``Content-Type``, or ``None``."""
header = self.getheader("content-type") or ""
match = re.search(r"charset=([^\s;]+)", header, re.I)
if match:
return match.group(1).strip("\"'")
return None
[docs]
@cached_property
def text(self):
"""Response body in str.
Decoded with :attr:`encoding` if ``Content-Type`` has a charset,
otherwise UTF-8. Raises :class:`ContentDecodingError` on failure.
"""
enc = self.encoding or "utf-8"
try:
return self.content.decode(enc)
except (LookupError, UnicodeDecodeError) as e:
raise ContentDecodingError(e)
[docs]
@cached_property
def json(self):
"""Load response body as json.
:raises: :class:`ContentDecodingError`
"""
try:
return json_loads(self.text)
except Exception as e:
raise ContentDecodingError(e)
[docs]
@cached_property
def cookies(self):
"""Cookies in dict"""
values = []
if hasattr(self.msg, "get_all"):
values = self.msg.get_all("set-cookie") or []
else:
header = self.getheader("set-cookie")
if header:
values = [header]
cookies = {}
for header in values:
c = Cookie.SimpleCookie()
try:
c.load(header)
except Cookie.CookieError:
continue
cookies.update((i.key, i.value) for i in c.values())
return cookies
[docs]
@cached_property
def cookiestring(self):
"""Cookie string"""
return "; ".join("%s=%s" % (k, v) for k, v in self.cookies.items())
[docs]
@cached_property
def links(self):
"""Links parsed from HTTP Link header"""
ret = []
linkheader = self.getheader("link")
if not linkheader:
return ret
for i in linkheader.split(","):
try:
url, params = i.split(";", 1)
except ValueError:
url, params = i, ""
link = {}
link["url"] = url.strip('''<> '"''')
for param in params.split(";"):
try:
k, v = param.split("=")
except ValueError:
break
link[k.strip(''' '"''')] = v.strip(''' '"''')
ret.append(link)
return ret
[docs]
def close(self):
"""Close the connection."""
try:
if self._r is not None:
self._r.close()
finally:
conn = getattr(self, "_conn", None)
self._conn = None
if conn is not None:
try:
conn.close()
except Exception:
pass
def __del__(self):
try:
self.close()
except Exception:
pass
class _CookieRequest:
"""Minimal request object for :mod:`http.cookiejar`."""
def __init__(self, url):
parsed = parse_url(url)
path = parsed["uri"] or "/"
self.full_url = "%s://%s%s" % (parsed["scheme"], parsed["http_host"], path)
self.type = parsed["scheme"]
self.host = parsed["http_host"]
self.origin_req_host = parsed["host"]
self.unverifiable = False
self.headers = {}
self.unredirected_hdrs = {}
def get_full_url(self):
return self.full_url
def get_origin_req_host(self):
return self.origin_req_host
def get_header(self, name, default=None):
return self.headers.get(name, default)
def has_header(self, name):
lname = name.lower()
return any(k.lower() == lname for k in self.headers) or any(
k.lower() == lname for k in self.unredirected_hdrs
)
def add_unredirected_header(self, key, val):
self.unredirected_hdrs[key] = val
class _CookieResponse:
def __init__(self, msg):
self._msg = msg
def info(self):
return self._msg
def _make_cookie(name, value, domain, path="/"):
return cookiejar.Cookie(
0,
name,
str(value),
None,
False,
domain,
True,
domain.startswith("."),
path or "/",
True,
False,
None,
True,
None,
None,
{},
)
[docs]
class Session:
"""A session object.
:class:`urlfetch.Session` holds common headers and a cookiejar.
Cookies from responses, :meth:`putcookie`, and the ``cookies`` argument
are only sent back to matching hosts. A cookie without an explicit
``domain`` is bound to the host of the next request.
:arg dict headers: Init headers.
:arg dict cookies: Init cookies (bound to the first request host).
:arg tuple auth: (username, password) for basic authentication.
"""
def __init__(self, headers=None, cookies=None, auth=None):
"""Init a :class:`~urlfetch.Session` object"""
#: headers
self.headers = {} if headers is None else headers.copy()
self._jar = cookiejar.CookieJar()
self._pending = {} if cookies is None else dict(cookies)
if auth and isinstance(auth, (list, tuple)):
auth = "%s:%s" % tuple(auth)
auth = base64.b64encode(auth.encode("utf-8"))
self.headers["Authorization"] = "Basic " + auth.decode("utf-8")
[docs]
def putcookie(self, key, value="", domain=None, path="/"):
"""Add a cookie.
If *domain* is omitted, the cookie is bound to the host of the
next request.
"""
if domain:
self._jar.set_cookie(_make_cookie(key, value, domain, path))
self._pending.pop(key, None)
else:
self._pending[key] = value
[docs]
def popcookie(self, key):
"""Remove a cookie from pending cookies and the cookiejar."""
if key in self._pending:
return self._pending.pop(key)
for c in list(self._jar):
if c.name == key:
self._jar.clear(c.domain, c.path, c.name)
return c.value
raise KeyError(key)
@property
def cookies(self):
"""Cookies in dict (pending cookies plus those from the cookiejar)."""
d = dict(self._pending)
for c in self._jar:
d[c.name] = c.value
return d
@cookies.setter
def cookies(self, value):
self._pending = {} if value is None else dict(value)
self._jar = cookiejar.CookieJar()
@property
def cookiestring(self):
"""Cookie string.
It's assignalbe, and will change :attr:`~.Session.cookies`
correspondingly.
>>> s = Session()
>>> s.cookiestring = 'foo=bar; 1=2'
>>> s.cookies
{'1': '2', 'foo': 'bar'}
"""
return "; ".join("%s=%s" % (k, v) for k, v in self.cookies.items())
@cookiestring.setter
def cookiestring(self, value):
""""Cookie string setter"""
c = Cookie.SimpleCookie(value)
self.cookies = dict((i.key, i.value) for i in c.values())
[docs]
def snapshot(self):
session = {"headers": self.headers.copy(), "cookies": self.cookies.copy()}
return session
[docs]
def _bind_pending(self, url):
if not self._pending or not url:
return
host = parse_url(url)["host"]
for name, value in self._pending.items():
self._jar.set_cookie(_make_cookie(name, value, host))
self._pending = {}
[docs]
def _store_cookies(self, r):
url = getattr(r, "url", None)
msg = getattr(r, "msg", None)
if not url or msg is None:
return
self._jar.extract_cookies(_CookieResponse(msg), _CookieRequest(url))
[docs]
def request(self, *args, **kwargs):
"""Issue a request."""
headers = self.headers.copy()
url = kwargs.get("url", args[0] if args else None)
self._bind_pending(url)
cookie_header = self._cookie_header(url)
if cookie_header:
headers["Cookie"] = cookie_header
headers.update(kwargs.get("headers") or {})
kwargs["headers"] = headers
r = request(*args, **kwargs)
self._store_cookies(r)
return r
[docs]
def fetch(self, *args, **kwargs):
"""Fetch an URL"""
data = kwargs.get("data", None)
files = kwargs.get("files") or {}
if (
kwargs.get("json") is not None
or (data and isinstance(data, (str, bytes, dict)))
or files
):
return self.post(*args, **kwargs)
return self.get(*args, **kwargs)
[docs]
def get(self, *args, **kwargs):
"""Issue a get request."""
kwargs["method"] = "GET"
return self.request(*args, **kwargs)
[docs]
def post(self, *args, **kwargs):
"""Issue a post request."""
kwargs["method"] = POST
return self.request(*args, **kwargs)
[docs]
def put(self, *args, **kwargs):
"""Issue a put request."""
kwargs["method"] = PUT
return self.request(*args, **kwargs)
[docs]
def delete(self, *args, **kwargs):
"""Issue a delete request."""
kwargs["method"] = DELETE
return self.request(*args, **kwargs)
[docs]
def head(self, *args, **kwargs):
"""Issue a head request."""
kwargs["method"] = HEAD
return self.request(*args, **kwargs)
[docs]
def options(self, *args, **kwargs):
"""Issue a options request."""
kwargs["method"] = OPTIONS
return self.request(*args, **kwargs)
[docs]
def trace(self, *args, **kwargs):
"""Issue a trace request."""
kwargs["method"] = TRACE
return self.request(*args, **kwargs)
[docs]
def patch(self, *args, **kwargs):
"""Issue a patch request."""
kwargs["method"] = PATCH
return self.request(*args, **kwargs)
[docs]
def fetch(*args, **kwargs):
"""fetch an URL.
:func:`~urlfetch.fetch` is a wrapper of :func:`~urlfetch.request`.
It calls :func:`~urlfetch.get` by default. If one of parameter ``data``
or parameter ``files`` is supplied, :func:`~urlfetch.post` is called.
"""
data = kwargs.get("data", None)
files = kwargs.get("files") or {}
if (
kwargs.get("json") is not None
or (data and isinstance(data, (str, bytes, dict)))
or files
):
return post(*args, **kwargs)
return get(*args, **kwargs)
REDIRECT_STATUSES = (301, 302, 303, 307, 308)
PRESERVE_METHOD_REDIRECTS = (307, 308)
def _origin(parsed):
port = parsed["port"]
if port is None:
port = 443 if parsed["scheme"] == "https" else 80
return parsed["scheme"], parsed["host"], port
def _no_proxy_hosts(trust_env):
ignore_hosts = list(PROXY_IGNORE_HOSTS)
if trust_env:
no_proxy = os.getenv("no_proxy") or os.getenv("NO_PROXY")
if no_proxy:
ignore_hosts = [
h.strip()
for h in no_proxy.replace(" ", ",").split(",")
if h.strip()
]
return ignore_hosts
def _close_conn(conn):
if conn is None:
return
try:
conn.close()
except Exception:
pass
def _strip_host_port(value):
value = value.strip()
if value.startswith("["):
end = value.find("]")
if end != -1:
return value[1:end]
return value.strip("[]")
if value.count(":") == 1:
name, port = value.rsplit(":", 1)
if port.isdigit():
return name
return value
def match_no_proxy(host, no_proxy):
"""Return True if *host* should bypass the proxy for a no_proxy rule."""
if not host or not no_proxy:
return False
host = _strip_host_port(host)
if "%" in host and ":" in host:
host = host.split("%", 1)[0]
rule = no_proxy.strip()
if rule == "*":
return True
rule = _strip_host_port(rule)
try:
host_ip = ipaddress.ip_address(host)
except ValueError:
host_ip = None
if "/" in rule:
try:
network = ipaddress.ip_network(rule, strict=False)
except ValueError:
return False
return host_ip is not None and host_ip in network
if host_ip is not None:
try:
return host_ip == ipaddress.ip_address(rule)
except ValueError:
return False
host_l = host.lower().rstrip(".")
rule_l = rule.lower().lstrip(".").rstrip(".")
if not rule_l:
return False
return host_l == rule_l or host_l.endswith("." + rule_l)
def _merge_headers(base, extra):
"""Overlay extra onto base. Header names are case-insensitive.
An existing name keeps its spelling, so later lookups of Host, Cookie,
Authorization, and Content-Type still match. The new value replaces it.
"""
if not extra:
return
folded = {name.lower(): name for name in base}
for name, value in extra.items():
current = folded.get(name.lower())
if current is None:
base[name] = value
folded[name.lower()] = name
else:
base[current] = value
[docs]
def request(
url,
method="GET",
params=None,
data=None,
headers=None,
timeout=None,
files=None,
randua=False,
auth=None,
length_limit=None,
proxies=None,
trust_env=True,
max_redirects=0,
source_address=None,
validate_certificate=None,
ssl_context=None,
json=None,
**kwargs
):
"""request an URL
:arg string url: URL to be fetched.
:arg string method: (optional) HTTP method, one of ``GET``, ``DELETE``,
``HEAD``, ``OPTIONS``, ``PUT``, ``POST``, ``TRACE``,
``PATCH``. ``GET`` is the default.
:arg dict/string params: (optional) Dict or string to attach to url as
querystring.
:arg dict headers: (optional) HTTP request headers.
:arg float timeout: (optional) Timeout in seconds
:arg files: (optional) Files to be sended
:arg randua: (optional) If ``True`` or ``path string``, use a random
user-agent in headers, instead of
``'urlfetch/' + __version__``
:arg tuple auth: (optional) (username, password) for basic authentication
:arg int length_limit: (optional) If ``None``, no limits on content length,
if the limit reached raised exception 'Content length
is more than ...'
:arg dict proxies: (optional) HTTP proxy, like {'http': '127.0.0.1:8888',
'https': '127.0.0.1:563'}
:arg bool trust_env: (optional) If ``True``, urlfetch will get infomations
from env, such as HTTP_PROXY, HTTPS_PROXY
:arg int max_redirects: (integer, optional) Max redirects allowed within a
request. Default is 0, which means redirects are
not allowed.
:arg tuple source_address: (optional) A tuple of (host, port) to
specify the source_address to bind to.
:arg json: (optional) JSON-serializable object sent as the request body
with ``Content-Type: application/json``. Cannot be combined
with ``data`` or ``files``.
:arg ssl.SSLContext ssl_context: (optional) SSL context for HTTPS
connections. Overrides ``validate_certificate``.
:arg bool validate_certificate: (optional) If ``False``, urlfetch skips
all the necessary certificate and hostname checks
:returns: A :class:`~urlfetch.Response` object
:raises: :class:`URLError`, :class:`UrlfetchException`,
:class:`TooManyRedirects`,
"""
def _proxy_auth_header(parsed_proxy):
if not (parsed_proxy and parsed_proxy["username"] and parsed_proxy["password"]):
return None
token = "%s:%s" % (parsed_proxy["username"], parsed_proxy["password"])
token = base64.b64encode(token.encode("utf-8")).decode("utf-8")
return "Basic " + token
def make_connection(conn_type, host, port, timeout, source_address):
"""Return HTTP or HTTPS connection."""
conn_kw = {"timeout": timeout, "source_address": source_address}
ctx = ssl_context
if ctx is None and validate_certificate is False:
ctx = ssl._create_unverified_context()
if conn_type == "http":
return HTTPConnection(host, port, **conn_kw)
if conn_type == "https":
return HTTPSConnection(host, port, context=ctx, **conn_kw)
raise URLError("Unknown Connection Type: %s" % conn_type)
def open_connection(parsed_url):
scheme = parsed_url["scheme"]
proxy = proxies.get(scheme)
if proxy and not any(
match_no_proxy(parsed_url["host"], host) for host in ignore_hosts
):
if "://" not in proxy:
proxy = "http://%s" % proxy
parsed_proxy = parse_url(proxy)
if scheme == "https":
conn = make_connection(
"https",
parsed_proxy["host"],
parsed_proxy["port"],
timeout,
source_address,
)
tunnel_headers = {}
auth = _proxy_auth_header(parsed_proxy)
if auth:
tunnel_headers["Proxy-Authorization"] = auth
conn.set_tunnel(
parsed_url["host"], parsed_url["port"] or 443, tunnel_headers
)
return conn, True, True, None
conn = make_connection(
parsed_proxy["scheme"],
parsed_proxy["host"],
parsed_proxy["port"],
timeout,
source_address,
)
auth = _proxy_auth_header(parsed_proxy)
return conn, True, False, auth
conn = make_connection(
scheme, parsed_url["host"], parsed_url["port"], timeout, source_address
)
return conn, False, False, None
def perform(url, method, data, reqheaders, parsed_url):
conn = None
try:
conn, via_proxy, tunneled, proxy_auth = open_connection(parsed_url)
if via_proxy and not tunneled and proxy_auth:
reqheaders["Proxy-Authorization"] = proxy_auth
else:
reqheaders.pop("Proxy-Authorization", None)
request_url = url if (via_proxy and not tunneled) else parsed_url["uri"]
conn.request(method, request_url, data, reqheaders)
resp = conn.getresponse()
return conn, resp
except socket.timeout as e:
_close_conn(conn)
raise Timeout(e)
except Exception as e:
_close_conn(conn)
if isinstance(e, (Timeout, UrlfetchException)):
raise
raise UrlfetchException(e)
headers = {} if headers is None else headers
files = {} if files is None else files
method = method.upper()
if method not in ALLOWED_METHODS:
raise UrlfetchException("Method should be one of " + ", ".join(ALLOWED_METHODS))
if params:
if isinstance(params, dict):
url = url_concat(url, params)
elif isinstance(params, (str, bytes)):
if url[-1] not in ("?", "&"):
url += "&" if ("?" in url) else "?"
url += params
parsed_url = parse_url(url)
reqheaders = {
"Accept": "*/*",
"Accept-Encoding": "gzip, deflate",
"User-Agent": random_useragent(randua),
"Host": parsed_url["http_host"],
}
# Proxy support
if proxies is None and trust_env:
proxies = PROXIES
if not proxies:
proxies = {}
ignore_hosts = _no_proxy_hosts(trust_env)
if not auth and parsed_url["username"] and parsed_url["password"]:
auth = (parsed_url["username"], parsed_url["password"])
if auth:
if isinstance(auth, (list, tuple)):
auth = "%s:%s" % tuple(auth)
auth = base64.b64encode(auth.encode("utf-8"))
reqheaders["Authorization"] = "Basic " + auth.decode("utf-8")
if json is not None and (data is not None or files):
raise UrlfetchException("json cannot be combined with data or files")
if files:
content_type, data = encode_multipart(data, files)
reqheaders["Content-Type"] = content_type
elif json is not None:
data = json_dumps(json)
reqheaders["Content-Type"] = "application/json"
elif isinstance(data, dict):
data = urlencode(data, True)
if isinstance(data, (str, bytes)) and not files and "Content-Type" not in reqheaders:
# httplib will set 'Content-Length', also you can set it by yourself
reqheaders["Content-Type"] = "application/x-www-form-urlencoded"
# what if the method is GET, HEAD or DELETE
# just do not make so much decisions for users
_merge_headers(reqheaders, headers)
start_time = time.time()
conn, resp = perform(url, method, data, reqheaders, parsed_url)
end_time = time.time()
total_time = end_time - start_time
history = []
response = Response.from_httplib(
resp,
connection=conn,
reqheaders=dict(reqheaders),
length_limit=length_limit,
history=history[:],
url=url,
total_time=total_time,
start_time=start_time,
)
while (
response.status in REDIRECT_STATUSES
and "location" in response.headers
and max_redirects
):
response.body, response.close(), history.append(response)
if len(history) > max_redirects:
raise TooManyRedirects("max_redirects exceeded")
if response.status not in PRESERVE_METHOD_REDIRECTS:
method = "GET"
data = None
reqheaders.pop("Content-Type", None)
reqheaders.pop("Content-Length", None)
location = response.headers["location"]
prev_origin = _origin(parsed_url)
if location[:2] == "//":
url = parsed_url["scheme"] + ":" + location
else:
url = urljoin(url, location)
parsed_url = parse_url(url)
reqheaders["Host"] = parsed_url["http_host"]
reqheaders["Referer"] = response.url
if prev_origin != _origin(parsed_url):
reqheaders.pop("Authorization", None)
reqheaders.pop("Cookie", None)
conn, resp = perform(url, method, data, reqheaders, parsed_url)
response = Response.from_httplib(
resp,
connection=conn,
reqheaders=dict(reqheaders),
length_limit=length_limit,
history=history[:],
url=url,
total_time=total_time,
start_time=start_time,
)
return response
##############################################################################
# Shortcuts and Helpers #######################################################
##############################################################################
def _partial_method(method):
func = partial(request, method=method)
func.__doc__ = "Issue a %s request" % method.lower()
func.__name__ = method.lower()
func.__module__ = request.__module__
return func
get = _partial_method(GET)
post = _partial_method(POST)
put = _partial_method(PUT)
delete = _partial_method(DELETE)
head = _partial_method(HEAD)
options = _partial_method(OPTIONS)
trace = _partial_method(TRACE)
patch = _partial_method(PATCH)
del _partial_method
class ObjectDict(dict):
"""Makes a dictionary behave like an object."""
def __getattr__(self, name):
try:
return self[name]
except KeyError:
raise AttributeError(name)
def __setattr__(self, name, value):
self[name] = value
[docs]
def parse_url(url):
"""Return a dictionary of parsed url
Including scheme, netloc, path, params, query, fragment, uri, username,
password, host, port and http_host
"""
try:
url = str(url)
except UnicodeDecodeError:
pass
if "://" in url:
scheme, url = url.split("://", 1)
else:
scheme = "http"
url = "http://" + url
parsed = urlsplit(url)
r = ObjectDict()
r["scheme"] = scheme
r["netloc"] = parsed.netloc
r["path"] = parsed.path
r["query"] = parsed.query
r["fragment"] = parsed.fragment
r["uri"] = parsed.path
if parsed.query:
r["uri"] += "?" + parsed.query
r["username"] = parsed.username
r["password"] = parsed.password
host = parsed.hostname.encode("idna").decode("utf-8")
r["host"] = r["hostname"] = host
try:
r["port"] = parsed.port
except ValueError:
r["port"] = None
if ":" in r["host"]:
http_host = "[%s]" % r["host"]
else:
http_host = r["host"]
if r["port"]:
r["http_host"] = "%s:%d" % (http_host, r["port"])
else:
r["http_host"] = http_host
return r
[docs]
def get_proxies_from_environ():
"""Get proxies from os.environ."""
proxies = {}
http_proxy = os.getenv("http_proxy") or os.getenv("HTTP_PROXY")
https_proxy = os.getenv("https_proxy") or os.getenv("HTTPS_PROXY")
if http_proxy:
proxies["http"] = http_proxy
if https_proxy:
proxies["https"] = https_proxy
return proxies
[docs]
def mb_code(s, coding=None, errors="replace"):
"""encoding/decoding helper."""
if isinstance(s, str):
return s if coding is None else s.encode(coding, errors=errors)
for c in ("utf-8", "gb2312", "gbk", "gb18030", "big5"):
try:
s = s.decode(c)
return s if coding is None else s.encode(coding, errors=errors)
except (UnicodeDecodeError, LookupError):
pass
return str(s, errors=errors)
[docs]
def random_useragent(filename=True):
"""Returns a User-Agent string randomly from file.
:arg string filename: (Optional) Path to the file from which a random
useragent is generated. By default it's ``True``, a file shipped
with this module will be used.
:returns: An user-agent string.
"""
import random
default_ua = "urlfetch/%s" % __version__
if isinstance(filename, (str, bytes)):
filenames = [filename]
else:
filenames = []
if filename and UAFILE:
filenames.append(UAFILE)
for filename in filenames:
try:
st = os.stat(filename)
if stat.S_ISREG(st.st_mode) and os.access(filename, os.R_OK):
break
except (OSError, AttributeError):
pass
else:
return default_ua
with open(filename, "rb") as f:
filesize = st.st_size
pos = 0
r = random.Random()
# try getting a valid line for no more than 3 times
for i in range(3):
pos += r.randint(0, filesize)
pos %= filesize
f.seek(pos)
# in case we are in middle of a line
f.readline()
line = f.readline()
if not line:
if f.tell() == filesize:
# end of file
f.seek(0)
line = f.readline()
line = line.strip()
if line and line[0] != "#":
return line
return default_ua
[docs]
def url_concat(url, args, keep_existing=True):
"""Concatenate url and argument dictionary
>>> url_concat("http://example.com/foo?a=b", dict(c="d"))
'http://example.com/foo?a=b&c=d'
:arg string url: URL being concat to.
:arg dict args: Args being concat.
:arg bool keep_existing: (Optional) Whether to keep the args which are
alreay in url, default is ``True``.
"""
if not args:
return url
if keep_existing:
if url[-1] not in ("?", "&"):
url += "&" if ("?" in url) else "?"
return url + urlencode(args, True)
else:
url, seq, query = url.partition("?")
query = parse_qs(query, True)
query.update(args)
return url + "?" + urlencode(query, True)
def _multipart_param(name, value):
value = (
str(value).replace("\\", "\\\\").replace('"', '\\"').replace("\r", " ").replace("\n", " ")
)
return '%s="%s"' % (name, value)
[docs]
def choose_boundary():
"""Generate a multipart boundry.
:returns: A boundary string
"""
global BOUNDARY_PREFIX
if BOUNDARY_PREFIX is None:
BOUNDARY_PREFIX = "urlfetch"
try:
uid = repr(os.getuid())
BOUNDARY_PREFIX += "." + uid
except AttributeError:
pass
try:
pid = repr(os.getpid())
BOUNDARY_PREFIX += "." + pid
except AttributeError:
pass
return "%s.%s" % (BOUNDARY_PREFIX, uuid.uuid4().hex)
[docs]
def encode_multipart(data, files):
"""Encode multipart.
:arg dict data: Data to be encoded
:arg dict files: Files to be encoded
:returns: Encoded binary string
:raises: :class:`UrlfetchException`
"""
body = BytesIO()
boundary = choose_boundary()
part_boundary = ("--%s\r\n" % boundary).encode("ascii")
writer = codecs.lookup("utf-8")[3]
if isinstance(data, dict):
for name, values in data.items():
if not isinstance(values, (list, tuple, set)):
# behave like urllib.urlencode(dict, True)
values = (values,)
for value in values:
body.write(part_boundary)
writer(body).write(
"Content-Disposition: form-data; %s\r\n"
% _multipart_param("name", name)
)
body.write(b"Content-Type: text/plain\r\n\r\n")
if isinstance(value, int):
value = str(value)
if isinstance(value, str):
writer(body).write(value)
else:
body.write(value)
body.write(b"\r\n")
for fieldname, f in files.items():
ctype = None
if isinstance(f, tuple):
if len(f) == 2:
filename, f = f
elif len(f) >= 3:
filename, f, ctype = f[0], f[1], f[2]
else:
raise UrlfetchException("file must has filename")
elif hasattr(f, "name"):
filename = basename(f.name)
else:
filename = None
raise UrlfetchException("file must has filename")
if hasattr(f, "read"):
value = f.read()
elif isinstance(f, (str, bytes)):
value = f
else:
value = str(f)
body.write(part_boundary)
if filename:
writer(body).write(
"Content-Disposition: form-data; %s; %s\r\n"
% (
_multipart_param("name", fieldname),
_multipart_param("filename", filename),
)
)
ctype = (
ctype
or mimetypes.guess_type(str(filename))[0]
or "application/octet-stream"
)
body.write(b"Content-Type: %s\r\n\r\n" % ctype.encode("ascii", "replace"))
else:
writer(body).write(
"Content-Disposition: form-data; %s\r\n"
% _multipart_param("name", fieldname)
)
body.write(b"Content-Type: text/plain\r\n\r\n")
if isinstance(value, str):
writer(body).write(value)
else:
body.write(value)
body.write(b"\r\n")
body.write(("--%s--\r\n" % boundary).encode("ascii"))
content_type = "multipart/form-data; boundary=%s" % boundary
return content_type, body.getvalue()
##############################################################################
# Constants and Globals #######################################################
##############################################################################
ALLOWED_METHODS = (GET, DELETE, HEAD, OPTIONS, PUT, POST, TRACE, PATCH)
PROXY_IGNORE_HOSTS = ("127.0.0.1", "localhost")
PROXIES = get_proxies_from_environ()
BOUNDARY_PREFIX = None
UAFILENAME = "urlfetch.useragents.list"
UAFILE = next(
(
i
for i in set(
(
pathjoin(sys.prefix, UAFILENAME),
pathjoin(sys.prefix, "local", UAFILENAME),
pathjoin(dirname(abspath(__file__)), UAFILENAME),
)
)
if os.path.isfile(i)
),
None,
)
