Copyright (C) 2021-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.

"""HERE platform service abstraction."""

import webbrowser
from typing import TYPE_CHECKING, Any, Dict, List, Optional, Union

from here.platform.api.base_api import BaseApi
from here.platform.api.factory.apifactory import API
from here.platform.auth import Auth
from here.platform.config.application_config import ApplicationConfig
from here.platform.config.platform_config import PlatformConfig
from here.platform.credentials import PlatformCredentials
from here.platform.environment import Environment
from here.platform.exceptions import PlatformException
from here.platform.utils import JsonDictDocument

if TYPE_CHECKING:
from here.platform.platform import Platform

[docs]
class ServiceConfiguration(JsonDictDocument):
"""
A JSON document for service configuration used in APIs loaded into a Dict.

Basic fields like name, summary, base_url, logging_url,
and tags are provided for reading.
"""

@property
def version(self) -> Optional[str]:
"""Version in the service config"""
return str(self.json["version"]) if "version" in self.json else None

@property
def name(self) -> Optional[str]:
"""Name in the service config"""
return str(self.json["name"]) if "name" in self.json else None

@property
def summary(self) -> Optional[str]:
"""Summary in the service config"""
return str(self.json["summary"]) if "summary" in self.json else None

@property
def base_url(self) -> Optional[str]:
"""Base URL in the service config"""
return str(self.json["baseUrl"]) if "baseUrl" in self.json else None

@property
def logging_url(self):
"""Logging URL of the service"""
return str(self.json["loggingUrl"]) if "loggingUrl" in self.json else None

@property
def tags(self) -> List[str]:
"""List of tags present in the service"""
return [str(t) for t in self.json.get("tags", [])]

[docs]
class ServiceVersion(JsonDictDocument):
"""
A JSON document for service version used in APIs loaded into a Dict.
"""

@property
def api_version(self) -> str:
"""API Version of the service"""
return str(self.json["apiVersion"])

@property
def service_version(self) -> str:
"""Service Version of the service"""
return str(self.json["serviceVersion"])

@property
def data_versions(self) -> List[Dict]:
"""List of data versions in the service"""
return list(self.json["dataVersions"])

[docs]
class ServiceHealth(JsonDictDocument):
"""
A JSON document for service health used in APIs loaded into a Dict.
"""

@property
def status(self) -> str:
"""Health status of the service"""
return str(self.json["status"])

[docs]
class Service:
"""
HERE platform service abstraction.

Access basic fields like name, summary, base_url,
logging_url, and tags of a service. Example: service.name

More fields can be read by using service configuration
Example: service.configuration
"""

def init(
self,
hrn: str,
configuration: ServiceConfiguration,
platform: "Platform",
):
"""
Instantiate the :class:Service for the given :param:hrn

:param hrn: the HERE Resource Name of the service
:param configuration: an instance of ServiceConfiguration
:param platform: instance of Platform
:raises ValueError: No service config for HRN given

For more information about proxy configuration, please see:
https://requests.readthedocs.io/en/master/user/advanced/#proxies

"""
self.hrn: str = hrn
self.platform = platform
self.version = configuration.version
self.name = configuration.name
self.summary = configuration.summary
self.base_url = configuration.base_url
self.logging_url: str = configuration.logging_url
self.tags: list = configuration.tags
if not configuration:
raise ValueError(f"No service config for HRN {hrn} given")
self.configuration: ServiceConfiguration = configuration
self._credentials: Optional[PlatformCredentials] = platform.credentials
self._environment: Environment = platform.environment
self._platform_config: PlatformConfig = platform.platform_config
self._application_config: ApplicationConfig = platform.application_config # type: ignore
self._proxies: dict = platform.proxies # type: ignore
self._auth: Auth = platform.auth # type: ignore
self._base_api: BaseApi = platform.base_api
self._api_factory = self.platform.api_factory

@property
def health(self):
"""Current health of the service"""
service_api = self._api_factory.get_api(API.SERVICE, service_base_url=self.base_url)
service_health = service_api.health()
return ServiceHealth.from_dict(service_health)

[docs]
def get_version(self):
"""Detailed version information of the service"""
service_version_dict = self._api_factory.get_api(
API.SERVICE, service_base_url=self.base_url
).version()
return ServiceVersion.from_dict(service_version_dict)

[docs]
def open_in_portal(self):
"""Opens the service page on HERE platform portal."""
portal_url = self._platform_config.portal_url
if portal_url is None:
raise RuntimeError("here_platform_portal_url is not present in configuration.")
webbrowser.open_new("/services/details//overview".format(portal_url, self.hrn))

def _format_url(self, path: str) -> str:
"""Builds absolute URL of the service using its base URL and given path"""
if not path.startswith("/"):
raise ValueError("Malformed path for service ''. Must begin with '/'".format(path))
return "".format(self.base_url, path)

[docs]
def get(
self,
path: str,
params: Optional[dict] = None,
headers: Optional[dict] = None,
as_json: bool = True,
success_codes: List[int] = [200],
**kwargs,
) -> Any:
"""
Perform a GET request on the service for the given :param:path of the service
Example, service.get("/path")

:param path: path to be called on the service.
:param params: Parameters to pass to the API.
:param headers: Request headers. Defaults to the Api headers property.
:param as_json: A boolean to indicate type of response. If True then response
will be returned in json format. If False, requests.Response object
will be returned
:param success_codes: List of http response codes which are treated as success
:param kwargs: Optional arguments that can be passed to requests' GET call
:return: response from the API.
:raises PlatformException: in case of unsuccessful response
"""
url = self._format_url(path)
resp = self._api_factory.get_api(API.SERVICE, service_base_url=self.base_url).get(
url, params, headers, **kwargs
)
if resp.status_code in success_codes:
return resp.json() if as_json else resp
else:
raise PlatformException(resp)

[docs]
def head(
self,
path: str,
params: Optional[dict] = None,
headers: Optional[dict] = None,
as_json: bool = False,
success_codes: List[int] = [200],
**kwargs,
) -> Any:
"""
Perform a HEAD request on the service for the given :param:path of the service
Example, service.head("/path")

:param path: path to be called on the service.
:param params: Parameters to pass to the API.
:param headers: Request headers. Defaults to the api headers property.
:param as_json: A boolean to indicate type of response. If True then response
will be returned in json format.
:param success_codes: List of http response codes which are treated as success
:param kwargs: Optional arguments that can be passed to requests' HEAD call
:return: response from the API.
:raises PlatformException: in case of unsuccessful response
"""
url = self._format_url(path)
resp = self._api_factory.get_api(API.SERVICE, service_base_url=self.base_url).head(
url, params, headers, **kwargs
)
if resp.status_code in success_codes:
return resp.json() if as_json else resp
else:
raise PlatformException(resp)

[docs]
def post(
self,
path: str,
data: Optional[Union[dict, List, bytes, str]] = None,
params: Optional[dict] = None,
headers: Optional[dict] = None,
as_json: bool = True,
success_codes: List[int] = [200, 201],
**kwargs,
) -> Any:
"""
Perform a POST request on the service for the given :param:path of the service
Example, service.post("/path", params={...})

:param path: path to be called on the service.
:param data: Dictionary, list of tuples, bytes, or file-like
object to send in the body of the http request.
:param params: Parameters to pass to the API.
:param headers: Request headers. Defaults to the api headers property.
Overrides the default headers one by one leaving the ones that are not overridden
:param as_json: A boolean to indicate type of response. If True then response
will be returned in json format.
:param success_codes: List of http response codes which are treated as success
:param kwargs: Optional arguments that can be passed to requests' POST call
:return: response from the API.
:raises PlatformException: in case of unsuccessful response
"""
url = self._format_url(path)
resp = self._api_factory.get_api(API.SERVICE, service_base_url=self.base_url).post(
url, data, params, headers, **kwargs
)
if resp.status_code in success_codes:
return resp.json() if as_json else resp
else:
raise PlatformException(resp)

[docs]
def put(
self,
path: str,
data: Optional[dict] = None,
params: Optional[dict] = None,
headers: Optional[dict] = None,
as_json: bool = True,
success_codes: List[int] = [200],
**kwargs,
) -> Any:
"""
Perform a PUT request on the service for the given :param:path of the service
Example, service.put("/path", params={...})

:param path: path to be called on the service.
: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 as_json: A boolean to indicate type of response. If True then response
will be returned in json format.
:param success_codes: List of http response codes which are treated as success
:param kwargs: Optional arguments that can be passed to requests' PUT call
:return: response from the API.
:raises PlatformException: in case of unsuccessful response
"""
url = self._format_url(path)
resp = self._api_factory.get_api(API.SERVICE, service_base_url=self.base_url).put(
url, data, params, headers, **kwargs
)
if resp.status_code in success_codes:
return resp.json() if as_json else resp
else:
raise PlatformException(resp)

[docs]
def patch(
self,
path: str,
data: Optional[dict] = None,
params: Optional[dict] = None,
headers: Optional[dict] = None,
as_json: bool = True,
success_codes: List[int] = [200],
**kwargs,
) -> Any:
"""
Perform a PATCH request on the service for the given :param:path of the service
Example, service.patch("/path", params={...})

:param path: path to be called on the service.
: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 as_json: A boolean to indicate type of response. If True then response
will be returned in json format.
:param success_codes: List of http response codes which are treated as success
:param kwargs: Optional arguments that can be passed to requests' PATCH call
:return: response from the API.
:raises PlatformException: in case of unsuccessful response
"""
url = self._format_url(path)
resp = self._api_factory.get_api(API.SERVICE, service_base_url=self.base_url).patch(
url, data, params, headers, **kwargs
)
if resp.status_code in success_codes:
return resp.json() if as_json else resp
else:
raise PlatformException(resp)

[docs]
def delete(
self,
path: str,
params: Optional[Dict] = None,
headers: Optional[dict] = None,
as_json: bool = True,
success_codes: List[int] = [200],
**kwargs,
) -> Any:
"""
Perform a DELETE request on the service for the given :param:path of the service
Example, service.delete("/path", params={...})

:param path: path to be called on the service.
:param params: parameters to pass to the API.
:param headers: Request headers. Defaults to the api headers property.
:param as_json: A boolean to indicate type of response. If True then response
will be returned in json format.
:param success_codes: List of http response codes which are treated as success
:param kwargs: Optional arguments that can be passed to requests' DELETE call
:raises PlatformException: in case of unsuccessful response
:return: response from the API.
"""
url = self._format_url(path)
resp = self._api_factory.get_api(API.SERVICE, service_base_url=self.base_url).delete(
url, params, headers, **kwargs
)
if resp.status_code in success_codes:
return resp.json() if as_json else resp
else:
raise PlatformException(resp)

def str(self) -> str:
"""
String representation of service
"""
return 'Service "HRN" "Name" "Version" "Base URL" '.format(
self.hrn, self.name, self.version, self.base_url
)