Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.8/site-packages/urllib3/__init__.py: 71%

1""" 

2Python HTTP library with thread-safe connection pooling, file post support, user friendly, and more 

3""" 

4 

5from __future__ import annotations 

6 

7# Set default logging handler to avoid "No handler found" warnings. 

8import logging 

9import typing 

10import warnings 

11from logging import NullHandler 

12 

13from . import exceptions 

14from ._base_connection import _TYPE_BODY 

15from ._collections import HTTPHeaderDict 

16from ._version import __version__ 

17from .connectionpool import HTTPConnectionPool, HTTPSConnectionPool, connection_from_url 

18from .filepost import _TYPE_FIELDS, encode_multipart_formdata 

19from .poolmanager import PoolManager, ProxyManager, proxy_from_url 

20from .response import BaseHTTPResponse, HTTPResponse 

21from .util.request import make_headers 

22from .util.retry import Retry 

23from .util.timeout import Timeout 

24 

25# Ensure that Python is compiled with OpenSSL 1.1.1+ 

26# If the 'ssl' module isn't available at all that's 

27# fine, we only care if the module is available. 

28try: 

29 import ssl 

30except ImportError: 

31 pass 

32else: 

33 if not ssl.OPENSSL_VERSION.startswith("OpenSSL "): # Defensive: 

34 warnings.warn( 

35 "urllib3 v2 only supports OpenSSL 1.1.1+, currently " 

36 f"the 'ssl' module is compiled with {ssl.OPENSSL_VERSION!r}. " 

37 "See: https://github.com/urllib3/urllib3/issues/3020", 

38 exceptions.NotOpenSSLWarning, 

39 ) 

40 elif ssl.OPENSSL_VERSION_INFO < (1, 1, 1): # Defensive: 

41 raise ImportError( 

42 "urllib3 v2 only supports OpenSSL 1.1.1+, currently " 

43 f"the 'ssl' module is compiled with {ssl.OPENSSL_VERSION!r}. " 

44 "See: https://github.com/urllib3/urllib3/issues/2168" 

45 ) 

46 

47__author__ = "Andrey Petrov (andrey.petrov@shazow.net)" 

48__license__ = "MIT" 

49__version__ = __version__ 

50 

51__all__ = ( 

52 "HTTPConnectionPool", 

53 "HTTPHeaderDict", 

54 "HTTPSConnectionPool", 

55 "PoolManager", 

56 "ProxyManager", 

57 "HTTPResponse", 

58 "Retry", 

59 "Timeout", 

60 "add_stderr_logger", 

61 "connection_from_url", 

62 "disable_warnings", 

63 "encode_multipart_formdata", 

64 "make_headers", 

65 "proxy_from_url", 

66 "request", 

67 "BaseHTTPResponse", 

68) 

69 

70logging.getLogger(__name__).addHandler(NullHandler()) 

71 

72 

73def add_stderr_logger( 

74 level: int = logging.DEBUG, 

75) -> logging.StreamHandler[typing.TextIO]: 

76 """ 

77 Helper for quickly adding a StreamHandler to the logger. Useful for 

78 debugging. 

79 

80 Returns the handler after adding it. 

81 """ 

82 # This method needs to be in this __init__.py to get the __name__ correct 

83 # even if urllib3 is vendored within another package. 

84 logger = logging.getLogger(__name__) 

85 handler = logging.StreamHandler() 

86 handler.setFormatter(logging.Formatter("%(asctime)s %(levelname)s %(message)s")) 

87 logger.addHandler(handler) 

88 logger.setLevel(level) 

89 logger.debug("Added a stderr logging handler to logger: %s", __name__) 

90 return handler 

91 

92 

93# ... Clean up. 

94del NullHandler 

95 

96 

97# All warning filters *must* be appended unless you're really certain that they 

98# shouldn't be: otherwise, it's very hard for users to use most Python 

99# mechanisms to silence them. 

100# SecurityWarning's always go off by default. 

101warnings.simplefilter("always", exceptions.SecurityWarning, append=True) 

102# InsecurePlatformWarning's don't vary between requests, so we keep it default. 

103warnings.simplefilter("default", exceptions.InsecurePlatformWarning, append=True) 

104 

105 

106def disable_warnings(category: type[Warning] = exceptions.HTTPWarning) -> None: 

107 """ 

108 Helper for quickly disabling all urllib3 warnings. 

109 """ 

110 warnings.simplefilter("ignore", category) 

111 

112 

113_DEFAULT_POOL = PoolManager() 

114 

115 

116def request( 

117 method: str, 

118 url: str, 

119 *, 

120 body: _TYPE_BODY | None = None, 

121 fields: _TYPE_FIELDS | None = None, 

122 headers: typing.Mapping[str, str] | None = None, 

123 preload_content: bool | None = True, 

124 decode_content: bool | None = True, 

125 redirect: bool | None = True, 

126 retries: Retry | bool | int | None = None, 

127 timeout: Timeout | float | int | None = 3, 

128 json: typing.Any | None = None, 

129) -> BaseHTTPResponse: 

130 """ 

131 A convenience, top-level request method. It uses a module-global ``PoolManager`` instance. 

132 Therefore, its side effects could be shared across dependencies relying on it. 

133 To avoid side effects create a new ``PoolManager`` instance and use it instead. 

134 The method does not accept low-level ``**urlopen_kw`` keyword arguments. 

135 """ 

136 

137 return _DEFAULT_POOL.request( 

138 method, 

139 url, 

140 body=body, 

141 fields=fields, 

142 headers=headers, 

143 preload_content=preload_content, 

144 decode_content=decode_content, 

145 redirect=redirect, 

146 retries=retries, 

147 timeout=timeout, 

148 json=json, 

149 )