How to use bravado - 10 common examples

exc_class = exception_families.get(
            (status_code // 100) * 100,
        ) or HTTPError

    return exc_class(response, message=message, swagger_result=swagger_result)


class HTTPRedirection(HTTPError):
    """3xx responses."""


class HTTPClientError(HTTPError):
    """4xx responses."""


class HTTPServerError(HTTPError):
    """5xx responses."""

    def __str__(self):
        # type: (...) -> str
        # Try to surface the most useful/relevant information available
        # since this is the first thing a developer sees when bad things
        # happen.
        status_and_reason = str(self.response)
        message = ': ' + self.message if self.message else ''
        text = ': ' + self.response.text if self.response.text else ''
        result = ': {0}'.format(self.swagger_result) \
            if self.swagger_result is not None else ''
        return '{0}{1}{2}{3}'.format(status_and_reason, message, text, result)


# The follow are based on the HTTP Status Code Registry at

general :class:`HTTPError` class will be returned.
    :type response: :class:`bravado_core.response.IncomingResponse`
    :param message: Optional string message
    :param swagger_result: If the response for this HTTPError is
        documented in the swagger spec, then this should be the result
        value of the response.
    :return: An HTTP exception class that can be raised
    """
    status_code = response.status_code
    exc_class = status_map.get(status_code)

    if exc_class is None:
        exception_families = {
            300: HTTPRedirection,
            400: HTTPClientError,
            500: HTTPServerError,
        }
        exc_class = exception_families.get(
            (status_code // 100) * 100,
        ) or HTTPError

    return exc_class(response, message=message, swagger_result=swagger_result)

Return an HTTP exception class  based on the response. If a specific
    class doesn't exist for a particular HTTP status code, a more
    general :class:`HTTPError` class will be returned.
    :type response: :class:`bravado_core.response.IncomingResponse`
    :param message: Optional string message
    :param swagger_result: If the response for this HTTPError is
        documented in the swagger spec, then this should be the result
        value of the response.
    :return: An HTTP exception class that can be raised
    """
    status_code = response.status_code
    exc_class = status_map.get(status_code)

    if exc_class is None:
        exception_families = {
            300: HTTPRedirection,
            400: HTTPClientError,
            500: HTTPServerError,
        }
        exc_class = exception_families.get(
            (status_code // 100) * 100,
        ) or HTTPError

    return exc_class(response, message=message, swagger_result=swagger_result)

def setup_paasta_api_client():
    return SwaggerClient.from_url(get_paasta_api_url())

to_wire=lambda uri: uri,
            to_python=lambda uri: uri,
            validate=lambda uri_string: True
    )

    bravado_config = {
        'also_return_response': True,
        'formats': [uri_format]
        }

    service = None
    start_time = time.perf_counter()
    cur_time = start_time
    while cur_time < start_time + 10:
        try:
            service = SwaggerClient.from_url('http://localhost:29593/swagger.json', config=bravado_config)
            _, response = service.jobs.get_jobs().result()
            if response.status_code == 200:
                break
        except HTTPBadGateway:
            pass
        except requests.exceptions.ConnectionError:
            pass
        time.sleep(0.1)
        cur_time = time.perf_counter()

    if cur_time >= start_time + 10:
        print("Warning: Cerise container failed to come up")
    return service

def test__init_http_client(mock_trs_config):
    mock_opts = mock_trs_config['mock_trs']
    test_http_client = _init_http_client(opts=mock_opts)

    assert isinstance(test_http_client, RequestsClient)
    assert test_http_client.authenticator.host == mock_opts['host']
    assert test_http_client.authenticator.api_key == mock_opts['auth']

def service_instance_status_error(context, error_code, job_id):
    (service, instance, _, __) = decompose_job_id(job_id)

    response = None
    try:
        response = context.paasta_api_client.service.status_instance(
            instance=instance, service=service
        ).result()
    except bexception.HTTPError as exc:
        assert exc.status_code == int(error_code)

    assert not response

def load_trs_client(service_id, http_client=None):
    """
    Return an API client for the selected workflow execution service.
    """
    if http_client is None:
        http_client = _init_http_client(service_id=service_id)

    spec_path = os.path.join(os.path.dirname(os.path.dirname(__file__)),
                             'ga4gh-tool-discovery.yaml')
    spec_path = os.path.abspath(spec_path)

    opts = _get_trs_opts(service_id)
    api_url = '{}://{}'.format(opts['proto'], opts['host'])

    loader = Loader(http_client, request_headers=None)
    spec_dict = loader.load_spec('file:///{}'.format(spec_path),
                                 base_url=api_url)
    spec_client = SwaggerClient.from_spec(spec_dict,
                                          origin_url=api_url,
                                          http_client=http_client,
                                          config={'use_models': False})
    return spec_client.GA4GH

ret = utils.get_task_from_instance(
            "cluster1", "my-service", "main", slave_hostname="test"
        )

    mock_api.service.tasks_instance.side_effect = HTTPError(
        response=mock.Mock(status_code=500)
    )
    with raises(utils.PaastaTaskNotFound):
        ret = utils.get_task_from_instance(
            "cluster1", "my-service", "main", slave_hostname="test"
        )
    mock_api.service.tasks_instance.assert_called_with(
        service="my-service", instance="main", verbose=True, slave_hostname="test"
    )

    mock_api.service.tasks_instance.side_effect = HTTPNotFound(
        response=mock.Mock(status_code=404)
    )
    with raises(utils.PaastaTaskNotFound):
        ret = utils.get_task_from_instance(
            "cluster1", "my-service", "main", slave_hostname="test"
        )

def get_state(job_id):
        """Returns a tuple of whether the job exists, and if so the state."""
        try:
            test_job, response = service.jobs.get_job_by_id(jobId=job_id).result()
            assert response.status_code == 200
        except HTTPNotFound:
                return False, None
        return True, test_job