我们从Python开源项目中，提取了以下50个代码示例，用于说明如何使用_socket.gaierror()。
def gethostbyname_ex(self, hostname, family=AF_INET): if isinstance(hostname, unicode): hostname = hostname.encode('ascii') elif not isinstance(hostname, str): raise TypeError('Expected string, not %s' % type(hostname).__name__) while True: ares = self.ares try: waiter = Waiter(self.hub) ares.gethostbyname(waiter, hostname, family) result = waiter.get() if not result[-1]: raise gaierror(-5, 'No address associated with hostname') return result except gaierror: if ares is self.ares: raise # "self.ares is not ares" means channel was destroyed (because we were forked)
def gethostbyname_ex(self, hostname, family=AF_INET): if PY3: if isinstance(hostname, str): hostname = hostname.encode('idna') elif not isinstance(hostname, (bytes, bytearray)): raise TypeError('Expected es(idna), not %s' % type(hostname).__name__) else: if isinstance(hostname, text_type): hostname = hostname.encode('ascii') elif not isinstance(hostname, str): raise TypeError('Expected string, not %s' % type(hostname).__name__) while True: ares = self.ares try: waiter = Waiter(self.hub) ares.gethostbyname(waiter, hostname, family) result = waiter.get() if not result[-1]: raise gaierror(-5, 'No address associated with hostname') return result except gaierror: if ares is self.ares: raise # "self.ares is not ares" means channel was destroyed (because we were forked)
def getaddrinfo(self, host, port, family=0, socktype=0, proto=0, flags=0): while True: ares = self.ares try: return self._getaddrinfo(host, port, family, socktype, proto, flags) except gaierror: if ares is self.ares: raise
def gethostbyaddr(self, ip_address): ip_address = _resolve_special(ip_address, AF_UNSPEC) while True: ares = self.ares try: return self._gethostbyaddr(ip_address) except gaierror: if ares is self.ares: raise
def getnameinfo(self, sockaddr, flags): while True: ares = self.ares try: return self._getnameinfo(sockaddr, flags) except gaierror: if ares is self.ares: raise
def gethostbyname_ex(self, hostname, family=AF_INET): if PY3: if isinstance(hostname, str): hostname = hostname.encode('idna') elif not isinstance(hostname, (bytes, bytearray)): raise TypeError('Expected es(idna), not %s' % type(hostname).__name__) else: if isinstance(hostname, text_type): hostname = hostname.encode('ascii') elif not isinstance(hostname, str): raise TypeError('Expected string, not %s' % type(hostname).__name__) while True: ares = self.ares try: waiter = Waiter(self.hub) ares.gethostbyname(waiter, hostname, family) result = waiter.get() if not result[-1]: raise gaierror(-5, 'No address associated with hostname') return result except gaierror: if ares is self.ares: if hostname == b'255.255.255.255': # The stdlib handles this case in 2.7 and 3.x, but ares does not. # It is tested by test_socket.py in 3.4. # HACK: So hardcode the expected return. return ('255.255.255.255', [], ['255.255.255.255']) raise # "self.ares is not ares" means channel was destroyed (because we were forked)
def __init__(self, hub=None): if hub is None: hub = get_hub() self.pool = hub.threadpool if _socket.gaierror not in hub.NOT_ERROR: # Do not cause lookup failures to get printed by the default # error handler. This can be very noisy. hub.NOT_ERROR += (_socket.gaierror, _socket.herror)
def test_listen_with_bad_port_type(self): TornadoWrapper.start_app() with self.assertRaisesRegexp(gaierror, 'Servname not supported'): TornadoWrapper.listen('not a integer')
def retrieve_data(self, session): """ Retrieve information about entity using an existing session. :param session: Requests session to use for data retrieval. :return: True if data about entity has been successfully retrieved and no filter callback excluded this entity, False otherwise. """ try: logger.info("Retrieving data for entity " + str(self) + "...") # execute pre_request_callbacks for callback in self.configuration.pre_request_callbacks: callback(self) # reduce request frequency as configured delay = randint(self.configuration.delay_min, self.configuration.delay_max) # delay between requests in milliseconds time.sleep(delay / 1000) # sleep for delay ms to prevent getting blocked # retrieve data if len(self.configuration.headers) > 0: response = session.get(self.uri, headers=self.configuration.headers) else: response = session.get(self.uri) if response.ok: logger.info("Successfully retrieved data for entity " + str(self) + ".") if self.configuration.raw_download: # raw download self.output_parameters[self.configuration.raw_parameter] = response.content else: # JSON API call # deserialize JSON string json_response = json.loads(response.text) # extract parameters according to parameter mapping self._extract_output_parameters(json_response) # execute post_request_callbacks for callback in self.configuration.post_request_callbacks: result = callback(self) # check if callback implements filter if isinstance(result, bool): if not result: logger.info("Entity removed because of filter callback " + str(callback) + ": " + str(self)) return False return True else: logger.error("Error " + str(response.status_code) + ": Could not retrieve data for entity " + str(self) + ". Response: " + str(response.content)) return False except (gaierror, ConnectionError, MaxRetryError, NewConnectionError): logger.error("An error occurred while retrieving data for entity " + str(self) + ".")
