Mercurial > repos > jpayne > bioproject_to_srr_2
comparison urllib3/util/response.py @ 7:5eb2d5e3bf22
planemo upload for repository https://toolrepo.galaxytrakr.org/view/jpayne/bioproject_to_srr_2/556cac4fb538
author | jpayne |
---|---|
date | Sun, 05 May 2024 23:32:17 -0400 |
parents | |
children |
comparison
equal
deleted
inserted
replaced
6:b2745907b1eb | 7:5eb2d5e3bf22 |
---|---|
1 from __future__ import annotations | |
2 | |
3 import http.client as httplib | |
4 from email.errors import MultipartInvariantViolationDefect, StartBoundaryNotFoundDefect | |
5 | |
6 from ..exceptions import HeaderParsingError | |
7 | |
8 | |
9 def is_fp_closed(obj: object) -> bool: | |
10 """ | |
11 Checks whether a given file-like object is closed. | |
12 | |
13 :param obj: | |
14 The file-like object to check. | |
15 """ | |
16 | |
17 try: | |
18 # Check `isclosed()` first, in case Python3 doesn't set `closed`. | |
19 # GH Issue #928 | |
20 return obj.isclosed() # type: ignore[no-any-return, attr-defined] | |
21 except AttributeError: | |
22 pass | |
23 | |
24 try: | |
25 # Check via the official file-like-object way. | |
26 return obj.closed # type: ignore[no-any-return, attr-defined] | |
27 except AttributeError: | |
28 pass | |
29 | |
30 try: | |
31 # Check if the object is a container for another file-like object that | |
32 # gets released on exhaustion (e.g. HTTPResponse). | |
33 return obj.fp is None # type: ignore[attr-defined] | |
34 except AttributeError: | |
35 pass | |
36 | |
37 raise ValueError("Unable to determine whether fp is closed.") | |
38 | |
39 | |
40 def assert_header_parsing(headers: httplib.HTTPMessage) -> None: | |
41 """ | |
42 Asserts whether all headers have been successfully parsed. | |
43 Extracts encountered errors from the result of parsing headers. | |
44 | |
45 Only works on Python 3. | |
46 | |
47 :param http.client.HTTPMessage headers: Headers to verify. | |
48 | |
49 :raises urllib3.exceptions.HeaderParsingError: | |
50 If parsing errors are found. | |
51 """ | |
52 | |
53 # This will fail silently if we pass in the wrong kind of parameter. | |
54 # To make debugging easier add an explicit check. | |
55 if not isinstance(headers, httplib.HTTPMessage): | |
56 raise TypeError(f"expected httplib.Message, got {type(headers)}.") | |
57 | |
58 unparsed_data = None | |
59 | |
60 # get_payload is actually email.message.Message.get_payload; | |
61 # we're only interested in the result if it's not a multipart message | |
62 if not headers.is_multipart(): | |
63 payload = headers.get_payload() | |
64 | |
65 if isinstance(payload, (bytes, str)): | |
66 unparsed_data = payload | |
67 | |
68 # httplib is assuming a response body is available | |
69 # when parsing headers even when httplib only sends | |
70 # header data to parse_headers() This results in | |
71 # defects on multipart responses in particular. | |
72 # See: https://github.com/urllib3/urllib3/issues/800 | |
73 | |
74 # So we ignore the following defects: | |
75 # - StartBoundaryNotFoundDefect: | |
76 # The claimed start boundary was never found. | |
77 # - MultipartInvariantViolationDefect: | |
78 # A message claimed to be a multipart but no subparts were found. | |
79 defects = [ | |
80 defect | |
81 for defect in headers.defects | |
82 if not isinstance( | |
83 defect, (StartBoundaryNotFoundDefect, MultipartInvariantViolationDefect) | |
84 ) | |
85 ] | |
86 | |
87 if defects or unparsed_data: | |
88 raise HeaderParsingError(defects=defects, unparsed_data=unparsed_data) | |
89 | |
90 | |
91 def is_response_to_head(response: httplib.HTTPResponse) -> bool: | |
92 """ | |
93 Checks whether the request of a response has been a HEAD-request. | |
94 | |
95 :param http.client.HTTPResponse response: | |
96 Response to check if the originating request | |
97 used 'HEAD' as a method. | |
98 """ | |
99 # FIXME: Can we do this somehow without accessing private httplib _method? | |
100 method_str = response._method # type: str # type: ignore[attr-defined] | |
101 return method_str.upper() == "HEAD" |