我们从Python开源项目中，提取了以下50个代码示例，用于说明如何使用socket._GLOBAL_DEFAULT_TIMEOUT。
def __init__(self, host=None, port=0, timeout=socket._GLOBAL_DEFAULT_TIMEOUT): """Constructor. When called without arguments, create an unconnected instance. With a hostname argument, it connects the instance; port number and timeout are optional. """ self.debuglevel = DEBUGLEVEL self.host = host self.port = port self.timeout = timeout self.sock = None self.rawq = '' self.irawq = 0 self.cookedq = '' self.eof = 0 self.iacseq = '' # Buffer for IAC sequence. self.sb = 0 # flag for SB and SE sequence. self.sbdataq = '' self.option_callback = None if host is not None: self.open(host, port, timeout)
def __init__(self, host, port=None, strict=None, timeout=socket._GLOBAL_DEFAULT_TIMEOUT, source_address=None): self.timeout = timeout self.source_address = source_address self.sock = None self._buffer = [] self.__response = None self.__state = _CS_IDLE self._method = None self._tunnel_host = None self._tunnel_port = None self._tunnel_headers = {} if strict is not None: self.strict = strict (self.host, self.port) = self._get_hostport(host, port) # This is stored as an instance variable to allow unittests # to replace with a suitable mock self._create_connection = socket.create_connection
def urlopen(url, data=None, timeout=socket._GLOBAL_DEFAULT_TIMEOUT, cafile=None, capath=None, cadefault=False, context=None): global _opener if cafile or capath or cadefault: if context is not None: raise ValueError( "You can't pass both context and any of cafile, capath, and " "cadefault" ) if not _have_ssl: raise ValueError('SSL support not available') context = ssl.create_default_context(purpose=ssl.Purpose.SERVER_AUTH, cafile=cafile, capath=capath) https_handler = HTTPSHandler(context=context) opener = build_opener(https_handler) elif context: https_handler = HTTPSHandler(context=context) opener = build_opener(https_handler) elif _opener is None: _opener = opener = build_opener() else: opener = _opener return opener.open(url, data, timeout)
def __init__(self, host=None, port=0, timeout=socket._GLOBAL_DEFAULT_TIMEOUT): """Constructor. When called without arguments, create an unconnected instance. With a hostname argument, it connects the instance; port number and timeout are optional. """ self.debuglevel = DEBUGLEVEL self.host = host self.port = port self.timeout = timeout self.sock = None self.rawq = '' self.irawq = 0 self.cookedq = '' self.eof = 0 self.iacseq = '' # Buffer for IAC sequence. self.sb = 0 # flag for SB and SE sequence. self.sbdataq = '' self.option_callback = None self._has_poll = hasattr(select, 'poll') if host is not None: self.open(host, port, timeout)
def test_socket_timeout(): """Test socket timeout validator.""" TEST_CONF_TIMEOUT = 'timeout' schema = vol.Schema( {vol.Required(TEST_CONF_TIMEOUT, default=None): cv.socket_timeout}) with pytest.raises(vol.Invalid): schema({TEST_CONF_TIMEOUT: 0.0}) with pytest.raises(vol.Invalid): schema({TEST_CONF_TIMEOUT: -1}) assert _GLOBAL_DEFAULT_TIMEOUT == schema({TEST_CONF_TIMEOUT: None})[TEST_CONF_TIMEOUT] assert 1.0 == schema({TEST_CONF_TIMEOUT: 1})[TEST_CONF_TIMEOUT]
def __init__(self, host='', user='', passwd='', acct='', keyfile=None, certfile=None, context=None, timeout=_GLOBAL_DEFAULT_TIMEOUT, source_address=None): if context is not None and keyfile is not None: raise ValueError("context and keyfile arguments are mutually " "exclusive") if context is not None and certfile is not None: raise ValueError("context and certfile arguments are mutually " "exclusive") self.keyfile = keyfile self.certfile = certfile if context is None: context = ssl._create_stdlib_context(self.ssl_version, certfile=certfile, keyfile=keyfile) self.context = context self._prot_p = False FTP.__init__(self, host, user, passwd, acct, timeout)
def __init__(self, host, port=None, key_file=None, cert_file=None, strict=None, timeout=socket._GLOBAL_DEFAULT_TIMEOUT, **kw): HTTPConnection.__init__(self, host, port, strict=strict, timeout=timeout, **kw) self.key_file = key_file self.cert_file = cert_file # Required property for Google AppEngine 1.9.0 which otherwise causes # HTTPS requests to go out as HTTP. (See Issue #356) self._protocol = 'https'
def clone(self): """ Create a copy of the timeout object Timeout properties are stored per-pool but each request needs a fresh Timeout object to ensure each one has its own start/stop configured. :return: a copy of the timeout object :rtype: :class:`Timeout` """ # We can't use copy.deepcopy because that will also create a new object # for _GLOBAL_DEFAULT_TIMEOUT, which socket.py uses as a sentinel to # detect the user default. return Timeout(connect=self._connect, read=self._read, total=self.total)
def __init__(self, host, port=None, key_file=None, cert_file=None, strict=None, timeout=socket._GLOBAL_DEFAULT_TIMEOUT, ssl_context=None, **kw): HTTPConnection.__init__(self, host, port, strict=strict, timeout=timeout, **kw) self.key_file = key_file self.cert_file = cert_file self.ssl_context = ssl_context # Required property for Google AppEngine 1.9.0 which otherwise causes # HTTPS requests to go out as HTTP. (See Issue #356) self._protocol = 'https'
def __init__(self, host, port=None, strict=None, timeout=socket._GLOBAL_DEFAULT_TIMEOUT, source_address=None): self.timeout = timeout self.source_address = source_address self.sock = None self._buffer = [] self.__response = None self.__state = _CS_IDLE self._method = None self._tunnel_host = None self._tunnel_port = None self._tunnel_headers = {} self._set_hostport(host, port) if strict is not None: self.strict = strict
def __init__(self, host, port=None, key_file=None, cert_file=None, strict=None, timeout=socket._GLOBAL_DEFAULT_TIMEOUT, source_address=None): HTTPConnection.__init__(self, host, port, strict, timeout, source_address) self.key_file = key_file self.cert_file = cert_file
def __init__(self, host='', port=0, local_hostname=None, timeout=socket._GLOBAL_DEFAULT_TIMEOUT): """Initialize a new instance. If specified, `host' is the name of the remote host to which to connect. If specified, `port' specifies the port to which to connect. By default, smtplib.SMTP_PORT is used. An SMTPConnectError is raised if the specified `host' doesn't respond correctly. If specified, `local_hostname` is used as the FQDN of the local host. By default, the local hostname is found using socket.getfqdn(). """ self.timeout = timeout self.esmtp_features = {} if host: (code, msg) = self.connect(host, port) if code != 220: raise SMTPConnectError(code, msg) if local_hostname is not None: self.local_hostname = local_hostname else: # RFC 2821 says we should use the fqdn in the EHLO/HELO verb, and # if that can't be calculated, that we should use a domain literal # instead (essentially an encoded IP address like [A.B.C.D]). fqdn = socket.getfqdn() if '.' in fqdn: self.local_hostname = fqdn else: # We can't find an fqdn hostname, so use a domain literal addr = '127.0.0.1' try: addr = socket.gethostbyname(socket.gethostname()) except socket.gaierror: pass self.local_hostname = '[%s]' % addr
def __init__(self, host='', port=0, local_hostname=None, keyfile=None, certfile=None, timeout=socket._GLOBAL_DEFAULT_TIMEOUT): self.keyfile = keyfile self.certfile = certfile SMTP.__init__(self, host, port, local_hostname, timeout)
def __init__(self, host='', user='', passwd='', acct='', timeout=_GLOBAL_DEFAULT_TIMEOUT): self.timeout = timeout if host: self.connect(host) if user: self.login(user, passwd, acct)
def __init__(self, host='', user='', passwd='', acct='', keyfile=None, certfile=None, timeout=_GLOBAL_DEFAULT_TIMEOUT): self.keyfile = keyfile self.certfile = certfile self._prot_p = False FTP.__init__(self, host, user, passwd, acct, timeout)
def urlopen(url, data=None, timeout=socket._GLOBAL_DEFAULT_TIMEOUT): global _opener if _opener is None: _opener = build_opener() return _opener.open(url, data, timeout)
def open(self, fullurl, data=None, timeout=socket._GLOBAL_DEFAULT_TIMEOUT): # accept a URL or a Request object if isinstance(fullurl, basestring): req = Request(fullurl, data) else: req = fullurl if data is not None: req.add_data(data) req.timeout = timeout protocol = req.get_type() # pre-process request meth_name = protocol+"_request" for processor in self.process_request.get(protocol, []): meth = getattr(processor, meth_name) req = meth(req) response = self._open(req, data) # post-process response meth_name = protocol+"_response" for processor in self.process_response.get(protocol, []): meth = getattr(processor, meth_name) response = meth(req, response) return response
def __init__(self, user, passwd, host, port, dirs, timeout=socket._GLOBAL_DEFAULT_TIMEOUT): self.user = user self.passwd = passwd self.host = host self.port = port self.dirs = dirs self.timeout = timeout self.init()
def __init__(self, host, port=POP3_PORT, timeout=socket._GLOBAL_DEFAULT_TIMEOUT): self.host = host self.port = port self.sock = socket.create_connection((host, port), timeout) self.file = self.sock.makefile('rb') self._debugging = 0 self.welcome = self._getresp()
def has_timeout(timeout): # python 2.6 if hasattr(socket, '_GLOBAL_DEFAULT_TIMEOUT'): return (timeout is not None and timeout is not socket._GLOBAL_DEFAULT_TIMEOUT) return (timeout is not None)
def __init__(self, host, port=None, key_file=None, cert_file=None, strict=None, timeout=socket._GLOBAL_DEFAULT_TIMEOUT, source_address=None): try: HTTPConnection.__init__(self, host, port, strict, timeout, source_address) except TypeError: # Python 2.6 HTTPConnection.__init__(self, host, port, strict, timeout) self.key_file = key_file self.cert_file = cert_file
def __init__(self, host, port=None, strict=None, timeout=socket._GLOBAL_DEFAULT_TIMEOUT, ssl_context=None): HTTPConnection.__init__(self, host, port, strict, timeout) if not hasattr(self, 'ssl_context'): self.ssl_context = None if ssl_context is not None: if not isinstance(ssl_context, SSL.Context): raise TypeError('Expecting OpenSSL.SSL.Context type for "' 'ssl_context" keyword; got %r instead' % ssl_context) self.ssl_context = ssl_context
