How to use the routingpy.exceptions function in routingpy

To help you get started, we’ve selected a few routingpy examples, based on popular ways it is used in public projects.

Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.

github gis-ops / routing-py / tests / test_base.py View on Github external
def test_retry_timeout(self):
        query = self.params
        responses.add(
            responses.POST,
            'https://httpbin.org/post',
            json=query,
            status=429,
            content_type='application/json'
        )

        client = RouterMock(base_url="https://httpbin.org", retry_over_query_limit=True, retry_timeout=3)
        with self.assertRaises(routingpy.exceptions.OverQueryLimit):
            client.directions(url='/post', post_params=query)
github gis-ops / routing-py / tests / test_base.py View on Github external
def test_router_by_name(self):
        for s in routingpy.routers._SERVICE_TO_ROUTER.keys():
            router = routingpy.routers.get_router_by_name(s)

        with self.assertRaises(routingpy.exceptions.RouterNotFound):
            router = routingpy.routers.get_router_by_name('orsm')
github gis-ops / routing-py / routingpy / routers / base.py View on Github external
def _get_body(response):
        status_code = response.status_code

        try:
            body = response.json()
        except json.decoder.JSONDecodeError:
            raise exceptions.JSONParseError("Can't decode JSON response:{}".format(response.text))

        if status_code == 429:
            raise exceptions.OverQueryLimit(status_code, body)

        if 400 <= status_code < 500:
            raise exceptions.RouterApiError(status_code, body)

        if 500 <= status_code:
            raise exceptions.RouterServerError(status_code, body)

        if status_code != 200:
            raise exceptions.RouterError(status_code, body)

        return body
github gis-ops / routing-py / routingpy / routers / base.py View on Github external
def _get_body(response):
        status_code = response.status_code

        try:
            body = response.json()
        except json.decoder.JSONDecodeError:
            raise exceptions.JSONParseError("Can't decode JSON response:{}".format(response.text))

        if status_code == 429:
            raise exceptions.OverQueryLimit(status_code, body)

        if 400 <= status_code < 500:
            raise exceptions.RouterApiError(status_code, body)

        if 500 <= status_code:
            raise exceptions.RouterServerError(status_code, body)

        if status_code != 200:
            raise exceptions.RouterError(status_code, body)

        return body
github gis-ops / routing-py / routingpy / routers / base.py View on Github external
:raises routingpy.exceptions.RouterError: when anything else happened while requesting.
        :raises routingpy.exceptions.JSONParseError: when the JSON response can't be parsed.
        :raises routingpy.exceptions.Timeout: when the request timed out.
        :raises routingpy.exceptions.TransportError: when something went wrong while trying to
            execute a request.

        :returns: raw JSON response.
        :rtype: dict
        """

        if not first_request_time:
            first_request_time = datetime.now()

        elapsed = datetime.now() - first_request_time
        if elapsed > self.retry_timeout:
            raise exceptions.Timeout()

        if retry_counter > 0:
            # 0.5 * (1.5 ^ i) is an increased sleep time of 1.5x per iteration,
            # starting at 0.5s when retry_counter=1. The first retry will occur
            # at 1, so subtract that first.
            delay_seconds = 1.5**(retry_counter - 1)

            # Jitter this value by 50% and pause.
            time.sleep(delay_seconds * (random.random() + 0.5))

        authed_url = self._generate_auth_url(url, get_params)

        # Default to the client-level self.requests_kwargs, with method-level
        # requests_kwargs arg overriding.
        requests_kwargs = requests_kwargs or {}
        final_requests_kwargs = dict(self.requests_kwargs, **requests_kwargs)
github gis-ops / routing-py / routingpy / routers / base.py View on Github external
# Only print URL and parameters for dry_run
        if dry_run:
            print(
                "url:\n{}\nParameters:\n{}".format(
                    self.base_url + authed_url, json.dumps(final_requests_kwargs, indent=2)
                )
            )
            return

        try:
            response = requests_method(self.base_url + authed_url, **final_requests_kwargs)
            self._req = response.request

        except requests.exceptions.Timeout:
            raise exceptions.Timeout()

        tried = retry_counter + 1

        if response.status_code in _RETRIABLE_STATUSES:
            # Retry request.
            warnings.warn(
                'Server down.\nRetrying for the {}{} time.'.format(tried, get_ordinal(tried)),
                UserWarning
            )

            return self._request(
                url, get_params, post_params, first_request_time, retry_counter + 1, requests_kwargs
            )

        try:
            result = self._get_body(response)
github gis-ops / routing-py / routingpy / routers / base.py View on Github external
# Retry request.
            warnings.warn(
                'Server down.\nRetrying for the {}{} time.'.format(tried, get_ordinal(tried)),
                UserWarning
            )

            return self._request(
                url, get_params, post_params, first_request_time, retry_counter + 1, requests_kwargs
            )

        try:
            result = self._get_body(response)

            return result

        except exceptions.RouterApiError:
            if self.skip_api_error:
                warnings.warn(
                    "Router {} returned an API error with "
                    "the following message:\n{}".format(self.__class__.__name__, response.text)
                )
                return

            raise

        except exceptions.RetriableRequest as e:
            if isinstance(e, exceptions.OverQueryLimit) and not self.retry_over_query_limit:
                raise

            warnings.warn(
                'Rate limit exceeded.\nRetrying for the {}{} time.'.format(tried, get_ordinal(tried)),
                UserWarning