我们从Python开源项目中，提取了以下8个代码示例，用于说明如何使用six.moves.http_client.SERVICE_UNAVAILABLE。
def test__do_refresh_request_failure_w_json_error_and_desc(self): response = http_mock.ResponseMock( {'status': http_client.SERVICE_UNAVAILABLE}) base_error = 'Ruckus' error_desc = 'Can you describe the ruckus' content = json.dumps({ 'error': base_error, 'error_description': error_desc, }) error_msg = '{0}: {1}'.format(base_error, error_desc) self._do_refresh_request_test_helper(response, content, error_msg)
def leave_swarm(self, force=False): """ Leave a swarm. Args: force (bool): Leave the swarm even if this node is a manager. Default: ``False`` Returns: ``True`` if the request went through. Raises: :py:class:`docker.errors.APIError` If the server returns an error. """ url = self._url('/swarm/leave') response = self._post(url, params={'force': force}) # Ignore "this node is not part of a swarm" error if force and response.status_code == http_client.NOT_ACCEPTABLE: return True # FIXME: Temporary workaround for 1.13.0-rc bug # https://github.com/docker/docker/issues/29192 if force and response.status_code == http_client.SERVICE_UNAVAILABLE: return True self._raise_for_status(response) return True
def test_retry_on_503(self, resp_mock): """Test retry for status code that should be retried (e.g. 503)""" resp_mock.return_value.status_code = http_client.SERVICE_UNAVAILABLE with self.assertRaises(restclient.MaxRequestRetriesError): restclient.get('http://foo.com', '/')
def index_0(self, *tokens): """Provides an administrative interface for search indexing. Returns no output if successful; otherwise the response body will contain the failure details. """ try: cmd = tokens[0] except IndexError: cmd = "" # These commands cause the operation requested to be queued # for later execution. This does mean that if the operation # fails, the client won't know about it, but this is necessary # since these are long running operations (are likely to exceed # connection timeout limits). try: if cmd == "refresh": # Update search indexes. self.__bgtask.put(self.repo.refresh_index, pub=self._get_req_pub()) else: err = "Unknown index subcommand: {0}".format( cmd) cherrypy.log(err) raise cherrypy.HTTPError(http_client.NOT_FOUND, err) except queue.Full: raise cherrypy.HTTPError(http_client.SERVICE_UNAVAILABLE, "Another operation is already in progress; try " "again later.")
def search_1(self, *args, **params): # Raise assorted errors; if not, call superclass search_1. if self.need_nasty(): errs = [http_client.NOT_FOUND, http_client.BAD_REQUEST, http_client.SERVICE_UNAVAILABLE] code = random.choice(errs) cherrypy.log("NASTY search_1: HTTP {0:d}".format(code)) raise cherrypy.HTTPError(code) return DepotHTTP.search_1(self, *args, **params)
def feed(depot, request, response, pub): if depot.repo.mirror: raise cherrypy.HTTPError(http_client.NOT_FOUND, "Operation not supported in current server mode.") if not depot.repo.get_catalog(pub).updates: raise cherrypy.HTTPError(http_client.SERVICE_UNAVAILABLE, "No update history; unable to generate feed.") return pkg.server.feed.handle(depot, request, response, pub)
def check_resp_status_and_retry(resp, image_id, url): # Note(Jesse): This branch sorts errors into those that are permanent, # those that are ephemeral, and those that are unexpected. if resp.status in (httplib.BAD_REQUEST, # 400 httplib.UNAUTHORIZED, # 401 httplib.PAYMENT_REQUIRED, # 402 httplib.FORBIDDEN, # 403 httplib.METHOD_NOT_ALLOWED, # 405 httplib.NOT_ACCEPTABLE, # 406 httplib.PROXY_AUTHENTICATION_REQUIRED, # 407 httplib.CONFLICT, # 409 httplib.GONE, # 410 httplib.LENGTH_REQUIRED, # 411 httplib.PRECONDITION_FAILED, # 412 httplib.REQUEST_ENTITY_TOO_LARGE, # 413 httplib.REQUEST_URI_TOO_LONG, # 414 httplib.UNSUPPORTED_MEDIA_TYPE, # 415 httplib.REQUESTED_RANGE_NOT_SATISFIABLE, # 416 httplib.EXPECTATION_FAILED, # 417 httplib.UNPROCESSABLE_ENTITY, # 422 httplib.LOCKED, # 423 httplib.FAILED_DEPENDENCY, # 424 httplib.UPGRADE_REQUIRED, # 426 httplib.NOT_IMPLEMENTED, # 501 httplib.HTTP_VERSION_NOT_SUPPORTED, # 505 httplib.NOT_EXTENDED, # 510 ): raise PluginError("Got Permanent Error response [%i] while " "uploading image [%s] to glance [%s]" % (resp.status, image_id, url)) # Nova service would process the exception elif resp.status == httplib.NOT_FOUND: # 404 exc = XenAPI.Failure('ImageNotFound') raise exc # NOTE(nikhil): Only a sub-set of the 500 errors are retryable. We # optimistically retry on 500 errors below. elif resp.status in (httplib.REQUEST_TIMEOUT, # 408 httplib.INTERNAL_SERVER_ERROR, # 500 httplib.BAD_GATEWAY, # 502 httplib.SERVICE_UNAVAILABLE, # 503 httplib.GATEWAY_TIMEOUT, # 504 httplib.INSUFFICIENT_STORAGE, # 507 ): raise RetryableError("Got Ephemeral Error response [%i] while " "uploading image [%s] to glance [%s]" % (resp.status, image_id, url)) else: # Note(Jesse): Assume unexpected errors are retryable. If you are # seeing this error message, the error should probably be added # to either the ephemeral or permanent error list. raise RetryableError("Got Unexpected Error response [%i] while " "uploading image [%s] to glance [%s]" % (resp.status, image_id, url))
