diff options
Diffstat (limited to '.venv/lib/python3.12/site-packages/requests_toolbelt/utils')
5 files changed, 540 insertions, 0 deletions
diff --git a/.venv/lib/python3.12/site-packages/requests_toolbelt/utils/__init__.py b/.venv/lib/python3.12/site-packages/requests_toolbelt/utils/__init__.py new file mode 100644 index 00000000..e69de29b --- /dev/null +++ b/.venv/lib/python3.12/site-packages/requests_toolbelt/utils/__init__.py diff --git a/.venv/lib/python3.12/site-packages/requests_toolbelt/utils/deprecated.py b/.venv/lib/python3.12/site-packages/requests_toolbelt/utils/deprecated.py new file mode 100644 index 00000000..c935783b --- /dev/null +++ b/.venv/lib/python3.12/site-packages/requests_toolbelt/utils/deprecated.py @@ -0,0 +1,91 @@ +# -*- coding: utf-8 -*- +"""A collection of functions deprecated in requests.utils.""" +import re +import sys + +from requests import utils + +find_charset = re.compile( + br'<meta.*?charset=["\']*(.+?)["\'>]', flags=re.I +).findall + +find_pragma = re.compile( + br'<meta.*?content=["\']*;?charset=(.+?)["\'>]', flags=re.I +).findall + +find_xml = re.compile( + br'^<\?xml.*?encoding=["\']*(.+?)["\'>]' +).findall + + +def get_encodings_from_content(content): + """Return encodings from given content string. + + .. code-block:: python + + import requests + from requests_toolbelt.utils import deprecated + + r = requests.get(url) + encodings = deprecated.get_encodings_from_content(r) + + :param content: bytestring to extract encodings from + :type content: bytes + :return: encodings detected in the provided content + :rtype: list(str) + """ + encodings = (find_charset(content) + find_pragma(content) + + find_xml(content)) + if (3, 0) <= sys.version_info < (4, 0): + encodings = [encoding.decode('utf8') for encoding in encodings] + return encodings + + +def get_unicode_from_response(response): + """Return the requested content back in unicode. + + This will first attempt to retrieve the encoding from the response + headers. If that fails, it will use + :func:`requests_toolbelt.utils.deprecated.get_encodings_from_content` + to determine encodings from HTML elements. + + .. code-block:: python + + import requests + from requests_toolbelt.utils import deprecated + + r = requests.get(url) + text = deprecated.get_unicode_from_response(r) + + :param response: Response object to get unicode content from. + :type response: requests.models.Response + """ + tried_encodings = set() + + # Try charset from content-type + encoding = utils.get_encoding_from_headers(response.headers) + + if encoding: + try: + return str(response.content, encoding) + except UnicodeError: + tried_encodings.add(encoding.lower()) + + encodings = get_encodings_from_content(response.content) + + for _encoding in encodings: + _encoding = _encoding.lower() + if _encoding in tried_encodings: + continue + try: + return str(response.content, _encoding) + except UnicodeError: + tried_encodings.add(_encoding) + + # Fall back: + if encoding: + try: + return str(response.content, encoding, errors='replace') + except TypeError: + pass + return response.text diff --git a/.venv/lib/python3.12/site-packages/requests_toolbelt/utils/dump.py b/.venv/lib/python3.12/site-packages/requests_toolbelt/utils/dump.py new file mode 100644 index 00000000..dec0e376 --- /dev/null +++ b/.venv/lib/python3.12/site-packages/requests_toolbelt/utils/dump.py @@ -0,0 +1,198 @@ +"""This module provides functions for dumping information about responses.""" +import collections + +from requests import compat + + +__all__ = ('dump_response', 'dump_all') + +HTTP_VERSIONS = { + 9: b'0.9', + 10: b'1.0', + 11: b'1.1', +} + +_PrefixSettings = collections.namedtuple('PrefixSettings', + ['request', 'response']) + + +class PrefixSettings(_PrefixSettings): + def __new__(cls, request, response): + request = _coerce_to_bytes(request) + response = _coerce_to_bytes(response) + return super(PrefixSettings, cls).__new__(cls, request, response) + + +def _get_proxy_information(response): + if getattr(response.connection, 'proxy_manager', False): + proxy_info = {} + request_url = response.request.url + if request_url.startswith('https://'): + proxy_info['method'] = 'CONNECT' + + proxy_info['request_path'] = request_url + return proxy_info + return None + + +def _format_header(name, value): + return (_coerce_to_bytes(name) + b': ' + _coerce_to_bytes(value) + + b'\r\n') + + +def _build_request_path(url, proxy_info): + uri = compat.urlparse(url) + proxy_url = proxy_info.get('request_path') + if proxy_url is not None: + request_path = _coerce_to_bytes(proxy_url) + return request_path, uri + + request_path = _coerce_to_bytes(uri.path) + if uri.query: + request_path += b'?' + _coerce_to_bytes(uri.query) + + return request_path, uri + + +def _dump_request_data(request, prefixes, bytearr, proxy_info=None): + if proxy_info is None: + proxy_info = {} + + prefix = prefixes.request + method = _coerce_to_bytes(proxy_info.pop('method', request.method)) + request_path, uri = _build_request_path(request.url, proxy_info) + + # <prefix><METHOD> <request-path> HTTP/1.1 + bytearr.extend(prefix + method + b' ' + request_path + b' HTTP/1.1\r\n') + + # <prefix>Host: <request-host> OR host header specified by user + headers = request.headers.copy() + host_header = _coerce_to_bytes(headers.pop('Host', uri.netloc)) + bytearr.extend(prefix + b'Host: ' + host_header + b'\r\n') + + for name, value in headers.items(): + bytearr.extend(prefix + _format_header(name, value)) + + bytearr.extend(prefix + b'\r\n') + if request.body: + if isinstance(request.body, compat.basestring): + bytearr.extend(prefix + _coerce_to_bytes(request.body)) + else: + # In the event that the body is a file-like object, let's not try + # to read everything into memory. + bytearr.extend(b'<< Request body is not a string-like type >>') + bytearr.extend(b'\r\n') + bytearr.extend(b'\r\n') + + +def _dump_response_data(response, prefixes, bytearr): + prefix = prefixes.response + # Let's interact almost entirely with urllib3's response + raw = response.raw + + # Let's convert the version int from httplib to bytes + version_str = HTTP_VERSIONS.get(raw.version, b'?') + + # <prefix>HTTP/<version_str> <status_code> <reason> + bytearr.extend(prefix + b'HTTP/' + version_str + b' ' + + str(raw.status).encode('ascii') + b' ' + + _coerce_to_bytes(response.reason) + b'\r\n') + + headers = raw.headers + for name in headers.keys(): + for value in headers.getlist(name): + bytearr.extend(prefix + _format_header(name, value)) + + bytearr.extend(prefix + b'\r\n') + + bytearr.extend(response.content) + + +def _coerce_to_bytes(data): + if not isinstance(data, bytes) and hasattr(data, 'encode'): + data = data.encode('utf-8') + # Don't bail out with an exception if data is None + return data if data is not None else b'' + + +def dump_response(response, request_prefix=b'< ', response_prefix=b'> ', + data_array=None): + """Dump a single request-response cycle's information. + + This will take a response object and dump only the data that requests can + see for that single request-response cycle. + + Example:: + + import requests + from requests_toolbelt.utils import dump + + resp = requests.get('https://api.github.com/users/sigmavirus24') + data = dump.dump_response(resp) + print(data.decode('utf-8')) + + :param response: + The response to format + :type response: :class:`requests.Response` + :param request_prefix: (*optional*) + Bytes to prefix each line of the request data + :type request_prefix: :class:`bytes` + :param response_prefix: (*optional*) + Bytes to prefix each line of the response data + :type response_prefix: :class:`bytes` + :param data_array: (*optional*) + Bytearray to which we append the request-response cycle data + :type data_array: :class:`bytearray` + :returns: Formatted bytes of request and response information. + :rtype: :class:`bytearray` + """ + data = data_array if data_array is not None else bytearray() + prefixes = PrefixSettings(request_prefix, response_prefix) + + if not hasattr(response, 'request'): + raise ValueError('Response has no associated request') + + proxy_info = _get_proxy_information(response) + _dump_request_data(response.request, prefixes, data, + proxy_info=proxy_info) + _dump_response_data(response, prefixes, data) + return data + + +def dump_all(response, request_prefix=b'< ', response_prefix=b'> '): + """Dump all requests and responses including redirects. + + This takes the response returned by requests and will dump all + request-response pairs in the redirect history in order followed by the + final request-response. + + Example:: + + import requests + from requests_toolbelt.utils import dump + + resp = requests.get('https://httpbin.org/redirect/5') + data = dump.dump_all(resp) + print(data.decode('utf-8')) + + :param response: + The response to format + :type response: :class:`requests.Response` + :param request_prefix: (*optional*) + Bytes to prefix each line of the request data + :type request_prefix: :class:`bytes` + :param response_prefix: (*optional*) + Bytes to prefix each line of the response data + :type response_prefix: :class:`bytes` + :returns: Formatted bytes of request and response information. + :rtype: :class:`bytearray` + """ + data = bytearray() + + history = list(response.history[:]) + history.append(response) + + for response in history: + dump_response(response, request_prefix, response_prefix, data) + + return data diff --git a/.venv/lib/python3.12/site-packages/requests_toolbelt/utils/formdata.py b/.venv/lib/python3.12/site-packages/requests_toolbelt/utils/formdata.py new file mode 100644 index 00000000..b0a909d2 --- /dev/null +++ b/.venv/lib/python3.12/site-packages/requests_toolbelt/utils/formdata.py @@ -0,0 +1,108 @@ +# -*- coding: utf-8 -*- +"""Implementation of nested form-data encoding function(s).""" +from .._compat import basestring +from .._compat import urlencode as _urlencode + + +__all__ = ('urlencode',) + + +def urlencode(query, *args, **kwargs): + """Handle nested form-data queries and serialize them appropriately. + + There are times when a website expects a nested form data query to be sent + but, the standard library's urlencode function does not appropriately + handle the nested structures. In that case, you need this function which + will flatten the structure first and then properly encode it for you. + + When using this to send data in the body of a request, make sure you + specify the appropriate Content-Type header for the request. + + .. code-block:: python + + import requests + from requests_toolbelt.utils import formdata + + query = { + 'my_dict': { + 'foo': 'bar', + 'biz': 'baz", + }, + 'a': 'b', + } + + resp = requests.get(url, params=formdata.urlencode(query)) + # or + resp = requests.post( + url, + data=formdata.urlencode(query), + headers={ + 'Content-Type': 'application/x-www-form-urlencoded' + }, + ) + + Similarly, you can specify a list of nested tuples, e.g., + + .. code-block:: python + + import requests + from requests_toolbelt.utils import formdata + + query = [ + ('my_list', [ + ('foo', 'bar'), + ('biz', 'baz'), + ]), + ('a', 'b'), + ] + + resp = requests.get(url, params=formdata.urlencode(query)) + # or + resp = requests.post( + url, + data=formdata.urlencode(query), + headers={ + 'Content-Type': 'application/x-www-form-urlencoded' + }, + ) + + For additional parameter and return information, see the official + `urlencode`_ documentation. + + .. _urlencode: + https://docs.python.org/3/library/urllib.parse.html#urllib.parse.urlencode + """ + expand_classes = (dict, list, tuple) + original_query_list = _to_kv_list(query) + + if not all(_is_two_tuple(i) for i in original_query_list): + raise ValueError("Expected query to be able to be converted to a " + "list comprised of length 2 tuples.") + + query_list = original_query_list + while any(isinstance(v, expand_classes) for _, v in query_list): + query_list = _expand_query_values(query_list) + + return _urlencode(query_list, *args, **kwargs) + + +def _to_kv_list(dict_or_list): + if hasattr(dict_or_list, 'items'): + return list(dict_or_list.items()) + return dict_or_list + + +def _is_two_tuple(item): + return isinstance(item, (list, tuple)) and len(item) == 2 + + +def _expand_query_values(original_query_list): + query_list = [] + for key, value in original_query_list: + if isinstance(value, basestring): + query_list.append((key, value)) + else: + key_fmt = key + '[%s]' + value_list = _to_kv_list(value) + query_list.extend((key_fmt % k, v) for k, v in value_list) + return query_list diff --git a/.venv/lib/python3.12/site-packages/requests_toolbelt/utils/user_agent.py b/.venv/lib/python3.12/site-packages/requests_toolbelt/utils/user_agent.py new file mode 100644 index 00000000..e9636a41 --- /dev/null +++ b/.venv/lib/python3.12/site-packages/requests_toolbelt/utils/user_agent.py @@ -0,0 +1,143 @@ +# -*- coding: utf-8 -*- +import collections +import platform +import sys + + +def user_agent(name, version, extras=None): + """Return an internet-friendly user_agent string. + + The majority of this code has been wilfully stolen from the equivalent + function in Requests. + + :param name: The intended name of the user-agent, e.g. "python-requests". + :param version: The version of the user-agent, e.g. "0.0.1". + :param extras: List of two-item tuples that are added to the user-agent + string. + :returns: Formatted user-agent string + :rtype: str + """ + if extras is None: + extras = [] + + return UserAgentBuilder( + name, version + ).include_extras( + extras + ).include_implementation( + ).include_system().build() + + +class UserAgentBuilder(object): + """Class to provide a greater level of control than :func:`user_agent`. + + This is used by :func:`user_agent` to build its User-Agent string. + + .. code-block:: python + + user_agent_str = UserAgentBuilder( + name='requests-toolbelt', + version='17.4.0', + ).include_implementation( + ).include_system( + ).include_extras([ + ('requests', '2.14.2'), + ('urllib3', '1.21.2'), + ]).build() + + """ + + format_string = '%s/%s' + + def __init__(self, name, version): + """Initialize our builder with the name and version of our user agent. + + :param str name: + Name of our user-agent. + :param str version: + The version string for user-agent. + """ + self._pieces = collections.deque([(name, version)]) + + def build(self): + """Finalize the User-Agent string. + + :returns: + Formatted User-Agent string. + :rtype: + str + """ + return " ".join([self.format_string % piece for piece in self._pieces]) + + def include_extras(self, extras): + """Include extra portions of the User-Agent. + + :param list extras: + list of tuples of extra-name and extra-version + """ + if any(len(extra) != 2 for extra in extras): + raise ValueError('Extras should be a sequence of two item tuples.') + + self._pieces.extend(extras) + return self + + def include_implementation(self): + """Append the implementation string to the user-agent string. + + This adds the the information that you're using CPython 2.7.13 to the + User-Agent. + """ + self._pieces.append(_implementation_tuple()) + return self + + def include_system(self): + """Append the information about the Operating System.""" + self._pieces.append(_platform_tuple()) + return self + + +def _implementation_tuple(): + """Return the tuple of interpreter name and version. + + Returns a string that provides both the name and the version of the Python + implementation currently running. For example, on CPython 2.7.5 it will + return "CPython/2.7.5". + + This function works best on CPython and PyPy: in particular, it probably + doesn't work for Jython or IronPython. Future investigation should be done + to work out the correct shape of the code for those platforms. + """ + implementation = platform.python_implementation() + + if implementation == 'CPython': + implementation_version = platform.python_version() + elif implementation == 'PyPy': + implementation_version = '%s.%s.%s' % (sys.pypy_version_info.major, + sys.pypy_version_info.minor, + sys.pypy_version_info.micro) + if sys.pypy_version_info.releaselevel != 'final': + implementation_version = ''.join([ + implementation_version, sys.pypy_version_info.releaselevel + ]) + elif implementation == 'Jython': + implementation_version = platform.python_version() # Complete Guess + elif implementation == 'IronPython': + implementation_version = platform.python_version() # Complete Guess + else: + implementation_version = 'Unknown' + + return (implementation, implementation_version) + + +def _implementation_string(): + return "%s/%s" % _implementation_tuple() + + +def _platform_tuple(): + try: + p_system = platform.system() + p_release = platform.release() + except IOError: + p_system = 'Unknown' + p_release = 'Unknown' + return (p_system, p_release) |