Python urllib3.exceptions 模块，NewConnectionError() 实例源码

def get_resource(self, resource, namespace="all"):
        ret, resources = None, list()
        try:
            ret, namespaced_resource = self._call_api_client(resource)
        except ApiException as ae:
            self.logger.warning("resource autocomplete disabled, encountered "
                                "ApiException", exc_info=1)
        except (NewConnectionError, MaxRetryError, ConnectTimeoutError):
            self.logger.warning("unable to connect to k8 cluster", exc_info=1)
        if ret:
            for i in ret.items:
                if namespace == "all" or not namespaced_resource:
                    resources.append((i.metadata.name, i.metadata.namespace))
                elif namespace == i.metadata.namespace:
                    resources.append((i.metadata.name, i.metadata.namespace))
        return resources

def get_info(self):
        if not self.is_checked:
            try:
                r = requests.get(str(self), timeout=(10, 3))
                if r.status_code == 200:
                    self.server = r.headers['Server']
                elif r.status_code >= 400:
                    raise TargetNotExistException(self.url)
            except requests.exceptions.ReadTimeout as rt:
                logger.exception(rt)

            try:
                url = re.compile(r"https?://(www\.)?")
                self.url_c = url.sub('', self.url).strip().strip('/')
                self.ip = socket.gethostbyname(self.url_c)
            except socket.gaierror as err:
                logger.exception(err)
            except NewConnectionError:
                raise TargetNotExistException(self.url)

            self.is_checked = True

def get_short_url(url, nickname, ui_locales) -> dict:
    """
    Shortens the url via external service.

    :param url: Url as string, which should be shortened
    :param nickname: current users nickname
    :param ui_locales: language of the discussion
    :rtype: dict
    :return: dictionary with the url, services name and the url of the service or an error
    """
    user.update_last_action(nickname)

    try:
        service = Shorteners.TINYURL
        service_url = 'http://tinyurl.com/'
        shortener = Shortener(service)
        short_url = format(shortener.short(url))
    except (ReadTimeout, ConnectionError, NewConnectionError) as e:
        logger('getter', 'get_short_url', repr(e), error=True)
        _tn = Translator(ui_locales)
        prepared_dict = {'error': _tn.get(_.serviceNotAvailable)}
        return prepared_dict

    prepared_dict = dict()
    prepared_dict['url'] = short_url
    prepared_dict['service'] = service
    prepared_dict['service_url'] = service_url
    prepared_dict['error'] = ''

    return prepared_dict

def _new_conn(self):
        """
        Establish a new connection via the SOCKS proxy.
        """
        extra_kw = {}
        if self.source_address:
            extra_kw['source_address'] = self.source_address

        if self.socket_options:
            extra_kw['socket_options'] = self.socket_options

        try:
            conn = socks.create_connection(
                (self.host, self.port),
                proxy_type=self._socks_options['socks_version'],
                proxy_addr=self._socks_options['proxy_host'],
                proxy_port=self._socks_options['proxy_port'],
                proxy_username=self._socks_options['username'],
                proxy_password=self._socks_options['password'],
                timeout=self.timeout,
                **extra_kw
            )

        except SocketTimeout as e:
            raise ConnectTimeoutError(
                self, "Connection to %s timed out. (connect timeout=%s)" %
                (self.host, self.timeout))

        except socks.ProxyError as e:
            # This is fragile as hell, but it seems to be the only way to raise
            # useful errors here.
            if e.socket_err:
                error = e.socket_err
                if isinstance(error, SocketTimeout):
                    raise ConnectTimeoutError(
                        self,
                        "Connection to %s timed out. (connect timeout=%s)" %
                        (self.host, self.timeout)
                    )
                else:
                    raise NewConnectionError(
                        self,
                        "Failed to establish a new connection: %s" % error
                    )
            else:
                raise NewConnectionError(
                    self,
                    "Failed to establish a new connection: %s" % e
                )

        except SocketError as e:  # Defensive: PySocks should catch all these.
            raise NewConnectionError(
                self, "Failed to establish a new connection: %s" % e)

        return conn


# We don't need to duplicate the Verified/Unverified distinction from
# urllib3/connection.py here because the HTTPSConnection will already have been
# correctly set to either the Verified or Unverified form by that module. This
# means the SOCKSHTTPSConnection will automatically be the correct type.