Source code for here.platform.api.factory.apifactory

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 manages single instance of APIs and Registries.
"""

import importlib
from collections import namedtuple
from enum import Enum
from typing import Any, Dict, Optional, Union

from here.platform.api.aaa_authorization_api import AAAAuthorizationApi
from here.platform.api.artifact_api import ArtifactApi
from here.platform.api.lookup_api import LookupApi
from here.platform.api.registry import LookupApiRegistry
from here.platform.api.service_registry_api import ServiceRegistryApi
from here.platform.auth import Auth
from here.platform.config.application_config import ApplicationConfig
from here.platform.config.platform_config import PlatformConfig

Namedtuple to store APIs' required info that'll be used in their instance creation

ApiData = namedtuple("ApiData", ["name", "module_name", "class_name", "api_version_name"])

[docs]
class API(Enum):
"""
Enum of APIs carrying information required during API instance creation.
"""

BASE = ApiData("base_api", "here.platform.api.base_api", "BaseApi", None)
DATA_CONFIG = ApiData(
"data_config_api", "here.platform.api.data_config_api", "DataConfigApi", None
)
LOOKUP = ApiData("lookup_api", "here.platform.api.lookup_api", "LookupApi", None)
ARTIFACT = ApiData("artifact_api", "here.platform.api.artifact_api", "ArtifactApi", None)
AAA_AUTH = ApiData(
"aaa_auth_api", "here.platform.api.aaa_authorization_api", "AAAAuthorizationApi", None
)
SERVICE_REGISTRY = ApiData(
"service_registry_api",
"here.platform.api.service_registry_api",
"ServiceRegistryApi",
None,
)
DATA_BLOB = ApiData("blob_api", "here.platform.api.data_blob_api", "DataBlobApi", "blob-v1")
DATA_VOLATILE_BLOB = ApiData(
"data_volatile_blob_api",
"here.platform.api.data_volatile_blob_api",
"DataVolatileBlobApi",
"volatile-blob-v1",
)
DATA_METADATA = ApiData(
"datametadeta_api", "here.platform.api.data_metadata_api", "DataMetadataApi", "metadata-v1"
)
DATA_PUBLISH = ApiData(
"data_publish_api", "here.platform.api.data_publish_api", "DataPublishApi", "publish-v2"
)
DATA_QUERY = ApiData(
"data_query_api", "here.platform.api.data_query_api", "DataQueryApi", "query-v1"
)
DATA_INDEX = ApiData(
"data_index_api", "here.platform.api.data_index_api", "DataIndexApi", "index-v1"
)
DATA_STREAM = ApiData(
"data_stream_api", "here.platform.api.data_stream_api", "DataStreamApi", "stream-v2"
)
DATA_INGEST = ApiData(
"data_ingest_api", "here.platform.api.data_ingest_api", "DataIngestApi", "ingest-v1"
)
DATA_INTERACTIVE = ApiData(
"data_interactive_api",
"here.platform.api.data_interactive_api",
"DataInteractiveApi",
"interactive-v1",
)
DATA_OBJECT_BLOB = ApiData(
"data_object_blob_api",
"here.platform.api.data_object_blob_api",
"DataObjectBlobApi",
"blob-v2",
)
DATA_STATISTICS = ApiData(
"data_statistics_api",
"here.platform.api.data_statistics_api",
"DataStatisticsApi",
"statistics-v1",
)
SERVICE = ApiData("service_api", "here.platform.api.service_api", "ServiceApi", None)
MAP_MATCHING = ApiData(
"map_matching_api", "here.platform.api.map_matching_api", "MapMatchingApi", None
)

class _Factory:
"""
Parent class for API factories
"""

def init(
self,
platform_config: PlatformConfig,
application_config: ApplicationConfig,
auth: Auth,
proxies: Dict,
local_environment: Optional[bool] = False,
lookup_api_registry: Optional[LookupApiRegistry] = LookupApiRegistry(),
):
"""
Instantiates factory for API instances.

:param platform_config: the properties of Platform Configuration.
:param application_config: an instance of ApplicationConfig.
:param auth: an Authentication instance.
:param proxies: a proxy configuration.
:param local_environment: boolean value to indicate whether
environment is local_environment.
:param lookup_api_registry: LookupApiRegistry instance.
"""
self._platform_config = platform_config
self._application_config = application_config
self._auth = auth
self._proxies = proxies
self._local_environment = local_environment
self._lookup_api_registry = lookup_api_registry

self._api_cache: Dict[API, Any] =
self._api_args = {"platform_config": self._platform_config, "application_config": self._application_config, "auth": self._auth, "proxies": self._proxies,}

def get_api(self, api: API) -> Any:
"""
Gets API against :param:api.

:param api: API enum value for which API instance is needed.
:return: API instance agaisnt the given :param:api.
"""
if api not in self._api_cache:
self._create_api(api)
return self._api_cache[api]

def _create_api(self, api: API) -> None:
"""
Skeleton function overridden in child classes to create
instance for given :param:api.

:param api: API enum value for which API instance is needed.
"""
pass

class _PlatformFactory(_Factory):
"""
Factory class for Platform API instances
"""

:param platform_config: the properties of Platform Configuration.
:param application_config: an instance of ApplicationConfig.
:param auth: an Authentication instance.
:param proxies: a proxy configuration.
:param local_environment: boolean value to indicate whether
environment is local_environment.
:param lookup_api_registry: LookupApiRegistry instance.
"""
super().init(
platform_config,
application_config,
auth,
proxies,
local_environment=local_environment,
lookup_api_registry=lookup_api_registry,
)

def _get_artifact_base_url(self) -> str:
"""
Gets Artifact URL from the Lookup API.

:return: Artifact URL.
"""
lookup_api: LookupApi = self.get_api(API.LOOKUP)
return str(lookup_api.get_impl_platform_api_list()["artifact"]["baseURL"])

def _get_account_url(self) -> str:
"""
Gets Account URL.

:return: Account URL.
:raises ValueError: If account_url is not present in the configuration.
"""
if self._platform_config.account_url is None:
raise ValueError("here_account_api_url is not present in the configuration.")
account_url: str = self._platform_config.account_url + "/authorization/v1.1"
return account_url

def _location_service_registry(self) -> str:
"""
Gets Location Service Registry URL from the Lookup API.

:return: Location Service Registry URL.
"""
lookup_api: LookupApi = self.get_api(API.LOOKUP)
return str(lookup_api.get_impl_platform_api_list()["location-service-registry"]["baseURL"])

def _create_api(self, api: API) -> None:
"""
Creates API instance for the given :param:api.

:param api: Enum type of the API to be created.
:raises NotImplementedError: If called when LOCAL environment (LDS endpoint) is active.
"""

if api in [API.AAA_AUTH, API.SERVICE_REGISTRY]:
if self._local_environment:
raise NotImplementedError(f"{api.name} API is not available in LOCAL environment")

if api == API.LOOKUP:
self._api_cache[api] = LookupApi(
lookup_api_registry=self._lookup_api_registry, # type: ignore
**self._api_args, # type: ignore
)
return
elif api == API.ARTIFACT:
self._api_cache[api] = ArtifactApi(
base_url=self._get_artifact_base_url(),
**self._api_args, # type: ignore
)
return
elif api == API.AAA_AUTH:
self._api_cache[api] = AAAAuthorizationApi(
base_url=self._get_account_url(),
**self._api_args, # type: ignore
)
return
elif api == API.SERVICE_REGISTRY:
self._api_cache[api] = ServiceRegistryApi(
base_url=self._location_service_registry(),
**self._api_args, # type: ignore
)
return

api_class = getattr(importlib.import_module(api.value.module_name), api.value.class_name)
self._api_cache[api] = api_class(**self._api_args)

class _CatalogApiFactory(_Factory):
"""
Factory class for Catalog API instances.
"""

def init(
self,
catalog_hrn: str,
platformfactory: PlatformFactory,
platform_config: PlatformConfig,
application_config: ApplicationConfig,
auth: Auth,
proxies: Dict,
local_environment: Optional[bool] = False,
):
"""
Instantiates factory for API instances.

:param catalog_hrn: HRN of the catalog for which :class:_CatalogFactory is created.
:param platform_factory: instance of :class:_PlatformFactory.
:param platform_config: the properties of Platform Configuration.
:param application_config: an instance of ApplicationConfig.
:param auth: an Authentication instance.
:param proxies: a proxy configuration.
:param local_environment: boolean value to indicate whether
environment is local_environment.
"""
super().init(
platform_config, application_config, auth, proxies, local_environment=local_environment
)
self._catalog_hrn = catalog_hrn
self._platform_factory = platform_factory
self._resource_apis: dict =

def _get_resource_apis(self) -> dict:
"""
Gets the collection of reource APIs for the current Catalog.
"""
if self._resource_apis is None or len(self._resource_apis) == 0:
if self._catalog_hrn is None:
raise ValueError("Catalog HRN is not set in the APIFactory")
lookup_api: LookupApi = self._platform_factory.get_api(API.LOOKUP)
self._resource_apis: dict = lookup_api.get_resource_api_list( # type: ignore
self._catalog_hrn
)
return self._resource_apis

def _create_api(self, api: API) -> None:
"""
Creates API instance for the given :param:api.

:param api: Enum type of the API to be created.
:raises NotImplementedError: If called when LOCAL environment (LDS endpoint) is active.
"""

if api in [
API.DATA_INGEST,
API.DATA_INTERACTIVE,
API.DATA_STATISTICS,
]:
if self._local_environment:
raise NotImplementedError(f"{api.name} API is not available in LOCAL environment")

api_class = getattr(importlib.import_module(api.value.module_name), api.value.class_name)
self._api_cache[api] = api_class(
self._get_resource_apis()[api.value.api_version_name]["baseURL"], **self._api_args
)

class _MapMatchingFactory(_Factory):
def init(
self,
map_matching_base_url: str,
platformfactory: PlatformFactory,
platform_config: PlatformConfig,
application_config: ApplicationConfig,
auth: Auth,
proxies: Dict,
local_environment: Optional[bool] = False,
):
super().init(platform_config, application_config, auth, proxies, local_environment)
self._map_matching_base_url = map_matching_base_url
self._platform_factory = platform_factory

def _create_api(self, api: API) -> None:
"""
Creates API instance for the given :param:api.

:param api: Enum type of the API to be created.
:raises NotImplementedError: If called when LOCAL environment (LDS endpoint) is active.
"""

if api in [API.SERVICE]:
if self._local_environment:
raise NotImplementedError(f"{api.name} API is not available in LOCAL environment")

api_class = getattr(importlib.import_module(api.value.module_name), api.value.class_name)
self._api_cache[api] = api_class(self._map_matching_base_url, **self._api_args)

class _ServiceFactory(_Factory):
"""
Factory class for Service API instances.
"""

def init(
self,
service_base_url: str,
platformfactory: PlatformFactory,
platform_config: PlatformConfig,
application_config: ApplicationConfig,
auth: Auth,
proxies: Dict,
local_environment: Optional[bool] = False,
):
"""
Instantiates factory for API instances.

:param service_base_url: base URL of the service for which
:class:_ServiceFactory is created.
:param platform_factory: instance of :class:_PlatformFactory.
:param platform_config: the properties of Platform Configuration.
:param application_config: an instance of ApplicationConfig.
:param auth: an Authentication instance.
:param proxies: a proxy configuration.
:param local_environment: boolean value to indicate whether
environment is local_environment.
"""
super().init(platform_config, application_config, auth, proxies, local_environment)
self._service_base_url = service_base_url
self._platform_factory = platform_factory

def _create_api(self, api: API) -> None:
"""
Creates API instance for the given :param:api.

:param api: Enum type of the API to be created.
:raises NotImplementedError: If called when LOCAL environment (LDS endpoint) is active.
"""

if api in [API.SERVICE]:
if self._local_environment:
raise NotImplementedError(f"{api.name} API is not available in LOCAL environment")

api_class = getattr(importlib.import_module(api.value.module_name), api.value.class_name)
self._api_cache[api] = api_class(self._service_base_url, **self._api_args)

[docs]
class APIFactory:
"""
Factory that is exposed to accept request for API creation.
"""

self.api_args = {"platform_config": platform_config, "application_config": application_config, "auth": auth, "proxies": proxies, "local_environment": local_environment,}

self.platform_factory: PlatformFactory = _PlatformFactory(
lookup_api_registry=lookup_api_registry, **self.api_args # type: ignore
)
self.catalog_factories: Dict[Union[str, Optional[str]], CatalogApiFactory] =
self.service_factories: Dict[Union[str, Optional[str]], ServiceFactory] =
self.map_matching_factories: Dict[Union[str, Optional[str]], MapMatchingFactory] =

[docs]
def remove_catalog_apis(self, catalog_hrn: str) -> None:
"""
Removes Factory of APIs against :param:catalog_hrn.

:param catalog_hrn: HRN of the catalog.
"""
if catalog_hrn in self._catalog_factories:
del self._catalog_factories[catalog_hrn]

[docs]
def get_api(
self,
api: API,
catalog_hrn: Optional[str] = None,
service_base_url: Optional[str] = None,
map_matching_base_url: Optional[str] = None,
) -> Any:
"""
Gets API against :param:api.

:param api: Name of the API to be fetched.
:param catalog_hrn: (Optional) HRN of the catalog for which
API need to be fetched.
:param service_base_url: (Optional) Base URL of the service
for whichAPI need to be fetched.
:param map_matching_base_url: Map matching base url
:return: API instance against the given :param:api.
:raises ValueError: If an invalid API value passed.
"""
if api in [
API.BASE,
API.DATA_CONFIG,
API.LOOKUP,
API.ARTIFACT,
API.AAA_AUTH,
API.SERVICE_REGISTRY,
]:
return self._platform_factory.get_api(api)
elif api in [
API.DATA_BLOB,
API.DATA_VOLATILE_BLOB,
API.DATA_METADATA,
API.DATA_PUBLISH,
API.DATA_QUERY,
API.DATA_INDEX,
API.DATA_STREAM,
API.DATA_INGEST,
API.DATA_INTERACTIVE,
API.DATA_OBJECT_BLOB,
API.DATA_STATISTICS,
]:
if catalog_hrn not in self._catalog_factories:
self.catalog_factories[catalog_hrn] = CatalogApiFactory(
catalog_hrn=catalog_hrn, # type: ignore
platform_factory=self._platform_factory,
**self.api_args, # type: ignore
)
return self._catalog_factories[catalog_hrn].get_api(api)
elif api in [API.SERVICE]:
if service_base_url not in self._service_factories:
self.service_factories[service_base_url] = ServiceFactory(
service_base_url=service_base_url, # type: ignore
platform_factory=self._platform_factory,
**self.api_args, # type: ignore
)
return self._service_factories[service_base_url].get_api(api)
elif api in [API.MAP_MATCHING]:
if map_matching_base_url not in self._map_matching_factories:
self.map_matching_factories[map_matching_base_url] = MapMatchingFactory(
map_matching_base_url=map_matching_base_url, # type: ignore
platform_factory=self._platform_factory,
**self.api_args, # type: ignore
)
return self._map_matching_factories[map_matching_base_url].get_api(api)
else:
raise ValueError("Invalid API value passed")