Spaces:
Sleeping
Sleeping
| from __future__ import annotations | |
| import http.client as httplib | |
| from email.errors import MultipartInvariantViolationDefect, StartBoundaryNotFoundDefect | |
| from ..exceptions import HeaderParsingError | |
| def is_fp_closed(obj: object) -> bool: | |
| """ | |
| Checks whether a given file-like object is closed. | |
| :param obj: | |
| The file-like object to check. | |
| """ | |
| try: | |
| # Check `isclosed()` first, in case Python3 doesn't set `closed`. | |
| # GH Issue #928 | |
| return obj.isclosed() # type: ignore[no-any-return, attr-defined] | |
| except AttributeError: | |
| pass | |
| try: | |
| # Check via the official file-like-object way. | |
| return obj.closed # type: ignore[no-any-return, attr-defined] | |
| except AttributeError: | |
| pass | |
| try: | |
| # Check if the object is a container for another file-like object that | |
| # gets released on exhaustion (e.g. HTTPResponse). | |
| return obj.fp is None # type: ignore[attr-defined] | |
| except AttributeError: | |
| pass | |
| raise ValueError("Unable to determine whether fp is closed.") | |
| def assert_header_parsing(headers: httplib.HTTPMessage) -> None: | |
| """ | |
| Asserts whether all headers have been successfully parsed. | |
| Extracts encountered errors from the result of parsing headers. | |
| Only works on Python 3. | |
| :param http.client.HTTPMessage headers: Headers to verify. | |
| :raises urllib3.exceptions.HeaderParsingError: | |
| If parsing errors are found. | |
| """ | |
| # This will fail silently if we pass in the wrong kind of parameter. | |
| # To make debugging easier add an explicit check. | |
| if not isinstance(headers, httplib.HTTPMessage): | |
| raise TypeError(f"expected httplib.Message, got {type(headers)}.") | |
| unparsed_data = None | |
| # get_payload is actually email.message.Message.get_payload; | |
| # we're only interested in the result if it's not a multipart message | |
| if not headers.is_multipart(): | |
| payload = headers.get_payload() | |
| if isinstance(payload, (bytes, str)): | |
| unparsed_data = payload | |
| # httplib is assuming a response body is available | |
| # when parsing headers even when httplib only sends | |
| # header data to parse_headers() This results in | |
| # defects on multipart responses in particular. | |
| # See: https://github.com/urllib3/urllib3/issues/800 | |
| # So we ignore the following defects: | |
| # - StartBoundaryNotFoundDefect: | |
| # The claimed start boundary was never found. | |
| # - MultipartInvariantViolationDefect: | |
| # A message claimed to be a multipart but no subparts were found. | |
| defects = [ | |
| defect | |
| for defect in headers.defects | |
| if not isinstance( | |
| defect, (StartBoundaryNotFoundDefect, MultipartInvariantViolationDefect) | |
| ) | |
| ] | |
| if defects or unparsed_data: | |
| raise HeaderParsingError(defects=defects, unparsed_data=unparsed_data) | |
| def is_response_to_head(response: httplib.HTTPResponse) -> bool: | |
| """ | |
| Checks whether the request of a response has been a HEAD-request. | |
| :param http.client.HTTPResponse response: | |
| Response to check if the originating request | |
| used 'HEAD' as a method. | |
| """ | |
| # FIXME: Can we do this somehow without accessing private httplib _method? | |
| method_str = response._method # type: str # type: ignore[attr-defined] | |
| return method_str.upper() == "HEAD" | |