Python exceptions 模块，AttributeError() 实例源码

def isDaemonAlive(hostAndPort="{0}:{1}".format(IPFSAPI_IP, IPFSAPI_PORT)):
    """Ensure that the IPFS daemon is running via HTTP before proceeding"""
    client = ipfsapi.Client(IPFSAPI_IP, IPFSAPI_PORT)

    try:
        # OSError if ipfs not installed, redundant of below
        # subprocess.call(['ipfs', '--version'], stdout=open(devnull, 'wb'))

        # ConnectionError/AttributeError if IPFS daemon not running
        client.id()
        return True
    except (ConnectionError, exceptions.AttributeError):
        logError("Daemon is not running at http://" + hostAndPort)
        return False
    except OSError:
        logError("IPFS is likely not installed. "
                 "See https://ipfs.io/docs/install/")
        sys.exit()
    except:
        logError('Unknown error in retrieving daemon status')
        logError(sys.exc_info()[0])

def address_port(address):
    '''
    Return a string describing port portion of an address
    If there is no port portion, returns None
    '''
    # Looks like an IPv4Address or IPv6Address
    try:
        # ignore type, it's always TCP
        return "{}".format(address.port)
    except AttributeError:
        pass
    # Looks like a HostnameAddress
    # (we don't yet support hostnames in connect and listen)
    try:
        return "{}".format(address.port)
    except AttributeError:
        pass
    # We don't know any other way to get a port
    return None

def __getattr__(self, key):
        if key.startswith("_") or not key in self._meta.storage_attributes:
            raise AttributeError("Unknown attribute %s" % key)
        else:
            try:
                return self._storage_dict[key]
            except KeyError:
                attr = self._meta.storage_attributes[key]
                if attr.optional:
                    return None
                elif attr.default is not None:
                    if callable(attr.default):
                        return attr.default(self._storage_dict)
                    else:
                        return attr.default
                else:
                    log.error("Missing attribute %s, %s" % (key, self._storage_dict))
                    raise AttributeError("attribute %s not found" % key)

def make_symbol(ldlib, name, symbol, restype, argtypes):
    """ Helper for library symbols generation
    :param ldlib: loaded library reference
    :param name: function call to use
    :param symbol: library symbol to attach function to
    :param restype: library symbol return type
    :param argtypes: list of library symbol parameters
    :return: None
    """
    try:
        ldlib[name] = ldlib.lib[symbol]
        ldlib[name].restype = restype
        ldlib[name].argtypes = argtypes
    except AttributeError:
        print ldlib.name, name, "import(%d): Symbol not found" % sys.exc_info()[-1].tb_lineno
    except TypeError:
        pass
    except Exception as ex:
        print name, "import(%d):" % sys.exc_info()[-1].tb_lineno, ex, type(ex)

def transport_local_info(transport):
    '''
    Return a string describing the local endpoint connected to transport
    '''
    try:
        local = transport.getHost()
    except AttributeError:
        return None
    if local is None:
        return None
    return address_info(local)

def transport_local_hostname(transport):
    '''
    Return a string describing the hostname of the local endpoint connected to
    transport
    '''
    try:
        local = transport.getHost()
    except AttributeError:
        return None
    if local is None:
        return None
    return address_hostname(local)

def address_hostname(address):
    '''
    Return a string describing the host portion of an address.
    '''
    # Looks like an IPv4Address or IPv6Address
    try:
        # ignore type, it's always TCP
        return "{}".format(address.host)
    except AttributeError:
        pass
    # Looks like a HostnameAddress
    # (we don't yet support hostnames in connect and listen)
    try:
        return "{}".format(address.hostname)
    except AttributeError:
        pass
    # Looks like a UNIXAddress
    try:
        # Handle host for UNIXAddress, which is always None
        if address.name is None:
            return None
        else:
            return "{}".format(address.name)
    except AttributeError:
        pass
    if address is None:
        return None
    # Just ask it how it wants to be represented
    return str(address)

def request(self, url):
        """Request handler"""
        # if True == self.proxy:
        #    proxyserver = self.reader.get_random_proxy()
        #    try:
        #        conn = urllib3.proxy_from_url(proxyserver, )
        #    except urllib3.exceptions.ProxySchemeUnknown as e:
        #        log.critical(e.message + ": " + proxyserver)
        # else:
        #    conn = urllib3.connection_from_url(url, )


        try:
            response = self.http.urlopen(self.DEFAULT_HTTP_METHOD, url, headers=self.HEADER, redirect=False,
                                         timeout=self.rest, release_conn=True)
        except (urllib3.exceptions.ConnectTimeoutError,
                urllib3.exceptions.MaxRetryError,
                urllib3.exceptions.HostChangedError,
                urllib3.exceptions.ReadTimeoutError,
                urllib3.exceptions.ProxyError
                ) as e:
            response = None
            self.iterator = Progress.line(url + ' -> ' + e.message, self.urls.__len__(), 'warning', self.iterator)
        except exceptions.AttributeError as e:
            log.critical(e.message)
        except TypeError as e:
            log.critical(e.message)

        time.sleep(self.delay)
        return self.response(response, url)

def get(self, host, params=()):
        # type: (object, object) -> object
        """Get metadata by url"""

        self.__is_server_online(host)
        self.__disable_verbose()
        self.__parse_params(params)
        scheme, host = urlparse(host).scheme, urlparse(host).netloc
        self.DEFAULT_HTTP_PROTOCOL = scheme + "://"
        self.urls = self.__get_urls(host)
        response = {}
        self.HEADER['user-agent'] = self.reader.get_random_user_agent()
        log.info("user-agent : " + self.HEADER['user-agent'])
        log.info('Thread num : ' + str(self.threads))

        try:
            httplib.HTTPConnection.debuglevel = self.debug

            if hasattr(urllib3, 'disable_warnings'):
                urllib3.disable_warnings()
            if scheme == "http":
                self.http = urllib3.HTTPConnectionPool(host.split(':')[0],
                                                       port=80 if len(host.split(':')) == 1 else int(
                                                           host.split(':')[1]), block=True, maxsize=10)
            elif scheme == "https":
                self.http = urllib3.HTTPSConnectionPool(host.split(':')[0],
                                                        port=443 if len(host.split(':')) == 1 else int(
                                                            host.split(':')[1]), block=True, maxsize=10)
            else:
                log.critical("not support http protocl, Exit now ")
                sys.exit(1);
            pool = threadpool.ThreadPool(self.threads)
            requests = threadpool.makeRequests(self.request, self.urls)
            for req in requests:
                pool.putRequest(req)
            time.sleep(1)
            pool.wait()
        except exceptions.AttributeError as e:
            log.critical(e.message)
        except KeyboardInterrupt:
            log.warning('Session canceled')
            sys.exit()

        self.counter['total'] = self.urls.__len__()
        self.counter['pools'] = pool.workers.__len__()

        response['count'] = self.counter
        response['result'] = self.result

        return response