(
self, host, port, timeout=None, local_address=None, socket_options=None
)
| 82 | self._family = family |
| 83 | |
| 84 | def connect_tcp( |
| 85 | self, host, port, timeout=None, local_address=None, socket_options=None |
| 86 | ): # pylint: disable=signature-differs |
| 87 | addresses = [] |
| 88 | _, expiration = _compute_times(timeout) |
| 89 | if dns.inet.is_address(host): |
| 90 | addresses.append(host) |
| 91 | elif self._bootstrap_address is not None: |
| 92 | addresses.append(self._bootstrap_address) |
| 93 | else: |
| 94 | timeout = _remaining(expiration) |
| 95 | family = self._family |
| 96 | if local_address: |
| 97 | family = dns.inet.af_for_address(local_address) |
| 98 | answers = self._resolver.resolve_name( |
| 99 | host, family=family, lifetime=timeout |
| 100 | ) |
| 101 | addresses = answers.addresses() |
| 102 | for address in addresses: |
| 103 | af = dns.inet.af_for_address(address) |
| 104 | if local_address is not None or self._local_port != 0: |
| 105 | if local_address is None: |
| 106 | local_address = "0.0.0.0" |
| 107 | source = dns.inet.low_level_address_tuple( |
| 108 | (local_address, self._local_port), af |
| 109 | ) |
| 110 | else: |
| 111 | source = None |
| 112 | try: |
| 113 | sock = make_socket(af, socket.SOCK_STREAM, source) |
| 114 | attempt_expiration = _expiration_for_this_attempt(2.0, expiration) |
| 115 | _connect( |
| 116 | sock, |
| 117 | dns.inet.low_level_address_tuple((address, port), af), |
| 118 | attempt_expiration, |
| 119 | ) |
| 120 | return _CoreSyncStream(sock) |
| 121 | except Exception: |
| 122 | pass |
| 123 | raise httpcore.ConnectError |
| 124 | |
| 125 | def connect_unix_socket( |
| 126 | self, path, timeout=None, socket_options=None |
nothing calls this directly
no test coverage detected