我们从Python开源项目中，提取了以下36个代码示例，用于说明如何使用gevent.socket.AF_INET。
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 gethostbyname(self, hostname, family=AF_INET): hostname = _resolve_special(hostname, family) return self.gethostbyname_ex(hostname, family)[-1][0]
def _getnameinfo(self, sockaddr, flags): if not isinstance(flags, int): raise TypeError('an integer is required') if not isinstance(sockaddr, tuple): raise TypeError('getnameinfo() argument 1 must be a tuple') address = sockaddr[0] if isinstance(address, unicode): address = address.encode('ascii') if not isinstance(address, str): raise TypeError('sockaddr[0] must be a string, not %s' % type(address).__name__) port = sockaddr[1] if not isinstance(port, int): raise TypeError('port must be an integer, not %s' % type(port)) waiter = Waiter(self.hub) result = self._getaddrinfo(address, str(sockaddr[1]), family=AF_UNSPEC, socktype=SOCK_DGRAM) if not result: raise elif len(result) != 1: raise error('sockaddr resolved to multiple addresses') family, socktype, proto, name, address = result[0] if family == AF_INET: if len(sockaddr) != 2: raise error("IPv4 sockaddr must be 2 tuple") elif family == AF_INET6: address = address[:2] + sockaddr[2:] self.ares.getnameinfo(waiter, address, flags) node, service = waiter.get() if service is None: service = '0' return node, service
def _getnameinfo(self, sockaddr, flags): if not isinstance(flags, int): raise TypeError('an integer is required') if not isinstance(sockaddr, tuple): raise TypeError('getnameinfo() argument 1 must be a tuple') address = sockaddr[0] if not PY3 and isinstance(address, text_type): address = address.encode('ascii') if not isinstance(address, string_types): raise TypeError('sockaddr[0] must be a string, not %s' % type(address).__name__) port = sockaddr[1] if not isinstance(port, int): raise TypeError('port must be an integer, not %s' % type(port)) waiter = Waiter(self.hub) result = self._getaddrinfo(address, str(sockaddr[1]), family=AF_UNSPEC, socktype=SOCK_DGRAM) if not result: reraise(*sys.exc_info()) elif len(result) != 1: raise error('sockaddr resolved to multiple addresses') family, socktype, proto, name, address = result[0] if family == AF_INET: if len(sockaddr) != 2: raise error("IPv4 sockaddr must be 2 tuple") elif family == AF_INET6: address = address[:2] + sockaddr[2:] self.ares.getnameinfo(waiter, address, flags) node, service = waiter.get() if service is None: service = '0' return node, service
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 get_raw_socket(cls, env): #pragma: no cover sock = None if env.get('uwsgi.version'): try: import uwsgi fd = uwsgi.connection_fd() sock = socket.fromfd(fd, socket.AF_INET, socket.SOCK_STREAM) except Exception as e: pass elif env.get('gunicorn.socket'): sock = env['gunicorn.socket'] if not sock: # attempt to find socket from wsgi.input input_ = env.get('wsgi.input') if input_: if hasattr(input_, '_sock'): raw = input_._sock sock = socket.socket(_sock=raw) elif hasattr(input_, 'raw'): sock = input_.raw._sock elif hasattr(input_, 'rfile'): # PY3 if hasattr(input_.rfile, 'raw'): sock = input_.rfile.raw._sock # PY2 else: sock = input_.rfile._sock return sock # ============================================================================
def _getnameinfo(self, sockaddr, flags): if not isinstance(flags, int): raise TypeError('an integer is required') if not isinstance(sockaddr, tuple): raise TypeError('getnameinfo() argument 1 must be a tuple') address = sockaddr[0] if not PY3 and isinstance(address, text_type): address = address.encode('ascii') if not isinstance(address, string_types): raise TypeError('sockaddr[0] must be a string, not %s' % type(address).__name__) port = sockaddr[1] if not isinstance(port, int): raise TypeError('port must be an integer, not %s' % type(port)) waiter = Waiter(self.hub) result = self._getaddrinfo(address, str(sockaddr[1]), family=AF_UNSPEC, socktype=SOCK_DGRAM) if not result: reraise(*sys.exc_info()) elif len(result) != 1: raise error('sockaddr resolved to multiple addresses') family, _socktype, _proto, _name, address = result[0] if family == AF_INET: if len(sockaddr) != 2: raise error("IPv4 sockaddr must be 2 tuple") elif family == AF_INET6: address = address[:2] + sockaddr[2:] self.ares.getnameinfo(waiter, address, flags) node, service = waiter.get() if service is None: if PY3: # ares docs: "If the query did not complete # successfully, or one of the values was not # requested, node or service will be NULL ". Python 2 # allows that for the service, but Python 3 raises # an error. This is tested by test_socket in py 3.4 err = gaierror('nodename nor servname provided, or not known') err.errno = 8 raise err service = '0' return node, service
