我们从Python开源项目中，提取了以下50个代码示例，用于说明如何使用tornado.httpclient._RequestProxy()。
def raw_fetch(self, headers, body): with closing(Resolver(io_loop=self.io_loop)) as resolver: with closing(SimpleAsyncHTTPClient(self.io_loop, resolver=resolver)) as client: conn = RawRequestHTTPConnection( self.io_loop, client, httpclient._RequestProxy( httpclient.HTTPRequest(self.get_url("/")), dict(httpclient.HTTPRequest._DEFAULTS)), None, self.stop, 1024 * 1024, resolver) conn.set_request( b"\r\n".join(headers + [utf8("Content-Length: %d\r\n" % len(body))]) + b"\r\n" + body) response = self.wait() response.rethrow() return response
def websocket_connect(url, io_loop=None, callback=None, connect_timeout=None): """Client-side websocket support. Takes a url and returns a Future whose result is a `WebSocketClientConnection`. .. versionchanged:: 3.2 Also accepts ``HTTPRequest`` objects in place of urls. """ if io_loop is None: io_loop = IOLoop.current() if isinstance(url, httpclient.HTTPRequest): assert connect_timeout is None request = url # Copy and convert the headers dict/object (see comments in # AsyncHTTPClient.fetch) request.headers = httputil.HTTPHeaders(request.headers) else: request = httpclient.HTTPRequest(url, connect_timeout=connect_timeout) request = httpclient._RequestProxy( request, httpclient.HTTPRequest._DEFAULTS) conn = WebSocketClientConnection(io_loop, request) if callback is not None: io_loop.add_future(conn.connect_future, callback) return conn.connect_future
def raw_fetch(self, headers, body): client = SimpleAsyncHTTPClient(self.io_loop) conn = RawRequestHTTPConnection( self.io_loop, client, httpclient._RequestProxy( httpclient.HTTPRequest(self.get_url("/")), dict(httpclient.HTTPRequest._DEFAULTS)), None, self.stop, 1024 * 1024, Resolver(io_loop=self.io_loop)) conn.set_request( b"\r\n".join(headers + [utf8("Content-Length: %d\r\n" % len(body))]) + b"\r\n" + body) response = self.wait() client.close() response.rethrow() return response
def test_reuse_request_from_response(self): # The response.request attribute should be an HTTPRequest, not # a _RequestProxy. # This test uses self.http_client.fetch because self.fetch calls # self.get_url on the input unconditionally. url = self.get_url('/hello') response = yield self.http_client.fetch(url) self.assertEqual(response.request.url, url) self.assertTrue(isinstance(response.request, HTTPRequest)) response2 = yield self.http_client.fetch(response.request) self.assertEqual(response2.body, b'Hello world!')
def test_request_set(self): proxy = _RequestProxy(HTTPRequest('http://example.com/', user_agent='foo'), dict()) self.assertEqual(proxy.user_agent, 'foo')
def test_default_set(self): proxy = _RequestProxy(HTTPRequest('http://example.com/'), dict(network_interface='foo')) self.assertEqual(proxy.network_interface, 'foo')
def test_both_set(self): proxy = _RequestProxy(HTTPRequest('http://example.com/', proxy_host='foo'), dict(proxy_host='bar')) self.assertEqual(proxy.proxy_host, 'foo')
def test_bad_attribute(self): proxy = _RequestProxy(HTTPRequest('http://example.com/'), dict()) with self.assertRaises(AttributeError): proxy.foo
def test_defaults_none(self): proxy = _RequestProxy(HTTPRequest('http://example.com/'), None) self.assertIs(proxy.auth_username, None)
def test_neither_set(self): proxy = _RequestProxy(HTTPRequest('http://example.com/'), dict()) self.assertIs(proxy.auth_username, None)
