我们从Python开源项目中，提取了以下14个代码示例，用于说明如何使用twisted.web.client._parse()。
def testParse(self): scheme, host, port, path = client._parse("http://127.0.0.1/") self.assertEquals(path, "/") self.assertEquals(port, 80) scheme, host, port, path = client._parse("https://127.0.0.1/") self.assertEquals(path, "/") self.assertEquals(port, 443) scheme, host, port, path = client._parse("http://spam:12345/") self.assertEquals(port, 12345) scheme, host, port, path = client._parse("http://foo ") self.assertEquals(host, "foo") self.assertEquals(path, "/") scheme, host, port, path = client._parse("http://egg:7890") self.assertEquals(port, 7890) self.assertEquals(host, "egg") self.assertEquals(path, "/")
def __init__(self, url, outputfile, contextFactory=None, *args, **kwargs): if hasattr(client, '_parse'): scheme, host, port, path = client._parse(url) else: try: from twisted.web.client import _URI as URI except ImportError: from twisted.web.client import URI uri = URI.fromBytes(url) scheme = uri.scheme host = uri.host port = uri.port or (443 if scheme == 'https' else 80) path = uri.path self.factory = HTTPProgressDownloader(url, outputfile, *args, **kwargs) if scheme == "https": from twisted.internet import ssl if contextFactory is None: contextFactory = ssl.ClientContextFactory() self.connection = reactor.connectSSL(host, port, self.factory, contextFactory) else: self.connection = reactor.connectTCP(host, port, self.factory)
def __init__(self, url, outputfile, contextFactory=None, *args, **kwargs): if hasattr(client, '_parse'): scheme, host, port, path = client._parse(url) else: try: from twisted.web.client import _URI as URI except ImportError: from twisted.web.client import URI uri = URI.fromBytes(url) scheme = uri.scheme host = uri.host port = uri.port path = uri.path self.factory = HTTPProgressDownloader(url, outputfile, *args, **kwargs) if scheme == 'https': from twisted.internet import ssl if contextFactory is None: contextFactory = ssl.ClientContextFactory() self.connection = reactor.connectSSL(host, port, self.factory, contextFactory) else: self.connection = reactor.connectTCP(host, port, self.factory)
def getPagePrxoy(url, proxy=None, contextFactory=None, *args, **kwargs): ''' proxy= { host:192.168.1.111, port:6666 } ''' kwargs["timeout"] = 60 if proxy is None: scheme, host, port, path = client._parse(url) factory = client.HTTPClientFactory(url, *args, **kwargs) if scheme == b'https': from twisted.internet import ssl if contextFactory is None: contextFactory = ssl.ClientContextFactory() reactor.connectSSL(client.nativeString(host), port, factory, contextFactory) else: reactor.connectTCP(client.nativeString(host), port, factory) return factory.deferred else: factory = client.HTTPClientFactory(url, *args, **kwargs) reactor.connectTCP(proxy["host"], proxy["port"], factory) return factory.deferred
def testFactoryInfo(self): url = self.getURL('file') scheme, host, port, path = client._parse(url) factory = client.HTTPClientFactory(url) reactor.connectTCP(host, port, factory) return factory.deferred.addCallback(self._cbFactoryInfo, factory)
def testFactoryInfo(self): url = self.getURL('file') scheme, host, port, path = client._parse(url) factory = client.HTTPClientFactory(url) reactor.connectSSL(host, port, factory, ssl.ClientContextFactory()) # The base class defines _cbFactoryInfo correctly for this return factory.deferred.addCallback(self._cbFactoryInfo, factory)
def getPage(url, contextFactory=None, *args, **kwargs): log.msg('Method: %s' % kwargs.get('method', 'GET')) log.msg('URI: %s' % url) try: log.msg('Headers: %r' % kwargs['headers']) except KeyError: pass try: log.msg('Payload: %r' % kwargs['postdata']) except KeyError: pass scheme, host, port, path = client._parse(url) factory = HTTPClientFactory(url, *args, **kwargs) if scheme == 'https': from twisted.internet import ssl if contextFactory is None: contextFactory = ssl.ClientContextFactory() reactor.connectSSL(host, port, factory, contextFactory) else: reactor.connectTCP(host, port, factory) def _eb(failure): log.msg('Failed.') log.msg(failure) return failure return factory.deferred.addCallback(_checkCacheControl).addErrback(_eb)
def getPage(url, contextFactory=None, *args, **kwargs): scheme, host, port, path = _parse(url) factory = LimitedHTTPClientFactory(url, *args, **kwargs) if scheme == 'https': from twisted.internet import ssl if contextFactory is None: contextFactory = ssl.ClientContextFactory() reactor.connectSSL(host, port, factory, contextFactory) else: reactor.connectTCP(host, port, factory) return factory.deferred
def request(self, method, uri, headers=None, bodyProducer=None): """ Issue a new request. @param method: The request method to send. @type method: C{str} @param uri: The request URI send. @type uri: C{str} @param scheme: A string like C{'http'} or C{'https'} (the only two supported values) to use to determine how to establish the connection. @param host: A C{str} giving the hostname which will be connected to in order to issue a request. @param port: An C{int} giving the port number the connection will be on. @param path: A C{str} giving the path portion of the request URL. @param headers: The request headers to send. If no I{Host} header is included, one will be added based on the request URI. @type headers: L{Headers} @param bodyProducer: An object which will produce the request body or, if the request body is to be empty, L{None}. @type bodyProducer: L{IBodyProducer} provider @return: A L{Deferred} which fires with the result of the request (a L{Response} instance), or fails if there is a problem setting up a connection over which to issue the request. It may also fail with L{SchemeNotSupported} if the scheme of the given URI is not supported. @rtype: L{Deferred} """ scheme, host, port, path = _parse(uri) if headers is None: headers = Headers() if not headers.hasHeader('host'): # This is a lot of copying. It might be nice if there were a bit # less. headers = Headers(dict(headers.getAllRawHeaders())) headers.addRawHeader( 'host', self._computeHostValue(scheme, host, port)) if self.persistent: sem = self._semaphores.get((scheme, host, port)) if sem is None: sem = DeferredSemaphore(self.maxConnectionsPerHostName) self._semaphores[scheme, host, port] = sem return sem.run(self._request, method, scheme, host, port, path, headers, bodyProducer) else: return self._request( method, scheme, host, port, path, headers, bodyProducer)
