Mercurial > repos > jpayne > bioproject_to_srr_2

comparison urllib3/util/connection.py @ 7:5eb2d5e3bf22

Find changesets by keywords (author, files, the commit message), revision number or hash, or revset expression.
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 socket
4 import typing
5
6 from ..exceptions import LocationParseError
7 from .timeout import _DEFAULT_TIMEOUT, _TYPE_TIMEOUT
8
9 _TYPE_SOCKET_OPTIONS = typing.Sequence[typing.Tuple[int, int, typing.Union[int, bytes]]]
10
11 if typing.TYPE_CHECKING:
12 from .._base_connection import BaseHTTPConnection
13
14
15 def is_connection_dropped(conn: BaseHTTPConnection) -> bool: # Platform-specific
16 """
17 Returns True if the connection is dropped and should be closed.
18 :param conn: :class:`urllib3.connection.HTTPConnection` object.
19 """
20 return not conn.is_connected
21
22
23 # This function is copied from socket.py in the Python 2.7 standard
24 # library test suite. Added to its signature is only `socket_options`.
25 # One additional modification is that we avoid binding to IPv6 servers
26 # discovered in DNS if the system doesn't have IPv6 functionality.
27 def create_connection(
28 address: tuple[str, int],
29 timeout: _TYPE_TIMEOUT = _DEFAULT_TIMEOUT,
30 source_address: tuple[str, int] | None = None,
31 socket_options: _TYPE_SOCKET_OPTIONS | None = None,
32 ) -> socket.socket:
33 """Connect to *address* and return the socket object.
34
35 Convenience function. Connect to *address* (a 2-tuple ``(host,
36 port)``) and return the socket object. Passing the optional
37 *timeout* parameter will set the timeout on the socket instance
38 before attempting to connect. If no *timeout* is supplied, the
39 global default timeout setting returned by :func:`socket.getdefaulttimeout`
40 is used. If *source_address* is set it must be a tuple of (host, port)
41 for the socket to bind as a source address before making the connection.
42 An host of '' or port 0 tells the OS to use the default.
43 """
44
45 host, port = address
46 if host.startswith("["):
47 host = host.strip("[]")
48 err = None
49
50 # Using the value from allowed_gai_family() in the context of getaddrinfo lets
51 # us select whether to work with IPv4 DNS records, IPv6 records, or both.
52 # The original create_connection function always returns all records.
53 family = allowed_gai_family()
54
55 try:
56 host.encode("idna")
57 except UnicodeError:
58 raise LocationParseError(f"'{host}', label empty or too long") from None
59
60 for res in socket.getaddrinfo(host, port, family, socket.SOCK_STREAM):
61 af, socktype, proto, canonname, sa = res
62 sock = None
63 try:
64 sock = socket.socket(af, socktype, proto)
65
66 # If provided, set socket level options before connecting.
67 _set_socket_options(sock, socket_options)
68
69 if timeout is not _DEFAULT_TIMEOUT:
70 sock.settimeout(timeout)
71 if source_address:
72 sock.bind(source_address)
73 sock.connect(sa)
74 # Break explicitly a reference cycle
75 err = None
76 return sock
77
78 except OSError as _:
79 err = _
80 if sock is not None:
81 sock.close()
82
83 if err is not None:
84 try:
85 raise err
86 finally:
87 # Break explicitly a reference cycle
88 err = None
89 else:
90 raise OSError("getaddrinfo returns an empty list")
91
92
93 def _set_socket_options(
94 sock: socket.socket, options: _TYPE_SOCKET_OPTIONS | None
95 ) -> None:
96 if options is None:
97 return
98
99 for opt in options:
100 sock.setsockopt(*opt)
101
102
103 def allowed_gai_family() -> socket.AddressFamily:
104 """This function is designed to work in the context of
105 getaddrinfo, where family=socket.AF_UNSPEC is the default and
106 will perform a DNS search for both IPv6 and IPv4 records."""
107
108 family = socket.AF_INET
109 if HAS_IPV6:
110 family = socket.AF_UNSPEC
111 return family
112
113
114 def _has_ipv6(host: str) -> bool:
115 """Returns True if the system can bind an IPv6 address."""
116 sock = None
117 has_ipv6 = False
118
119 if socket.has_ipv6:
120 # has_ipv6 returns true if cPython was compiled with IPv6 support.
121 # It does not tell us if the system has IPv6 support enabled. To
122 # determine that we must bind to an IPv6 address.
123 # https://github.com/urllib3/urllib3/pull/611
124 # https://bugs.python.org/issue658327
125 try:
126 sock = socket.socket(socket.AF_INET6)
127 sock.bind((host, 0))
128 has_ipv6 = True
129 except Exception:
130 pass
131
132 if sock:
133 sock.close()
134 return has_ipv6
135
136
137 HAS_IPV6 = _has_ipv6("::1")