Copyright (C) 2020-2022 HERE Global B.V. and its affiliate(s).

All rights reserved.

This software and other materials contain proprietary information

controlled by HERE and are protected by applicable copyright legislation.

Any use and utilization of this software and other materials and

disclosure to any third parties is conditional upon having a separate

agreement with HERE for the access, use, utilization or disclosure of this

software. In the absence of such agreement, the use of the software is not

allowed.

"""
This module contains a :class:BaseApi class to perform API operations.
"""
import logging
import urllib.request
from collections.abc import MutableMapping
from datetime import datetime
from typing import TYPE_CHECKING, Optional, Union
from urllib.parse import urlparse
from uuid import uuid4

import backoff
import requests
import requests.adapters

from importlib.metadata import version # isort:skip

version = version("here-platform") # isort:skip

from here.platform.config import ApplicationConfig, PlatformConfig
from here.platform.exceptions import (
AuthenticationException,
DigestMismatchException,
InequalReadsException,
NonceAlreadyUsedException,
ResourceLimitExceededException,
ServiceUnavailableException,
TooManyRequestsException,
)
from requests import ConnectionError, Response

if TYPE_CHECKING:
from here.platform.auth import Auth

[docs]
class BaseApi:
"""
This class provides access to some Restful API operations.
"""

def init(
self,
platform_config: PlatformConfig,
application_config: ApplicationConfig,
auth: Optional["Auth"],
proxies: Optional[dict] = None,
):
"""
Instantiate API with auth token.

:param platform_config: a mandatory :class:PlatformConfig object to provide
configuration information for the API.
:param application_config: a mandatory :class:ApplicationConfig object to provide
configuration information for the API.
:param auth: an Authentication instance.
:param proxies: an optional proxy configuration. Defaults to the environment proxy
configuration.
"""
self.auth = auth
self.platform_config = platform_config
self.application_config = application_config
self.proxies = proxies or urllib.request.getproxies()

Full URL is provided for platform config api when using Local DataService.

TODO: Provide a more direct way to check for LDS.

self.is_local = bool(urlparse(self.platform_config.api).scheme)

self._user_agent = "PySDK/" + version

create our own dedicated HTTP connection pool.

self._http_pool = requests.Session()
adapter = requests.adapters.HTTPAdapter()

use our pool for both http and https

self._http_pool.mount("http://", adapter)
self._http_pool.mount("https://", adapter)

self._is_recording = (
bool(self.application_config.additional_parameters["is_recording"])
if self.application_config
and "is_recording" in self.application_config.additional_parameters
else False
)

@property
def headers(self) -> dict:
"""
Return HTTP request headers with Bearer token in Authorization
field.

:return: authorization tokens
"""
return {"Authorization": f"Bearer {self.auth.token}"} if self.auth else

[docs]
def format_url(self, base_url: str, path: str) -> str:
"""
Formats the URL based on the base URL and path.

:param base_url: The base URL for the request.
:param path: The path for the URL. This is expected to have the leading '/'.
:return: The formatted URL.
"""
assert path.startswith("/")
return f"{base_url}{path}"

[docs]
def get(
self,
url: str,
params: Optional[dict] = None,
headers: Optional[MutableMapping] = None,
**kwargs,
) -> Response:
"""
Perform a get request of an API at a specified URL with backoff.

:param url: URL of the API.
:param params: Parameters to pass to the API.
:param headers: Request headers. Defaults to the Api headers property.
:param kwargs: Optional arguments that request takes.
:return: response from the API.
"""
return self.request("GET", url, params=params, headers=headers, **kwargs)

[docs]
def head(
self,
url: str,
params: Optional[dict] = None,
headers: Optional[MutableMapping] = None,
**kwargs,
) -> Response:
"""
Perform a head request of an API at specified URL with backoff.

:param url: URL of the API.
:param params: Parameters to pass to the API.
:param headers: Request headers. Defaults to the api headers property.
:param kwargs: Optional arguments that request takes.
:return: response from the API.
"""
return self.request("HEAD", url, params=params, headers=headers, **kwargs)

[docs]
def post(
self,
url: str,
data: Optional[Union[dict, list, bytes, str]] = None,
params: Optional[dict] = None,
headers: Optional[MutableMapping] = None,
**kwargs,
) -> Response:
"""
Perform a post request of an API at a specified URL with backoff.

:param url: URL of the API.
:param data: Post data for http request.
:param params: Parameters to pass to the API.
:param headers: Request headers. Defaults to the api headers property.
:param kwargs: Optional arguments that request takes.
:return: response from the API.
"""
if isinstance(data, (dict, list)):
return self.request("POST", url, json=data, params=params, headers=headers, **kwargs)
else:
return self.request("POST", url, data=data, params=params, headers=headers, **kwargs)

[docs]
def put(
self,
url: str,
data: Optional[Union[dict, list, bytes, str]] = None,
params: Optional[dict] = None,
headers: Optional[MutableMapping] = None,
**kwargs,
) -> Response:
"""
Perform a put request of an API at a specified URL with backoff.

:param url: URL of the API
:param data: Put data for http request.
:param params: Parameters to pass to the API.
:param headers: Request headers. Defaults to the api headers property.
:param kwargs: Optional arguments that request takes.
:return: response from the API.
"""
if isinstance(data, (dict, list)):
return self.request("PUT", url, json=data, params=params, headers=headers, **kwargs)
else:
return self.request("PUT", url, data=data, params=params, headers=headers, **kwargs)

[docs]
def patch(
self,
url: str,
data: Optional[Union[dict, list, bytes, str]] = None,
params: Optional[dict] = None,
headers: Optional[MutableMapping] = None,
**kwargs,
) -> Response:
"""
Perform a patch request of an API at a specified URL with backoff.

:param url: URL of the API
:param data: Patch data for http request.
:param params: Parameters to pass to the API.
:param headers: Request headers. Defaults to the api headers property.
:param kwargs: Optional arguments that request takes.
:return: response from the API.
"""
if isinstance(data, (dict, list)):
return self.request("PATCH", url, json=data, params=params, headers=headers, **kwargs)
else:
return self.request("PATCH", url, data=data, params=params, headers=headers, **kwargs)

[docs]
def delete(
self,
url: str,
params: Optional[dict] = None,
headers: Optional[MutableMapping] = None,
**kwargs,
) -> Response:
"""
Perform a delete request of an API at a specified URL with backoff.

:param url: URL of the API
:param params: parameters to pass to the API.
:param headers: Request headers. Defaults to the api headers property.
:param kwargs: Optional arguments that request takes.
:return: response from the API.
"""
return self.request("DELETE", url, params=params, headers=headers, **kwargs)

[docs]
def request(
self,
method: str,
url: str,
headers: Optional[MutableMapping] = None,
stream: bool = False,
**kwargs,
) -> Response:
"""
Performs a request of an API at a specified URL with a backoff.
:param method: The HTTP method to perform.
:param url: URL of the API
:param headers: Request headers. Defaults to the api headers property.
:param stream: whether to stream data.
:param kwargs: Optional arguments that request takes.
:return: response from the API.
"""

headers = headers or self.headers
headers["User-Agent"] = self._user_agent

x-idempotency-key is used by multiple APIs to identify a request and detect retries where

otherwise errors such as 409 (conflict) may be returned if a retry is interpreted as a

completely separate request.

headers["x-idempotency-key"] = str(uuid4())

def validate_resp_content_length(resp):
"""
Validate content actual length vs expected length.
If length differs raise InequalReadsException.

:param resp: Response object.
:raises InequalReadsException: if content actual vs expected length differs.
"""
expected_length = resp.headers.get("Content-Length")
if expected_length:
expected_length = int(expected_length)
actual_length = int(resp.raw.tell())

The length check must not be done if runnung pytest recording.

if not self._is_recording:
if actual_length != expected_length:
raise InequalReadsException(resp)

Inner backoff loop retries on connection error. Outer backoff loop retries based on too

many requests.

@backoff.on_exception(
backoff.expo,
(
DigestMismatchException,
InequalReadsException,
NonceAlreadyUsedException,
ServiceUnavailableException,
TooManyRequestsException,
requests.exceptions.ChunkedEncodingError,
requests.exceptions.ReadTimeout,
),
max_time=self.application_config.retry_max_time,
max_value=self.application_config.retry_max_wait,
)
@backoff.on_exception(
backoff.constant,
(ConnectionError, requests.exceptions.ConnectTimeout),
max_time=self.application_config.connect_retry_max_time,
)
def request_impl() -> Response:
"""
Perform a request of an API at a specified URL.

:raises TooManyRequestsException: If platform responds with HTTP 429.
:raises ServiceUnavailableException: If platform responds with HTTP 408, 500, 502, 503,
or 504.
:raises AuthenticationException: If platform responds with HTTP 401 or 403.
:raises NonceAlreadyUsedException: If platform responds with HTTP 401 due to nonce
re-use and retries have been exceeded.
:raises ResourceLimitExceededException: If platform responds with HTTP 402.
:raises DigestMismatchException: If platform responds with HTTP 400 due to the provided
digest not matching the data.
:return: response from the API.
"""
logging.debug(f"{method} {url}")
start_time = datetime.now()
resp = self._http_pool.request(
method,
url,
headers=headers,
proxies=self.proxies,
timeout=(
self.application_config.connect_timeout,
self.application_config.read_timeout,
),
stream=stream,
**kwargs,
)
run_time = datetime.now() - start_time
logging.debug(f"{method} {url} - {resp.status_code} - {run_time.total_seconds()} sec.")

By-Passing HEAD method because Response will have headers['content-length']

but no message body. So that InequalReadsException is not raised.

The HTTP HEAD method requests the headers that would be returned

if the HEAD request's URL was instead requested with the HTTP GET method.

For example,if a URL might produce a large download,

a HEAD request could read its Content-Length header to check the filesize

without actually downloading the file.

https://developer.mozilla.org/en-US/docs/Web/HTTP/Methods/HEAD

if (
not stream
and method not in ["HEAD"]
and resp.status_code not in [404, 429, 500, 502, 503, 504]
):
validate_resp_content_length(resp=resp)

if resp.status_code == 429:
raise TooManyRequestsException(resp)

Request timeout, internal server error, invalid gateway, service unavailable,

or gateway timeout.

elif resp.status_code in [408, 500, 502, 503, 504]:
raise ServiceUnavailableException(resp)
elif resp.status_code == 402:
raise ResourceLimitExceededException(resp)
elif resp.status_code in [401, 403]:
try:

Special error case for the nonce already used, which just needs a retry.

is_nonce_error = (
resp.json()["errorCode"] == NonceAlreadyUsedException.AAA_ERROR_CODE
)
except Exception:

If the error code couldn't be retrieved, assume it's not a nonce error.

is_nonce_error = False
if is_nonce_error:
raise NonceAlreadyUsedException(resp)
raise AuthenticationException(resp)
elif resp.status_code == 400:

Check for digest mismatch.

is_digest_error = False
try:
for detail in resp.json()["detail"]:
if detail.get("name") == "digest":
is_digest_error = True
break
except Exception:

If the error name couldn't be retrieved, assume it's not a digest error.

pass
if is_digest_error:
raise DigestMismatchException(resp)
return resp

result_resp: Response = request_impl()
return result_resp