我们从Python开源项目中,提取了以下23个代码示例,用于说明如何使用urllib2.HTTPErrorProcessor()。
def test_errors(self): h = urllib2.HTTPErrorProcessor() o = h.parent = MockOpener() url = "http://example.com/" req = Request(url) # all 2xx are passed through r = MockResponse(200, "OK", {}, "", url) newr = h.http_response(req, r) self.assertTrue(r is newr) self.assertTrue(not hasattr(o, "proto")) # o.error not called r = MockResponse(202, "Accepted", {}, "", url) newr = h.http_response(req, r) self.assertTrue(r is newr) self.assertTrue(not hasattr(o, "proto")) # o.error not called r = MockResponse(206, "Partial content", {}, "", url) newr = h.http_response(req, r) self.assertTrue(r is newr) self.assertTrue(not hasattr(o, "proto")) # o.error not called # anything else calls o.error (and MockOpener returns None, here) r = MockResponse(502, "Bad gateway", {}, "", url) self.assertTrue(h.http_response(req, r) is None) self.assertEqual(o.proto, "http") # o.error called self.assertEqual(o.args, (req, r, 502, "Bad gateway", {}))
def _urllib2_fetch(self, uri, params, method=None): # install error processor to handle HTTP 201 response correctly if self.opener == None: self.opener = urllib2.build_opener(HTTPErrorProcessor) urllib2.install_opener(self.opener) if method and method == 'GET': uri = self._build_get_uri(uri, params) req = PlivoUrlRequest(uri) else: req = PlivoUrlRequest(uri, urllib.urlencode(params)) if method and (method == 'DELETE' or method == 'PUT'): req.http_method = method authstring = base64.encodestring('%s:%s' % (self.auth_id, self.auth_token)) authstring = authstring.replace('\n', '') req.add_header("Authorization", "Basic %s" % authstring) response = urllib2.urlopen(req) return response.read()
def __init__(self, base_url, exc_class=None, logger=None): """ @param base_url: The base url to the API. @param exc_class: An exception class to handle non-200 results. Creates an HTTP(S) client to connect to the Cloudera Manager API. """ self._base_url = base_url.rstrip('/') self._exc_class = exc_class or RestException self._logger = logger or LOG self._headers = { } # Make a basic auth handler that does nothing. Set credentials later. self._passmgr = urllib2.HTTPPasswordMgrWithDefaultRealm() authhandler = urllib2.HTTPBasicAuthHandler(self._passmgr) # Make a cookie processor cookiejar = cookielib.CookieJar() self._opener = urllib2.build_opener( HTTPErrorProcessor(), urllib2.HTTPCookieProcessor(cookiejar), authhandler)
def getHtml(url, tryEncode=True, followRedirect=True): class NoRedirection(urllib2.HTTPErrorProcessor): def http_response(self, request, response): code, msg, hdrs = response.code, response.msg, response.info() return response https_response = http_response if followRedirect: opener=urllib2.build_opener() else: opener=urllib2.build_opener(NoRedirection) try: page=opener.open(url) pageHtml=page.read() except: opener.close() return None if tryEncode: try: charset = re.findall('charset=(.*?)$', page.info()['Content-Type'])[0].lower() if charset != 'utf-8': pageHtml = pageHtml.decode(charset) #?????? ???????? ? ??????????? except: pass pageHtml = strUniEncode(pageHtml) opener.close() return pageHtml
def http_response(self, request, response): code = response.code if config.TRACE_API_CALLS: logger.info("took %2.2fs: (%i)" % (time.time()-request.start_time,code)) if code/100 in (2, 4, 5): return response else: urllib2.HTTPErrorProcessor.http_response(self, request, response)
def _prepare_http_request(self, uri, params, method='POST'): # install error processor to handle HTTP 201 response correctly if self.opener is None: self.opener = urllib2.build_opener(HTTPErrorProcessor) urllib2.install_opener(self.opener) proxy_url = self.proxy_url if proxy_url: proxy = proxy_url.split('http://')[1] proxyhandler = urllib2.ProxyHandler({'http': proxy}) opener = urllib2.build_opener(proxyhandler) urllib2.install_opener(opener) if method and method == 'GET': uri = self._build_get_uri(uri, params) _request = HTTPUrlRequest(uri) else: _request = HTTPUrlRequest(uri, urllib.urlencode(params)) if method and (method == 'DELETE' or method == 'PUT'): _request.http_method = method _request.add_header('User-Agent', self.USER_AGENT) if self.auth_id and self.auth_token: # append the POST variables sorted by key to the uri # and transform None to '' and unicode to string s = uri for k, v in sorted(params.items()): if k: if v is None: x = '' else: x = str(v) s += k + x # compute signature and compare signatures signature = base64.encodestring(hmac.new(self.auth_token, s, sha1).\ digest()).strip() _request.add_header("X-PLIVO-SIGNATURE", "%s" % signature) return _request
def _GetOpener(self): """Returns an OpenerDirector that supports cookies and ignores redirects. Returns: A urllib2.OpenerDirector object. """ opener = urllib2.OpenerDirector() opener.add_handler(urllib2.ProxyHandler()) opener.add_handler(urllib2.UnknownHandler()) opener.add_handler(urllib2.HTTPHandler()) opener.add_handler(urllib2.HTTPDefaultErrorHandler()) opener.add_handler(urllib2.HTTPSHandler()) opener.add_handler(urllib2.HTTPErrorProcessor()) if self.save_cookies: self.cookie_file = os.path.expanduser("~/.codereview_upload_cookies") self.cookie_jar = cookielib.MozillaCookieJar(self.cookie_file) if os.path.exists(self.cookie_file): try: self.cookie_jar.load() self.authenticated = True StatusUpdate("Loaded authentication cookies from %s" % self.cookie_file) except (cookielib.LoadError, IOError): # Failed to load cookies - just ignore them. pass else: # Create an empty cookie file with mode 600 fd = os.open(self.cookie_file, os.O_CREAT, 0600) os.close(fd) # Always chmod the cookie file os.chmod(self.cookie_file, 0600) else: # Don't save cookies across runs of update.py. self.cookie_jar = cookielib.CookieJar() opener.add_handler(urllib2.HTTPCookieProcessor(self.cookie_jar)) return opener
def _GetOpener(self): """Returns an OpenerDirector that supports cookies and ignores redirects. Returns: A urllib2.OpenerDirector object. """ opener = urllib2.OpenerDirector() opener.add_handler(urllib2.ProxyHandler()) opener.add_handler(urllib2.UnknownHandler()) opener.add_handler(urllib2.HTTPHandler()) opener.add_handler(urllib2.HTTPDefaultErrorHandler()) opener.add_handler(urllib2.HTTPSHandler()) opener.add_handler(urllib2.HTTPErrorProcessor()) if self.save_cookies: self.cookie_file = os.path.expanduser("~/.codereview_upload_cookies") self.cookie_jar = cookielib.MozillaCookieJar(self.cookie_file) if os.path.exists(self.cookie_file): try: self.cookie_jar.load() self.authenticated = True StatusUpdate("Loaded authentication cookies from %s" % self.cookie_file) except (cookielib.LoadError, IOError): # Failed to load cookies - just ignore them. pass else: # Create an empty cookie file with mode 600 fd = os.open(self.cookie_file, os.O_CREAT, 0o600) os.close(fd) # Always chmod the cookie file os.chmod(self.cookie_file, 0o600) else: # Don't save cookies across runs of update.py. self.cookie_jar = cookielib.CookieJar() opener.add_handler(urllib2.HTTPCookieProcessor(self.cookie_jar)) return opener
def http_response(self, request, response): if 200 <= response.code < 300: return response return urllib2.HTTPErrorProcessor.http_response(self, request, response)
def _GetOpener(self): """Returns an OpenerDirector that supports cookies and ignores redirects. Returns: A urllib2.OpenerDirector object. """ opener = urllib2.OpenerDirector() opener.add_handler(fancy_urllib.FancyProxyHandler()) opener.add_handler(urllib2.UnknownHandler()) opener.add_handler(urllib2.HTTPHandler()) opener.add_handler(urllib2.HTTPDefaultErrorHandler()) opener.add_handler(fancy_urllib.FancyHTTPSHandler()) opener.add_handler(urllib2.HTTPErrorProcessor()) opener.add_handler(ContentEncodingHandler()) if self.save_cookies: self.cookie_jar.filename = os.path.expanduser( HttpRpcServer.DEFAULT_COOKIE_FILE_PATH) if os.path.exists(self.cookie_jar.filename): try: self.cookie_jar.load() self.authenticated = True logger.debug("Loaded authentication cookies from %s", self.cookie_jar.filename) except (OSError, IOError, cookielib.LoadError), e: logger.debug("Could not load authentication cookies; %s: %s", e.__class__.__name__, e) self.cookie_jar.filename = None else: try: fd = os.open(self.cookie_jar.filename, os.O_CREAT, 0600) os.close(fd) except (OSError, IOError), e: logger.debug("Could not create authentication cookies file; %s: %s", e.__class__.__name__, e) self.cookie_jar.filename = None opener.add_handler(urllib2.HTTPCookieProcessor(self.cookie_jar)) return opener