Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def __init__(self, *args, **kwargs):
super(Session, self).__init__(*args, **kwargs)
# Additionally to mounting our variant of the usual HTTP and HTTPS
# adapter, also mount it for some variants of the default schemes that
# are limited to some specific address family only
adapter = HTTPAdapter()
for scheme in ("http", "https"):
self.mount("{0}://".format(scheme), adapter)
for name in AF2NAME.values():
self.mount("{0}+{1}://".format(scheme, name), adapter)
def request(method, url, **kwargs):
with Session() as session:
return session.request(method=method, url=url, **kwargs)
def request(self, method, url, *args, **kwargs):
family = kwargs.pop("family", socket.AF_UNSPEC)
if family != socket.AF_UNSPEC:
# Inject provided address family value as extension to scheme
url = urllib.parse.urlparse(url)
url = url._replace(scheme="{0}+{1}".format(url.scheme, AF2NAME[int(family)]))
url = url.geturl()
return super(Session, self).request(method, url, *args, **kwargs)
for scheme in ("http", "https"):
self.mount("{0}://".format(scheme), adapter)
for name in AF2NAME.values():
self.mount("{0}+{1}://".format(scheme, name), adapter)
def request(self, method, url, *args, **kwargs):
family = kwargs.pop("family", socket.AF_UNSPEC)
if family != socket.AF_UNSPEC:
# Inject provided address family value as extension to scheme
url = urllib.parse.urlparse(url)
url = url._replace(scheme="{0}+{1}".format(url.scheme, AF2NAME[int(family)]))
url = url.geturl()
return super(Session, self).request(method, url, *args, **kwargs)
session = Session
# Import other `requests` stuff to make the top-level API of this more compatible
from requests import (
__title__, __description__, __url__, __version__, __build__, __author__,
__author_email__, __license__, __copyright__, __cake__,
exceptions, utils, packages, codes,
Request, Response, PreparedRequest,
RequestException, Timeout, URLRequired, TooManyRedirects, HTTPError,
ConnectionError, FileModeWarning, ConnectTimeout, ReadTimeout
)
# Re-implement the top-level โsession-lessโ API
def request(method, url, **kwargs):