我们从Python开源项目中，提取了以下27个代码示例，用于说明如何使用httplib2.ServerNotFoundError()。
def getGDataOAuthToken(gdataObj, credentials=None): if not credentials: credentials = getClientCredentials(API.FAM2_SCOPES) try: credentials.refresh(httplib2.Http(disable_ssl_certificate_validation=GC.Values[GC.NO_VERIFY_SSL])) except httplib2.ServerNotFoundError as e: systemErrorExit(NETWORK_ERROR_RC, str(e)) except oauth2client.client.AccessTokenRefreshError as e: return handleOAuthTokenError(str(e), False) gdataObj.additional_headers[u'Authorization'] = u'Bearer {0}'.format(credentials.access_token) if not GC.Values[GC.DOMAIN]: GC.Values[GC.DOMAIN] = credentials.id_token.get(u'hd', u'UNKNOWN').lower() if not GC.Values[GC.CUSTOMER_ID]: GC.Values[GC.CUSTOMER_ID] = GC.MY_CUSTOMER GM.Globals[GM.ADMIN] = credentials.id_token.get(u'email', u'UNKNOWN').lower() GM.Globals[GM.OAUTH2_CLIENT_ID] = credentials.client_id gdataObj.domain = GC.Values[GC.DOMAIN] gdataObj.source = GAM_INFO return True
def getGDataUserCredentials(api, user, i, count): userEmail = convertUIDtoEmailAddress(user) _, _, api_version, cred_family = API.getVersion(api) disc_file, discovery = readDiscoveryFile(api_version) GM.Globals[GM.CURRENT_API_USER] = userEmail credentials = getClientCredentials(cred_family) try: GM.Globals[GM.CURRENT_API_SCOPES] = list(set(list(discovery[u'auth'][u'oauth2'][u'scopes'])).intersection(credentials.scopes)) except KeyError: invalidDiscoveryJsonExit(disc_file) if not GM.Globals[GM.CURRENT_API_SCOPES]: systemErrorExit(NO_SCOPES_FOR_API_RC, Msg.NO_SCOPES_FOR_API.format(discovery.get(u'title', api_version))) credentials = getSvcAcctCredentials(GM.Globals[GM.CURRENT_API_SCOPES], userEmail) try: credentials.refresh(httplib2.Http(disable_ssl_certificate_validation=GC.Values[GC.NO_VERIFY_SSL])) return (userEmail, credentials) except httplib2.ServerNotFoundError as e: systemErrorExit(NETWORK_ERROR_RC, str(e)) except oauth2client.client.AccessTokenRefreshError as e: handleOAuthTokenError(str(e), True) entityUnknownWarning(Ent.USER, userEmail, i, count) return (userEmail, None)
def testGetViaHttpsKeyCert(self): # At this point I can only test # that the key and cert files are passed in # correctly to httplib. It would be nice to have # a real https endpoint to test against. http = httplib2.Http(timeout=2) http.add_certificate("akeyfile", "acertfile", "bitworking.org") try: (response, content) = http.request("https://bitworking.org", "GET") except AttributeError: self.assertEqual(http.connections["https:bitworking.org"].key_file, "akeyfile") self.assertEqual(http.connections["https:bitworking.org"].cert_file, "acertfile") except IOError: # Skip on 3.2 pass try: (response, content) = http.request("https://notthere.bitworking.org", "GET") except httplib2.ServerNotFoundError: self.assertEqual(http.connections["https:notthere.bitworking.org"].key_file, None) self.assertEqual(http.connections["https:notthere.bitworking.org"].cert_file, None) except IOError: # Skip on 3.2 pass
def submit(self, code=None): if not code: raise NoCodeError() # post the form to bpaste.net body = urlencode(dict(code=code, lexer=self.lexer, expiry=self.expiry)) try: response, content = self.http.request(self.url, 'POST', body=body, headers=self.headers) except ServerNotFoundError: raise CodeUploadError(-1, None) # check the response if response.status == 302 and response['location']: return response['location'] raise CodeUploadError(response.status, content)
def testGetUnknownServer(self): self.http.force_exception_to_status_code = False try: self.http.request("http://fred.bitworking.org/") self.fail( 'An httplib2.ServerNotFoundError Exception must be thrown on ' 'an unresolvable server.') except httplib2.ServerNotFoundError: pass # Now test with exceptions turned off self.http.force_exception_to_status_code = True (response, content) = self.http.request("http://fred.bitworking.org/") self.assertEqual(response['content-type'], 'text/plain') self.assertTrue(content.startswith(b"Unable to find")) self.assertEqual(response.status, 400)
def test_httplib2_in_out(): hostname = str(uuid4()) port = 9999 url = 'http://%s:%s/' % (hostname, port) http = Http() with Httplib2Interceptor(app=app, url=url) as target_url: response, content = http.request(target_url) assert response.status == 200 assert 'WSGI intercept successful!' in content.decode('utf-8') # outside the context manager the intercept does not work with py.test.raises(ServerNotFoundError): http.request(url) # Requests
def getAPIversionHttpService(api): hasLocalJSON = API.hasLocalJSON(api) api, version, api_version, cred_family = API.getVersion(api) httpObj = httplib2.Http(disable_ssl_certificate_validation=GC.Values[GC.NO_VERIFY_SSL], cache=GM.Globals[GM.CACHE_DIR]) if not hasLocalJSON: retries = 3 for n in range(1, retries+1): try: service = googleapiclient.discovery.build(api, version, http=httpObj, cache_discovery=False) if GM.Globals[GM.CACHE_DISCOVERY_ONLY]: httpObj.cache = None return (api_version, httpObj, service, cred_family) except httplib2.ServerNotFoundError as e: systemErrorExit(NETWORK_ERROR_RC, str(e)) except googleapiclient.errors.UnknownApiNameOrVersion as e: systemErrorExit(GOOGLE_API_ERROR_RC, Msg.UNKNOWN_API_OR_VERSION.format(str(e), __author__)) except (googleapiclient.errors.InvalidJsonError, KeyError, ValueError): httpObj.cache = None if n != retries: waitOnFailure(n, retries, INVALID_JSON_RC, Msg.INVALID_JSON_INFORMATION) continue systemErrorExit(INVALID_JSON_RC, Msg.INVALID_JSON_INFORMATION) except (http_client.ResponseNotReady, httplib2.SSLHandshakeError, socket.error) as e: errMsg = u'Connection error: {0}'.format(str(e) or repr(e)) if n != retries: waitOnFailure(n, retries, SOCKET_ERROR_RC, errMsg) continue systemErrorExit(SOCKET_ERROR_RC, errMsg) disc_file, discovery = readDiscoveryFile(api_version) try: service = googleapiclient.discovery.build_from_document(discovery, http=httpObj) if GM.Globals[GM.CACHE_DISCOVERY_ONLY]: httpObj.cache = None return (api_version, httpObj, service, cred_family) except (KeyError, ValueError): invalidDiscoveryJsonExit(disc_file)
def buildGAPIServiceObject(api, user): userEmail = convertUIDtoEmailAddress(user) _, httpObj, service, _ = getAPIversionHttpService(api) GM.Globals[GM.CURRENT_API_USER] = userEmail GM.Globals[GM.CURRENT_API_SCOPES] = API.getSvcAcctScopes(api) credentials = getSvcAcctCredentials(GM.Globals[GM.CURRENT_API_SCOPES], userEmail) try: service._http = credentials.authorize(httpObj) except httplib2.ServerNotFoundError as e: systemErrorExit(NETWORK_ERROR_RC, str(e)) except oauth2client.client.AccessTokenRefreshError as e: return (userEmail, handleOAuthTokenError(str(e), True)) return (userEmail, service)
def checkServiceAccount(users): checkForExtraneousArguments() all_scopes_pass = True all_scopes, jcount = API.getSortedSvcAcctScopesList() i, count, users = getEntityArgument(users) for user in users: i += 1 user = convertUIDtoEmailAddress(user) entityPerformActionNumItems([Ent.USER, user], jcount, Ent.SCOPE, i, count) Ind.Increment() j = 0 for scope in all_scopes: j += 1 try: credentials = getSvcAcctCredentials(scope, user) credentials.refresh(httplib2.Http(disable_ssl_certificate_validation=GC.Values[GC.NO_VERIFY_SSL])) result = u'PASS' except httplib2.ServerNotFoundError as e: systemErrorExit(NETWORK_ERROR_RC, str(e)) except oauth2client.client.HttpAccessTokenRefreshError: result = u'FAIL' all_scopes_pass = False entityActionPerformedMessage([Ent.SCOPE, u'{0:60}'.format(scope)], result, j, jcount) Ind.Decrement() service_account = credentials.serialization_data[u'client_id'] _, _, user_domain = splitEmailAddressOrUID(user) printBlankLine() if all_scopes_pass: printLine(Msg.SCOPE_AUTHORIZATION_PASSED.format(service_account)) else: printErrorMessage(SCOPES_NOT_AUTHORIZED, Msg.SCOPE_AUTHORIZATION_FAILED.format(user_domain, service_account, u',\n'.join(all_scopes)))
def updatePhoto(users): cd = buildGAPIObject(API.DIRECTORY) filenamePattern = getString(Cmd.OB_PHOTO_FILENAME_PATTERN) checkForExtraneousArguments() p = re.compile(u'^(ht|f)tps?://.*$') i, count, users = getEntityArgument(users) for user in users: i += 1 user, userName, _ = splitEmailAddressOrUID(user) filename = _substituteForUser(filenamePattern, user, userName) if p.match(filename): try: status, image_data = httplib2.Http(disable_ssl_certificate_validation=GC.Values[GC.NO_VERIFY_SSL]).request(filename, u'GET') if status[u'status'] != u'200': entityActionFailedWarning([Ent.USER, user, Ent.PHOTO, filename], Msg.NOT_ALLOWED, i, count) continue if status[u'content-location'] != filename: entityActionFailedWarning([Ent.USER, user, Ent.PHOTO, filename], Msg.NOT_FOUND, i, count) continue except (httplib2.HttpLib2Error, httplib2.ServerNotFoundError, httplib2.CertificateValidationUnsupported) as e: entityActionFailedWarning([Ent.USER, user, Ent.PHOTO, filename], str(e), i, count) continue else: image_data = readFile(filename, mode=u'rb', continueOnError=True, displayError=True) if image_data is None: entityActionFailedWarning([Ent.USER, user, Ent.PHOTO, filename], None, i, count) continue body = {u'photoData': base64.urlsafe_b64encode(image_data)} try: callGAPI(cd.users().photos(), u'update', throw_reasons=[GAPI.USER_NOT_FOUND, GAPI.FORBIDDEN, GAPI.INVALID_INPUT], userKey=user, body=body, fields=u'') entityActionPerformed([Ent.USER, user, Ent.PHOTO, filename], i, count) except GAPI.invalidInput as e: entityActionFailedWarning([Ent.USER, user, Ent.PHOTO, filename], str(e), i, count) except (GAPI.userNotFound, GAPI.forbidden): entityUnknownWarning(Ent.USER, user, i, count) # gam <UserTypeEntity> delete photo
def testGetUnknownServer(self): self.http.force_exception_to_status_code = False try: self.http.request("http://fred.bitworking.org/") self.fail("An httplib2.ServerNotFoundError Exception must be thrown on an unresolvable server.") except httplib2.ServerNotFoundError: pass # Now test with exceptions turned off self.http.force_exception_to_status_code = True (response, content) = self.http.request("http://fred.bitworking.org/") self.assertEqual(response['content-type'], 'text/plain') self.assertTrue(content.startswith("Unable to find")) self.assertEqual(response.status, 400)
def testGetUnknownServer(self): self.http.force_exception_to_status_code = False try: self.http.request("http://fred.bitworking.org/") self.fail("An httplib2.ServerNotFoundError Exception must be thrown on an unresolvable server.") except httplib2.ServerNotFoundError: pass # Now test with exceptions turned off self.http.force_exception_to_status_code = True (response, content) = self.http.request("http://fred.bitworking.org/") self.assertEqual(response['content-type'], 'text/plain') self.assertTrue(content.startswith(b"Unable to find")) self.assertEqual(response.status, 400)
def main(): flags = parse_cmdline() logger = configure_logs(flags.logfile) pageTokenFile = PageTokenFile(flags.ptokenfile) for i in range(RETRY_NUM): try: service = build_service(flags) pageToken = pageTokenFile.get() deletionList, pageTokenBefore, pageTokenAfter = \ get_deletion_list(service, pageToken, flags) pageTokenFile.save(pageTokenBefore) listEmpty = delete_old_files(service, deletionList, flags) except client.HttpAccessTokenRefreshError: print('Authentication error') except httplib2.ServerNotFoundError as e: print('Error:', e) except TimeoutError: print('Timeout: Google backend error.') print('Retries unsuccessful. Abort action.') return else: break time.sleep(RETRY_INTERVAL) else: print("Retries unsuccessful. Abort action.") return if listEmpty: pageTokenFile.save(pageTokenAfter)
def testGetViaHttpsKeyCert(self): # At this point I can only test # that the key and cert files are passed in # correctly to httplib. It would be nice to have # a real https endpoint to test against. http = httplib2.Http(timeout=2) http.add_certificate("akeyfile", "acertfile", "bitworking.org") try: (response, content) = http.request("https://bitworking.org", "GET") except AttributeError: self.assertEqual( http.connections["https:bitworking.org"].key_file, "akeyfile") self.assertEqual( http.connections[ "https:bitworking.org"].cert_file, "acertfile") except IOError: # Skip on 3.2 pass try: (response, content) = http.request( "https://notthere.bitworking.org", "GET") except httplib2.ServerNotFoundError: self.assertEqual( http.connections["https:notthere.bitworking.org"].key_file, None) self.assertEqual( http.connections["https:notthere.bitworking.org"].cert_file, None) except IOError: # Skip on 3.2 pass
def _map_exception(e): """Maps an exception from urlib3 to httplib2.""" if isinstance(e, urllib3.exceptions.MaxRetryError): if not e.reason: return e e = e.reason message = e.args[0] if e.args else '' if isinstance(e, urllib3.exceptions.ResponseError): if 'too many redirects' in message: return httplib2.RedirectLimit(message) if isinstance(e, urllib3.exceptions.NewConnectionError): if ('Name or service not known' in message or 'nodename nor servname provided, or not known' in message): return httplib2.ServerNotFoundError( 'Unable to find hostname.') if 'Connection refused' in message: return socket.error((errno.ECONNREFUSED, 'Connection refused')) if isinstance(e, urllib3.exceptions.DecodeError): return httplib2.FailedToDecompressContent( 'Content purported as compressed but not uncompressable.', httplib2.Response({'status': 500}), '') if isinstance(e, urllib3.exceptions.TimeoutError): return socket.timeout('timed out') if isinstance(e, urllib3.exceptions.SSLError): return ssl.SSLError(*e.args) return e
def doGAMCheckForUpdates(forceCheck=False): import calendar def _gamLatestVersionNotAvailable(): if forceCheck: systemErrorExit(NETWORK_ERROR_RC, Msg.GAM_LATEST_VERSION_NOT_AVAILABLE) current_version = __version__ now_time = calendar.timegm(time.gmtime()) if forceCheck: check_url = GAM_ALL_RELEASES # includes pre-releases else: last_check_time_str = readFile(GM.Globals[GM.LAST_UPDATE_CHECK_TXT], continueOnError=True, displayError=False) last_check_time = int(last_check_time_str) if last_check_time_str and last_check_time_str.isdigit() else 0 if last_check_time > now_time-604800: return check_url = GAM_LATEST_RELEASE # latest full release try: _, c = httplib2.Http(disable_ssl_certificate_validation=GC.Values[GC.NO_VERIFY_SSL]).request(check_url, u'GET', headers={u'Accept': u'application/vnd.github.v3.text+json'}) try: release_data = json.loads(c) except ValueError: _gamLatestVersionNotAvailable() return if isinstance(release_data, list): release_data = release_data[0] # only care about latest release if not isinstance(release_data, dict) or u'tag_name' not in release_data: _gamLatestVersionNotAvailable() return latest_version = release_data[u'tag_name'] if latest_version[0].lower() == u'v': latest_version = latest_version[1:] if forceCheck or (latest_version > current_version): printKeyValueList([u'Version Check', None]) Ind.Increment() printKeyValueList([u'Current', current_version]) printKeyValueList([u' Latest', latest_version]) Ind.Decrement() if latest_version <= current_version: writeFile(GM.Globals[GM.LAST_UPDATE_CHECK_TXT], str(now_time), continueOnError=True, displayError=forceCheck) return announcement = release_data.get(u'body_text', u'No details about this release') writeStderr(u'\nGAM %s release notes:\n\n' % latest_version) writeStderr(announcement) try: printLine(Msg.HIT_CONTROL_C_TO_UPDATE) time.sleep(15) except KeyboardInterrupt: import webbrowser webbrowser.open(release_data[u'html_url']) printLine(Msg.GAM_EXITING_FOR_UPDATE) sys.exit(0) writeFile(GM.Globals[GM.LAST_UPDATE_CHECK_TXT], str(now_time), continueOnError=True, displayError=forceCheck) return except (httplib2.HttpLib2Error, httplib2.ServerNotFoundError, httplib2.CertificateValidationUnsupported): return
def callGData(service, function, soft_errors=False, throw_errors=None, retry_errors=None, **kwargs): import gdata.apps.service if throw_errors is None: throw_errors = [] if retry_errors is None: retry_errors = [] all_retry_errors = GDATA.NON_TERMINATING_ERRORS+retry_errors method = getattr(service, function) retries = 10 for n in range(1, retries+1): try: return method(**kwargs) except gdata.apps.service.AppsForYourDomainException as e: error_code, error_message = checkGDataError(e, service) if (n != retries) and (error_code in all_retry_errors): waitOnFailure(n, retries, error_code, error_message) continue if error_code in throw_errors: if error_code in GDATA.ERROR_CODE_EXCEPTION_MAP: raise GDATA.ERROR_CODE_EXCEPTION_MAP[error_code](error_message) raise if soft_errors: stderrErrorMsg(u'{0} - {1}{2}'.format(error_code, error_message, [u'', u': Giving up.'][n > 1])) return None if error_code == GDATA.INSUFFICIENT_PERMISSIONS: APIAccessDeniedExit() systemErrorExit(GOOGLE_API_ERROR_RC, u'{0} - {1}'.format(error_code, error_message)) except oauth2client.client.AccessTokenRefreshError as e: handleOAuthTokenError(str(e), GDATA.SERVICE_NOT_APPLICABLE in throw_errors) raise GDATA.ERROR_CODE_EXCEPTION_MAP[GDATA.SERVICE_NOT_APPLICABLE](str(e)) except (http_client.ResponseNotReady, httplib2.SSLHandshakeError, socket.error) as e: errMsg = u'Connection error: {0}'.format(str(e) or repr(e)) if n != retries: waitOnFailure(n, retries, SOCKET_ERROR_RC, errMsg) continue if soft_errors: writeStderr(convertUTF8(u'\n{0}{1} - Giving up.\n'.format(ERROR_PREFIX, errMsg))) return None systemErrorExit(SOCKET_ERROR_RC, errMsg) except httplib2.ServerNotFoundError as e: systemErrorExit(NETWORK_ERROR_RC, str(e))
def callGAPI(service, function, silent_errors=False, soft_errors=False, throw_reasons=None, retry_reasons=None, **kwargs): if throw_reasons is None: throw_reasons = [] if retry_reasons is None: retry_reasons = [] all_retry_reasons = GAPI.DEFAULT_RETRY_REASONS+retry_reasons method = getattr(service, function) retries = 10 svcparms = dict(kwargs.items()+GM.Globals[GM.EXTRA_ARGS_LIST]) for n in range(1, retries+1): try: return method(**svcparms).execute() except googleapiclient.errors.HttpError as e: http_status, reason, message = checkGAPIError(e, soft_errors=soft_errors, silent_errors=silent_errors, retryOnHttpError=n < 3, service=service) if http_status == -1: continue if http_status == 0: return None if (n != retries) and (reason in all_retry_reasons): waitOnFailure(n, retries, reason, message) continue if reason in throw_reasons: if reason in GAPI.REASON_EXCEPTION_MAP: raise GAPI.REASON_EXCEPTION_MAP[reason](message) raise e if soft_errors: stderrErrorMsg(u'{0}: {1} - {2}{3}'.format(http_status, reason, message, [u'', u': Giving up.'][n > 1])) return None if reason == GAPI.INSUFFICIENT_PERMISSIONS: APIAccessDeniedExit() systemErrorExit(HTTP_ERROR_RC, formatHTTPError(http_status, reason, message)) except oauth2client.client.AccessTokenRefreshError as e: handleOAuthTokenError(str(e), GAPI.SERVICE_NOT_AVAILABLE in throw_reasons) raise GAPI.REASON_EXCEPTION_MAP[GAPI.SERVICE_NOT_AVAILABLE](str(e)) except httplib2.CertificateValidationUnsupported: noPythonSSLExit() except (http_client.ResponseNotReady, httplib2.SSLHandshakeError, socket.error) as e: errMsg = u'Connection error: {0}'.format(str(e) or repr(e)) if n != retries: waitOnFailure(n, retries, SOCKET_ERROR_RC, errMsg) continue if soft_errors: writeStderr(convertUTF8(u'\n{0}{1} - Giving up.\n'.format(ERROR_PREFIX, errMsg))) return None systemErrorExit(SOCKET_ERROR_RC, errMsg) except ValueError as e: if service._http.cache is not None: service._http.cache = None continue systemErrorExit(GOOGLE_API_ERROR_RC, str(e)) except TypeError as e: systemErrorExit(GOOGLE_API_ERROR_RC, str(e)) except httplib2.ServerNotFoundError as e: systemErrorExit(NETWORK_ERROR_RC, str(e))
