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])
python类AttributeError()的实例源码
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
