19018276231

Committed 02 Nov 2025 09:17PM UTC coverage: 88.943%. First build

Build # 19018276231

Build Type

Pull #150

github

Committed by

web-flow

Commit Message

Merge e90838d9f into eb20b436a

Pull Request Pull Request #150: feat: add geotiff support valhalla isochrones

Run Details

9 of 9 new or added lines in 1 file covered. (100.0%)

1649 of 1854 relevant lines covered (88.94%)

0.89 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

92.5

/routingpy/client_base.py

# -*- coding: utf-8 -*-
# Copyright (C) 2021 GIS OPS UG
#
#
# Licensed under the Apache License, Version 2.0 (the "License"); you may not
# use this file except in compliance with the License. You may obtain a copy of
# the License at
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
# License for the specific language governing permissions and limitations under
# the License.
#
"""
Core client functionality, common across all routers.
"""


try:
    from . import __version__
except (ModuleNotFoundError, ImportError):
    __version__ = "None"

from abc import ABCMeta, abstractmethod
from datetime import timedelta
from urllib.parse import urlencode

import requests

_DEFAULT_USER_AGENT = "routingpy/v{}".format(__version__)
_RETRIABLE_STATUSES = set([503])


class options(object):
    """
    Contains default configuration options for all routers, e.g. `timeout` and `proxies`. Each router can take the same
    configuration values, which will override the values set in `options` object.

    Example for overriding default values for `user_agent` and proxies:

    >>> from routingpy.routers import options
    >>> from routingpy.routers import MapboxOSRM
    >>> options.default_user_agent = 'amazing_routing_app'
    >>> options.default_proxies = {'https': '129.125.12.0'}
    >>> router = MapboxOSRM(my_key)
    >>> print(router.client.headers)
    {'User-Agent': 'amazing_routing_app', 'Content-Type': 'application/x-www-form-urlencoded'}
    >>> print(router.client.proxies)
    {'https': '129.125.12.0'}

    Attributes:
        self.default_timeout:
            Combined connect and read timeout for HTTP requests, in
            seconds. Specify "None" for no timeout. Integer.

        self.default_retry_timeout:
            Timeout across multiple retriable requests, in
            seconds. Integer.

        self.default_retry_over_query_limit:
            If True, client will not raise an exception
            on HTTP 429, but instead jitter a sleeping timer to pause between
            requests until HTTP 200 or retry_timeout is reached. Boolean.

        self.default_skip_api_error:
            Continue with batch processing if a :class:`routingpy.exceptions.RouterApiError` is
            encountered (e.g. no route found). If False, processing will discontinue and raise an error. Boolean.

        self.default_user_agent:
            User-Agent to send with the requests to routing API. String.

        self.default_proxies:
            Proxies passed to the requests library. Dictionary.
    """

    default_timeout = 60
    default_retry_timeout = 60
    default_retry_over_query_limit = True
    default_skip_api_error = False
    default_user_agent = _DEFAULT_USER_AGENT
    default_proxies = None


# To avoid trouble when respecting timeout for individual routers (i.e. can't be None, since that's no timeout)
DEFAULT = type("object", (object,), {"__repr__": lambda self: "DEFAULT"})()


class BaseClient(metaclass=ABCMeta):
    """Abstract base class every client inherits from. Authentication is handled in each subclass."""

    def __init__(
        self,
        base_url,
        user_agent=None,
        timeout=DEFAULT,
        retry_timeout=None,
        retry_over_query_limit=None,
        skip_api_error=None,
        **kwargs
    ):
        """
        :param base_url: The base URL for the request. All routers must provide a default.
            Should not have a trailing slash.
        :type base_url: string

        :param user_agent: User-Agent to send with the requests to routing API.
            Overrides ``options.default_user_agent``.
        :type user_agent: string

        :param timeout: Combined connect and read timeout for HTTP requests, in
            seconds. Specify "None" for no timeout.
        :type timeout: int

        :param retry_timeout: Timeout across multiple retriable requests, in
            seconds.
        :type retry_timeout: int

        :param retry_over_query_limit: If True, client will not raise an exception
            on HTTP 429, but instead jitter a sleeping timer to pause between
            requests until HTTP 200 or retry_timeout is reached.
        :type retry_over_query_limit: bool

        :param skip_api_error: Continue with batch processing if a :class:`routingpy.exceptions.RouterApiError` is
            encountered (e.g. no route found). If False, processing will discontinue and raise an error. Default False.
        :type skip_api_error: bool

        :param **kwargs: Additional keyword arguments.
        :type **kwargs: dict
        """
        self.base_url = base_url

        self.retry_over_query_limit = (
            retry_over_query_limit
            if retry_over_query_limit is False
            else options.default_retry_over_query_limit
        )
        self.retry_timeout = timedelta(seconds=retry_timeout or options.default_retry_timeout)

        self.skip_api_error = skip_api_error or options.default_skip_api_error

        self.headers = {
            "User-Agent": user_agent or options.default_user_agent,
            "Content-Type": "application/json",
        }

        self.timeout = timeout if timeout != DEFAULT else options.default_timeout

        self.kwargs = kwargs

        self._req = None

    @abstractmethod
    def _request(
        self,
        url,
        get_params={},
        post_params=None,
        first_request_time=None,
        retry_counter=0,
        dry_run=None,
    ):
        """Performs HTTP GET/POST with credentials, returning the body as
        JSON.

        :param url: URL path for the request. Should begin with a slash.
        :type url: string

        :param get_params: HTTP GET parameters.
        :type get_params: dict or list of tuples

        :param post_params: HTTP POST parameters. Only specified by calling method.
        :type post_params: dict

        :param first_request_time: The time of the first request (None if no
            retries have occurred).
        :type first_request_time: :class:`datetime.datetime`

        :param retry_counter: The number of this retry, or zero for first attempt.
        :type retry_counter: int

        :param dry_run: If true, only prints URL and parameters. true or false.
        :type dry_run: bool

        :raises routingpy.exceptions.RouterApiError: when the API returns an error due to faulty configuration.
        :raises routingpy.exceptions.RouterServerError: when the API returns a server error.
        :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
        """
        pass

    @staticmethod
    def _generate_auth_url(path, params):
        """Returns the path and query string portion of the request URL, first
        adding any necessary parameters.

        :param path: The path portion of the URL.
        :type path: string

        :param params: URL parameters.
        :type params: dict or list of key/value tuples

        :rtype: string

        """

        if not params:
            return path

        if isinstance(params, dict):
            params = sorted(dict(**params).items())
        elif isinstance(params, (list, tuple)):
            params = params

        return path + "?" + requests.utils.unquote_unreserved(urlencode(params))

1	# -- coding: utf-8 --
2	# Copyright (C) 2021 GIS OPS UG
3	#
4	#
5	# Licensed under the Apache License, Version 2.0 (the "License"); you may not
6	# use this file except in compliance with the License. You may obtain a copy of
7	# the License at
8	#
9	# http://www.apache.org/licenses/LICENSE-2.0
10	#
11	# Unless required by applicable law or agreed to in writing, software
12	# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
13	# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
14	# License for the specific language governing permissions and limitations under
15	# the License.
16	#
17	"""
18	Core client functionality, common across all routers.
19	"""
20
21
22	try:	1✔
23	from . import __version__	1✔
24	except (ModuleNotFoundError, ImportError):	×
25	__version__ = "None"	×
26
27	from abc import ABCMeta, abstractmethod	1✔
28	from datetime import timedelta	1✔
29	from urllib.parse import urlencode	1✔
30
31	import requests	1✔
32
33	_DEFAULT_USER_AGENT = "routingpy/v{}".format(__version__)	1✔
34	_RETRIABLE_STATUSES = set([503])	1✔
35
36
37	class options(object):	1✔
38	"""
39	Contains default configuration options for all routers, e.g. `timeout` and `proxies`. Each router can take the same
40	configuration values, which will override the values set in `options` object.
41
42	Example for overriding default values for `user_agent` and proxies:
43
44	>>> from routingpy.routers import options
45	>>> from routingpy.routers import MapboxOSRM
46	>>> options.default_user_agent = 'amazing_routing_app'
47	>>> options.default_proxies = {'https': '129.125.12.0'}
48	>>> router = MapboxOSRM(my_key)
49	>>> print(router.client.headers)
50	{'User-Agent': 'amazing_routing_app', 'Content-Type': 'application/x-www-form-urlencoded'}
51	>>> print(router.client.proxies)
52	{'https': '129.125.12.0'}
53
54	Attributes:
55	self.default_timeout:
56	Combined connect and read timeout for HTTP requests, in
57	seconds. Specify "None" for no timeout. Integer.
58
59	self.default_retry_timeout:
60	Timeout across multiple retriable requests, in
61	seconds. Integer.
62
63	self.default_retry_over_query_limit:
64	If True, client will not raise an exception
65	on HTTP 429, but instead jitter a sleeping timer to pause between
66	requests until HTTP 200 or retry_timeout is reached. Boolean.
67
68	self.default_skip_api_error:
69	Continue with batch processing if a :class:`routingpy.exceptions.RouterApiError` is
70	encountered (e.g. no route found). If False, processing will discontinue and raise an error. Boolean.
71
72	self.default_user_agent:
73	User-Agent to send with the requests to routing API. String.
74
75	self.default_proxies:
76	Proxies passed to the requests library. Dictionary.
77	"""
78
79	default_timeout = 60	1✔
80	default_retry_timeout = 60	1✔
81	default_retry_over_query_limit = True	1✔
82	default_skip_api_error = False	1✔
83	default_user_agent = _DEFAULT_USER_AGENT	1✔
84	default_proxies = None	1✔
85
86
87	# To avoid trouble when respecting timeout for individual routers (i.e. can't be None, since that's no timeout)
88	DEFAULT = type("object", (object,), {"__repr__": lambda self: "DEFAULT"})()	1✔
89
90
91	class BaseClient(metaclass=ABCMeta):	1✔
92	"""Abstract base class every client inherits from. Authentication is handled in each subclass."""
93
94	def __init__(	1✔
95	self,
96	base_url,
97	user_agent=None,
98	timeout=DEFAULT,
99	retry_timeout=None,
100	retry_over_query_limit=None,
101	skip_api_error=None,
102	**kwargs
103	):
104	"""
105	:param base_url: The base URL for the request. All routers must provide a default.
106	Should not have a trailing slash.
107	:type base_url: string
108
109	:param user_agent: User-Agent to send with the requests to routing API.
110	Overrides ``options.default_user_agent``.
111	:type user_agent: string
112
113	:param timeout: Combined connect and read timeout for HTTP requests, in
114	seconds. Specify "None" for no timeout.
115	:type timeout: int
116
117	:param retry_timeout: Timeout across multiple retriable requests, in
118	seconds.
119	:type retry_timeout: int
120
121	:param retry_over_query_limit: If True, client will not raise an exception
122	on HTTP 429, but instead jitter a sleeping timer to pause between
123	requests until HTTP 200 or retry_timeout is reached.
124	:type retry_over_query_limit: bool
125
126	:param skip_api_error: Continue with batch processing if a :class:`routingpy.exceptions.RouterApiError` is
127	encountered (e.g. no route found). If False, processing will discontinue and raise an error. Default False.
128	:type skip_api_error: bool
129
130	:param **kwargs: Additional keyword arguments.
131	:type **kwargs: dict
132	"""
133	self.base_url = base_url	1✔
134
135	self.retry_over_query_limit = (	1✔
136	retry_over_query_limit
137	if retry_over_query_limit is False
138	else options.default_retry_over_query_limit
139	)
140	self.retry_timeout = timedelta(seconds=retry_timeout or options.default_retry_timeout)	1✔
141
142	self.skip_api_error = skip_api_error or options.default_skip_api_error	1✔
143
144	self.headers = {	1✔
145	"User-Agent": user_agent or options.default_user_agent,
146	"Content-Type": "application/json",
147	}
148
149	self.timeout = timeout if timeout != DEFAULT else options.default_timeout	1✔
150
151	self.kwargs = kwargs	1✔
152
153	self._req = None	1✔
154
155	@abstractmethod	1✔
156	def _request(	1✔
157	self,
158	url,
159	get_params={},
160	post_params=None,
161	first_request_time=None,
162	retry_counter=0,
163	dry_run=None,
164	):
165	"""Performs HTTP GET/POST with credentials, returning the body as
166	JSON.
167
168	:param url: URL path for the request. Should begin with a slash.
169	:type url: string
170
171	:param get_params: HTTP GET parameters.
172	:type get_params: dict or list of tuples
173
174	:param post_params: HTTP POST parameters. Only specified by calling method.
175	:type post_params: dict
176
177	:param first_request_time: The time of the first request (None if no
178	retries have occurred).
179	:type first_request_time: :class:`datetime.datetime`
180
181	:param retry_counter: The number of this retry, or zero for first attempt.
182	:type retry_counter: int
183
184	:param dry_run: If true, only prints URL and parameters. true or false.
185	:type dry_run: bool
186
187	:raises routingpy.exceptions.RouterApiError: when the API returns an error due to faulty configuration.
188	:raises routingpy.exceptions.RouterServerError: when the API returns a server error.
189	:raises routingpy.exceptions.RouterError: when anything else happened while requesting.
190	:raises routingpy.exceptions.JSONParseError: when the JSON response can't be parsed.
191	:raises routingpy.exceptions.Timeout: when the request timed out.
192	:raises routingpy.exceptions.TransportError: when something went wrong while trying to
193	execute a request.
194
195	:returns: raw JSON response.
196	:rtype: dict
197	"""
198	pass	×
199
200	@staticmethod	1✔
201	def _generate_auth_url(path, params):	1✔
202	"""Returns the path and query string portion of the request URL, first
203	adding any necessary parameters.
204
205	:param path: The path portion of the URL.
206	:type path: string
207
208	:param params: URL parameters.
209	:type params: dict or list of key/value tuples
210
211	:rtype: string
212
213	"""
214
215	if not params:	1✔
216	return path	1✔
217
218	if isinstance(params, dict):	1✔
219	params = sorted(dict(**params).items())	1✔
220	elif isinstance(params, (list, tuple)):	1✔
221	params = params	1✔
222
223	return path + "?" + requests.utils.unquote_unreserved(urlencode(params))	1✔

mthh / routingpy / 19018276231

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous