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.

"""HERE platform module
"""
import json
import logging
import time
from typing import Any, Dict, Iterator, List, Optional, cast

from here.platform.adapter import Adapter
from here.platform.adapter_default import DefaultAdapter
from here.platform.api.aaa_authorization_api import AAAAuthorizationApi
from here.platform.api.aaa_oauth2_api import AAAOauth2Api
from here.platform.api.artifact_api import AccessType, ArtifactApi
from here.platform.api.base_api import BaseApi
from here.platform.api.data_config_api import DataConfigApi
from here.platform.api.factory.apifactory import API, APIFactory
from here.platform.api.lookup_api import LookupApi
from here.platform.api.map_matching_api import MapMatchingApi
from here.platform.api.registry import LookupApiRegistry
from here.platform.api.service_registry_api import ServiceRegistryApi
from here.platform.artifact import Artifact, ArtifactConfiguration
from here.platform.auth import Auth
from here.platform.catalog import Catalog
from here.platform.config.application_config import ApplicationConfig
from here.platform.config.platform_config import PlatformConfig
from here.platform.credentials import CredentialsType, PlatformCredentials
from here.platform.environment import Environment
from here.platform.exceptions import AuthenticationException, PlatformException
from here.platform.layer import Layer
from here.platform.map_matcher import MapMatcher
from here.platform.model.subscription import (
InteractiveMapSubscription,
InteractiveMapSubscriptionStatus,
InteractiveMapSubscriptionType,
InteractiveMapUnsubscribe,
)
from here.platform.project import Project
from here.platform.schema import SchemaRegistry
from here.platform.schema_api import Schema, SchemaConfiguration
from here.platform.service import Service, ServiceConfiguration

logger = logging.getLogger(name)

[docs]
class Platform:
"""
This class is responsible for interacting with the HERE platform.

It requires PlatformCredentials, ApplicationConfig, Environment and proxies configuration.
If not provided, there are defined using default values or standard configuration files.
"""

def init(
self,
credentials: Optional[PlatformCredentials] = None,
application_config: Optional[ApplicationConfig] = None,
environment: Optional[Environment] = None,
proxies: Optional[dict] = None,
billing_tag: Optional[str] = None,
adapter: Optional[Adapter] = None,
project_hrn: Optional[str] = None,
platform_config: Optional[PlatformConfig] = None,
**kwargs,
):
"""
Instantiate Platform.

:param credentials: an instance of PlatformCredentials
:param application_config: an instance of ApplicationConfig
:param environment: an instance of Environment
:param proxies: dictionary with proxy configuration per protocol,
e.g. {"http": "http://...", "https": "https://..."},
None to obtain proxy configuration from environment variables.
:param adapter: the Adapter to transform data in different representation.
A default one is provided in case None is specified.
:param billing_tag: A string to represent a grouping of billing records.
:param project_hrn: A string representing a project.
:param platform_config: The properties of Platform Configuration.
:param **kwargs: Arbitrary keyword arguments passed to underlying function.

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

self._environment = environment or Environment.from_default()
self._local_environment = self._environment == Environment.LOCAL
self.credentials: Optional[PlatformCredentials] = None
if not self._local_environment:

Get credentials only when non-local environment

self.credentials = credentials or PlatformCredentials.from_default()
self.application_config = application_config or ApplicationConfig.from_default()
self.application_config.additional_parameters = kwargs.copy()
self._platform_config = platform_config or PlatformConfig.from_environment(
self._environment
)
self._polling_wait = (
0.1 if self._local_environment else self.application_config.polling_wait
)
self._retry_max_time = (
5 if self._local_environment else self.application_config.retry_max_time
)

self.proxies = proxies

self.default_adapter: Adapter = DefaultAdapter()
self.adapter: Adapter = adapter or self.default_adapter
self.lookup_api_registry = LookupApiRegistry()
self.billing_tag = billing_tag
self._project_hrn = project_hrn

if self._platform_config.account_url is None:
self.auth = None
else:
if (
self.credentials is not None
and self.credentials.cred_type == CredentialsType.Credentials
):
aaa_oauth2_api = AAAOauth2Api(
base_url=self.credentials.cred_properties["endpoint"],
platform_config=self.platform_config,
application_config=self.application_config,
proxies=proxies,
)
self.auth = Auth(self.credentials, aaa_oauth2_api=aaa_oauth2_api)
elif (
self.credentials is not None
and self.credentials.cred_type == CredentialsType.Token
):
assert self.credentials
self.auth = Auth(self.credentials)
else:

Unexpected credentials type

assert False

if self._project_hrn and self.auth is not None:
self.auth.set_scope(self._project_hrn)

self.api_factory: APIFactory = APIFactory(
platform_config=self.platform_config,
application_config=self.application_config,
auth=cast(Auth, self.auth),
proxies=cast(dict, self.proxies),
local_environment=self._local_environment,
lookup_api_registry=self.lookup_api_registry,
)

self._schema_reg: SchemaRegistry = None # type: ignore

@property
def base_api(self) -> BaseApi:
"""
Lazy loads the service BaseApi.

:return: BaseApi instance.
"""
return cast(BaseApi, self.api_factory.get_api(API.BASE))

@property
def data_config_api(self) -> DataConfigApi:
"""
Lazy loads the service DataConfig API.

:return: DataConfigApi instance.
"""
return cast(DataConfigApi, self.api_factory.get_api(API.DATA_CONFIG))

@property
def lookup_api(self) -> LookupApi:
"""
Lazy loads the service Lookup API.

:return: LookupApi instance.
"""
return cast(LookupApi, self.api_factory.get_api(API.LOOKUP))

@property
def artifact_api(self) -> ArtifactApi:
"""
Lazy loads the service Artifact API.

:return: ArtifactApi instance.
"""
return cast(ArtifactApi, self.api_factory.get_api(API.ARTIFACT))

@property
def schema_registry(self) -> SchemaRegistry:
"""
Lazy loads the service SchemaRegistry API.

:return: singleton SchemaRegistry instance.
"""
if self._schema_reg is None:
self._schema_reg = SchemaRegistry(
self.platform_config, self.api_factory.get_api(API.ARTIFACT)
)
return self._schema_reg

@property
def aaa_auth_api(self) -> AAAAuthorizationApi:
"""
Lazy loads the service AAAAuthorization API.

This API is not implemented in the Local Data Service.

:return: AAAAuthorization instance.
"""
return cast(AAAAuthorizationApi, self.api_factory.get_api(API.AAA_AUTH))

@property
def service_registry_api(self) -> ServiceRegistryApi:
"""
Lazy loads the service registry API.

This API is not implemented in the Local Data Service.

:return: ServiceRegistryApi instance.
"""
return cast(ServiceRegistryApi, self.api_factory.get_api(API.SERVICE_REGISTRY))

@property
def map_matching_api(self) -> MapMatchingApi:
"""
Lazy loads the service registry API.

This API is not implemented in the Local Data Service.

:return: ServiceRegistryApi instance.
"""
return cast(MapMatchingApi, self.api_factory.get_api(API.MAP_MATCHING))

@property
def environment(self) -> Environment:
"""
Return the platform environment.

:return: environment
:rtype: Environment
"""
return self._environment

@environment.setter
def environment(self, environment):
"""
Set the platform environment and platform config based on environment.

:param environment: environment
"""
self._environment = environment
self._platform_config = PlatformConfig.from_environment(environment)

@property
def platform_config(self) -> PlatformConfig:
"""
Return the platform config.

:return: Platform configuration
:rtype: PlatformConfig
"""
return self._platform_config

[docs]
def get_status(self, as_type="json", event_type="incident"):
"""
Get platform status

:param as_type: format of the response
:param event_type: type of event
:return: Platform status in the desired format
:raises ValueError: Incorrect parameter value
"""
platform_status_url = self.platform_config.platform_status_url
if platform_status_url is None:
raise ValueError("platform_status_url is not present in configuration.")
if as_type not in ["json", "xml"]:
msg = 'Parameter as_type must be "xml" or "json" (the default).'
raise ValueError(msg)
params = {"format": as_type, "event": event_type}
resp = self.base_api.get(url=platform_status_url, params=params)
if as_type == "xml":
status = resp.text
else:
status = resp.json()
return status

[docs]
def get_catalog(
self, hrn: str, adapter: Optional[Adapter] = None, billing_tag: Optional[str] = None
) -> Catalog:
"""
Return a :class:Catalog object for the given HRN.

:param hrn: a string representing a HERE Resource Name.
:param adapter: the Adapter to transform data between different representations.
The platform adapter is used case None is specified.
:param billing_tag: A string to represent a grouping of billing records.
If None, platform billing tag will be used,if present.

:return: :class:Catalog object
:raises ValueError: if the catalog does not exist or credentials
don't provide access to it

Usage::

from here.platform import Platform
platform = Platform()
platform.get_catalog("hrn:here:data::olp-here:oma-3") #doctest: +ELLIPSIS
<here.platform.catalog.Catalog object at 0x...>

"""
billing_tag = billing_tag or self.billing_tag
if self.data_config_api.catalog_exists(catalog_hrn=hrn):
return Catalog(hrn=hrn, platform=self, adapter=adapter, billing_tag=billing_tag)
else:
raise ValueError(
f"Catalog {hrn} does not exist or credentials don't provide access to it"
)

[docs]
def catalog_exists(self, hrn: str) -> bool:
"""
Check whether a catalog with the specified HRN exists.

:param hrn: a HERE Resource Name
:return: a boolean value indicating if a catalog exists
"""
return self.data_config_api.catalog_exists(catalog_hrn=hrn, billing_tag=self.billing_tag)

[docs]
def create_catalog(
self,
id: str,
name: str,
summary: str,
description: str,
billing_tag: Optional[str] = None,
**details,
) -> Catalog:
"""
Create a catalog and return a :class:Catalog object for the given metadata.

:param id: An identifier unique within the realm, used to construct the catalog HRN
:param name: The short name for the catalog
:param summary: A one-sentence summary of the catalog
:param description: A detailed description of the catalog and what it contains
:param details: Optional catalog details - tags, layers, version, notifications and replication.
See the documentation for the catalog configuration for more informations
:param billing_tag: A string to represent a grouping of billing records.
If None, platform billing tag will be used,if present.
:return: a :class:Catalog object

Usage::

from here.platform import Platform
platform = Platform()
catalog = platform.create_catalog(id="create-catalog", name="name", # doctest: +SKIP
summary="summary", description="description", tags=["tag1", "tag2"]) # doctest: +SKIP # noqa: E501
"""
data = details.copy()
data["id"] = id
data["name"] = name
data["summary"] = summary
data["description"] = description

response = self.data_config_api.create_catalog(data, billing_tag=self.billing_tag)
status_response, complete = self.data_config_api.get_catalog_status(
response["href"], self.billing_tag
)
while not complete:
logging.debug(f"status polling wait {self._polling_wait} sec.")
time.sleep(self._polling_wait)
status_response, complete = self.data_config_api.get_catalog_status(
response["href"], self.billing_tag
)
hrn = status_response.get("hrn")
if not hrn:

LocalDataServer returns a different response than does Platform

hrn = status_response["item"]["href"].rsplit("/", 1)[-1]
return Catalog(hrn, platform=self, billing_tag=billing_tag)

[docs]
def list_catalogs(self, **filters: str) -> List[Catalog]:
"""
List all the catalogs accessible on the HERE platform.

Optionally, search and return only catalogs specified by some filter criteria.
It does not return catalog that credentials don't provide access to.

:param filters: keywords to search for
:return: a list of :class:Catalog objects

Usage::

from here.platform import Platform
platform = Platform()
platform.list_catalogs(coverage="CN") # doctest: +ELLIPSIS
[<here.platform.catalog.Catalog object at ...]
"""
catalogs_info = self.data_config_api.get_catalogs(**filters)
try:
catalog_items = catalogs_info["results"]["items"]
except KeyError:
catalog_items = []
catalogs = []
for cat in catalog_items:
try:
catalogs.append(Catalog(cat["hrn"], platform=self))
except AuthenticationException:
logger.warning("Unable to access HRN . Skipping.".format(cat["hrn"]))
except PlatformException as ex:
logger.warning(
"Unable to look up HRN due to . Skipping.".format(cat["hrn"], str(ex))
)
return catalogs

[docs]
def modify_catalog(self, hrn: str, **details):
"""
Modify the catalog details, including name, summary, description.

This replaces the complete details of the catalog.

After this call succeedes, obtain a new catalog via get_catalog
to have access to the modified details, including affected layers.

:param hrn: the HERE Resource Name of the catalog
:param details: Optional catalog details - tags, layers, version, notifications and replication.
See the documentation for the catalog configuration for more informations.

The documentation for the API endpoint used can be found here:
|modify_catalog|

.. |modify_catalog| raw:: html

Modify a Catalog # noqa E501

Usage::

from here.platform import Platform
platform = Platform()
platform.modify_catalog(hrn="hrn:here:data::olp-here:update", name='updated-name', description='updated-description') # doctest: +SKIP
"""
if not details:
return
response = self.data_config_api.patch_catalog(hrn, data=details)
status_response, complete = self.data_config_api.get_catalog_status(
response["href"], billing_tag=self.billing_tag
)
while not complete:
logging.debug(f"status polling wait {self._polling_wait} sec.")
time.sleep(self._polling_wait)
status_response, complete = self.data_config_api.get_catalog_status(
response["href"], billing_tag=self.billing_tag
)

[docs]
def update_catalog(
self,
hrn: str,
name: Optional[str] = None,
summary: Optional[str] = None,
description: Optional[str] = None,
new_layers: Optional[List[Dict]] = None,
**details,
):
"""
Update the catalog details.

This replaces the complete details of the catalog.

After this call succeedes, obtain a new catalog via get_catalog
to have access to the updated details, including affected layers.

:param hrn: the HERE Resource Name of the catalog
:param name: the optional new short name of the catalog.
:param summary: the optional new summary of the catalog.
:param description: the optional new detailed description of the catalog.
:param new_layers: the optional list of new layers - each new layer needs to be a dict
For the format please see the documentation for the catalog configuration.
Please note that layers can only be added to the catalog using this function.
:param details: Optional catalog details - tags, version, notifications and replication.
See the documentation for the catalog configuration for more information.

The documentation for the API endpoint used can be found here:
|update_catalog|

.. |update_catalog| raw:: html

Updates a Catalog # noqa E501

Usage::

from here.platform import Platform
platform = Platform()
new_layer = # put the configuration of a new layer here as a dict
platform.update_catalog(hrn='hrn:here:data::olp-here:update', name='updated', new_layers=[new_layer]) # doctest: +SKIP
"""
data = details.copy()
assert not ("layers" in data and new_layers is not None), (
"You can either define the new_layers or overwrite"
" the existing layers but not both at the same time."
)
catalog = self.get_catalog(hrn)
catalog_conf = catalog.configuration

the "id" field cannot be updated however it must be in the request body

data["id"] = catalog_conf.id
data["name"] = name if name is not None else catalog_conf.name
data["summary"] = summary if summary is not None else catalog_conf.summary
data["description"] = description if description is not None else catalog_conf.description
if "layers" not in data:
data["layers"] = [lyr.json for lyr in catalog_conf.layers]
if new_layers is not None:
data["layers"] += new_layers
response = self.data_config_api.update_catalog(
catalog_hrn=hrn, data=data, billing_tag=self.billing_tag
)
status_response, complete = self.data_config_api.get_catalog_status(
response["href"], billing_tag=self.billing_tag
)
while not complete:
logging.debug(f"status polling wait {self._polling_wait} sec.")
time.sleep(self._polling_wait)
status_response, complete = self.data_config_api.get_catalog_status(
response["href"], billing_tag=self.billing_tag
)

[docs]
def delete_catalog(self, hrn: str):
"""
Delete a catalog along with the layers it contains.

:param hrn: the HERE Resource Name of the catalog

Usage::

from here.platform import Platform
platform = Platform()
platform.delete_catalog("hrn:here:data::olp-here:delete") # doctest: +SKIP
"""
response = self.data_config_api.delete_catalog(hrn, self.billing_tag)
while True:
logging.debug(f"status polling wait {self._polling_wait} sec.")
time.sleep(self._polling_wait)
status_response, complete = self.data_config_api.get_catalog_status(
response["href"], billing_tag=self.billing_tag
)
status = status_response["status"]
logger.debug(f"Catalog delete: {hrn} state: {status}")
if complete:
self.api_factory.remove_catalog_apis(hrn)
logger.info(f"Catalog deletion for hrn: {hrn} finished with status: {status}")
return

[docs]
def get_project(self, hrn: str) -> Project:
"""
Return a :class:Project object for the given HRN.

:param hrn: a string representing a HERE Resource Name.

:return: the Project.

Usage::

from here.platform import Platform
platform = Platform()
platform.get_project("hrn:here:authorization::olp-here:project/test")# doctest: +SKIP
"""
return Project(hrn, platform=self)

[docs]
def create_project(self, project_id: str, project_name: str, project_desc: str) -> Project:
"""
Create a project.

:param project_id: A unique id for the project.
:param project_name: The short name for the project.
:param project_desc: A detailed description of the project and what it contains.
:return: a :class 'Project' object.

Usage::

from here.platform import Platform
platform = Platform()
project = platform.create_project(id="create-project", name="name",# doctest: +SKIP
description="description") # doctest: +SKIP
"""
payload = {"id": project_id, "name": project_name, "description": project_desc}
response = self.aaa_auth_api.create_project(body=payload)
return Project(response["hrn"], platform=self)

[docs]
def list_projects(
self,
limit: Optional[int] = None,
can_manage: Optional[bool] = None,
is_member: Optional[bool] = None

page_token: Optional[str] = None

) -> List[Project]:
"""
Get the list of projects that you are a project admin or a member based on the can_manage
and is_member.

:param limit: Number of entries to be returned in the response
:param can_manage: If true returns all projects of which the
caller(user/app) is a project admin
:param is_member: If true returns all projects of which the caller(user/app) is a member

:return: a list of projects.

Usage::

from here.platform import Platform
platform = Platform()
resp = platform.list_projects(limit=10) # doctest: +SKIP
"""
response = self.aaa_auth_api.get_project_list(
limit=str(limit) if limit else None,

page_token=page_token,

can_manage=can_manage,
is_member=is_member,
)

return [Project(p["hrn"], self) for p in response["items"]]

[docs]
def update_project(self, hrn: str, project_name: str, project_desc: str):
"""
Update the project metadata.

:param hrn: HRN identifying the project
:param project_name: The short name for the project.
:param project_desc: A detailed description of the project and what it contains.
"""
payload = {"name": project_name, "description": project_desc}
self.aaa_auth_api.patch_project(project=hrn, body=payload)

[docs]
def delete_project(self, hrn: str):
"""
Delete a project along with the catalogs it contains.

:param hrn: HRN identifying the project
"""
self.aaa_auth_api.delete_project(hrn)

[docs]
def leave_project(self, hrn: str):
"""
Remove the caller from the specified Project.

:param hrn: HRN identifying the project

Usage::

from here.platform import Platform
platform = Platform()
platform.leave_project("hrn:here:authorization::" # doctest: +SKIP
"olp-here:project/test") # doctest: +SKIP
"""
self.aaa_auth_api.leave_project(hrn)

[docs]
def list_layer_details(self, **filters: str) -> List[dict]:
"""
List all the layer details accessible on the HERE platform.

Optionally, search and return only layers specified by some filter criteria.
It does not return layers that credentials don't provide access to.

:param filters: keywords to search for
:return: list of layer details

Usage::

from here.platform import Platform
import pandas as pd
platform = Platform()
layers = platform.list_layer_details(coverage='US') # doctest: +ELLIPSIS
"""
filters["verbose"] = "True"
catalogs_info = self.data_config_api.get_catalogs(**filters)
catalogs = catalogs_info["results"]["items"]
layer_list = []
for cat in catalogs:
layers = cat["layers"]
try:
catalog = self.get_catalog(cat["hrn"])
except AuthenticationException:
logger.info(f"App doesn't have access to {cat['hrn']}.")
continue
for layer in layers:
layer = catalog.get_layer(layer_id=layer["id"])
layer_list.append(layer.get_details())
return layer_list

[docs]
def list_layers(self, **filters: str) -> List[Layer]:
"""
List all the layer details accessible on the HERE platform.

Optionally, search and return only layers specified by some filter criteria.
It does not return layers that credentials don't provide access to.

:param filters: keywords to search for
:return: dataframe with layer details

Usage::

from here.platform import Platform
import pandas as pd
platform = Platform()
layer_df = platform.list_layers(coverage='US') # doctest: +ELLIPSIS
"""
filters["verbose"] = "True"
catalogs_info = self.data_config_api.get_catalogs(**filters)
catalogs = catalogs_info["results"]["items"]
layer_list = []
for cat in catalogs:
layers = cat["layers"]
try:
catalog = self.get_catalog(cat["hrn"])
except AuthenticationException:
logger.info(f"App doesn't have access to {cat['hrn']}.")
continue
for layer in layers:
layer = catalog.get_layer(layer_id=layer["id"])
layer_list.append(layer)
return layer_list

[docs]
def list_services(self) -> List[Service]:
"""
List all the services accessible on the HERE platform.

:return: a list of :class:Service objects

Usage::

from here.platform import Platform
platform = Platform()
platform.list_services() # doctest: +ELLIPSIS
[<here.platform.service.Service object at ...]
"""
services_info = self.service_registry_api.get_services()
services = [
Service(
hrn=service_info["hrn"],
configuration=ServiceConfiguration.from_dict(json_dict=service_info),
platform=self,
)
for service_info in services_info
]
return services

[docs]
def get_service(self, hrn: str) -> Service:
"""
Return a :class:Service object for the given HRN.

:param hrn: a string representing a HERE Resource Name.
:return: :class:Service object

Usage::

from here.platform import Platform
platform = Platform()
platform.get_service("hrn:here:service::olp-here:routing-8") #doctest: +ELLIPSIS
<here.platform.service.Service object at 0x...>

"""
service_info = self.service_registry_api.get_service(service_hrn=hrn)
service_config = ServiceConfiguration.from_dict(json_dict=service_info)
return Service(hrn=hrn, configuration=service_config, platform=self)

[docs]
def clone_catalog(
self,
source: Catalog,
id: str,
name: Optional[str] = None,
summary: Optional[str] = None,
description: Optional[str] = None,
billing_tag: Optional[str] = None,
layers: Optional[List] = None,
):
"""
Create a catalog from a source catalog and return
a :class:Catalog object. Cloning will be restricted to configuration.
No data will be cloned or copied from source to new catalog.

:param source: Catalog object from which new catalog need to be cloned.
:param id: An identifier unique within the realm, used to construct the catalog HRN
:param name: The short name for the catalog
:param summary: A one-sentence summary of the catalog
:param description: A detailed description of the catalog and what it contains
:param billing_tag: A string to represent a grouping of billing records.
If None, platform billing tag will be used,if present.
:param layers: A list of layer ids to be added in clone catalog.
These layer ids must be present in source catalog.
If None, all layers present in source catalog will be copied.
:return: a :class:Catalog object

"""
source_details = source.get_details()
layer_list = []
if layers:
for layer in layers:
layer = source.get_layer(layer_id=layer)
layer_details = layer.get_details()
layer_list.append(layer_details)
cat = self.create_catalog(
id=id,
name=name if name else source_details["name"],
summary=summary if summary else source_details["summary"],
description=description if description else source_details["description"],
billing_tag=billing_tag,
layers=layer_list if layers else source_details["layers"],
)
return cat

[docs]
def map_matcher_map_versions(self) -> List[MapMatcher]:
"""
List all available map matcher map versions
"""
map_matching_info = self.map_matching_api.map_matcher_map_versions()
map_matcher_list = []
map_matcher_dict = json.loads(map_matching_info.replace("'", '"'))
for map_matcher in map_matcher_dict["versions"]:
map_matcher_obj = MapMatcher(
version=map_matcher["version"], hrn=map_matcher["hrn"], platform=self
)
map_matcher_list.append(map_matcher_obj)
return map_matcher_list

[docs]
def get_map_matcher_map_version(self, version: str = "latest") -> MapMatcher:
"""
Gets information for a specific map matcher map version
:param version: map version to use or latest to use the latest available version
:return: a :class:MapMatcher object
"""
map_version = self.map_matching_api.get_map_matcher_map_version(version)
map_matcher_dict = json.loads(map_version.replace("'", '"'))
return MapMatcher(
version=map_matcher_dict["version"], hrn=map_matcher_dict["hrn"], platform=self
)

[docs]
def list_artifacts(
self,
access: Optional[AccessType] = None,
sort: Optional[str] = None,
order: Optional[str] = None,
from_param: Optional[str] = None,
group_id: Optional[str] = None,
artifact_id: Optional[str] = None,
limit: Optional[int] = None,
offset: Optional[int] = None,
) -> List[Artifact]:
"""
List all the artifacts accessible on the HERE platform.

:param access: Comma separated list of access types (with OR semantic) to filter result.
If 'default', returns artifacts user has read access to.
If 'orgAdmin', returns artifacts
OrgAdmin user have access to.
:param sort: sort parameter
:param order: order of sorting. Available values : ASC, DESC
:param from_param: from parameter
:param group_id: Applies filtering based on groupId
:param artifact_id: Applies filtering based on artifactId
:param limit: limit number of records
:param offset: offset number of records
:return: response from the API.
"""

artifacts = self.artifact_api.list_artifacts(
access=access,
sort=sort,
order=order,
from_param=from_param,
group_id=group_id,
artifact_id=artifact_id,
limit=limit,
offset=offset,
)
artifacts_list = [
Artifact(
hrn=artifact["hrn"],
configuration=ArtifactConfiguration.from_dict(json_dict=artifact),
platform=self,
)
for artifact in artifacts["items"]
]
return artifacts_list

[docs]
def get_artifact(self, artifact_hrn: str) -> Artifact:
"""
Return the information about artifact (hrn, groupId, artifactId, version) and linked
files.

:param artifact_hrn: The HRN of the artifact.
:return: response from the API.
"""
artifact_details = self.artifact_api.get_artifact(artifact_hrn=artifact_hrn)
return Artifact(
hrn=artifact_details["artifact"]["hrn"],
configuration=ArtifactConfiguration.from_dict(json_dict=artifact_details["artifact"]),
platform=self,
)

[docs]
def delete_artifact(self, artifact_hrn: str, force: Optional[bool] = None) -> bool:
"""
Delete the artifact and related files by given HRN.

:param artifact_hrn: The HRN of the schema.
:param force: The flag to force the deletion even if the artifact
is linked to a project. By default force flag is set to false.
:return: response from the API.
"""
return self.artifact_api.delete_artifact(artifact_hrn=artifact_hrn)

[docs]
def list_schemas(
self,
access: Optional[str] = None,
sort: Optional[str] = None,
order: Optional[str] = None,
from_param: Optional[str] = None,
group_id: Optional[str] = None,
artifact_id: Optional[str] = None,
limit: Optional[int] = None,
offset: Optional[int] = None,
) -> List[Schema]:
"""
Return the list of available schemas.

:param access: Comma separated list of access types (with OR semantic) to filter result.
If 'default', returns schema user have read access to. If 'orgAdmin', returns catalogs
OrgAdmin user have access to.
:param sort: sort parameter
:param order: order of sorting
:param from_param: from parameter
:param group_id: Applies filtering based on groupId
:param artifact_id: Applies filtering based on artifactId
:param limit: limit number of records
:param offset: offset number of records
:return: response from the API.
"""
schemas = self.artifact_api.list_schemas(
access=access,
sort=sort,
order=order,
from_param=from_param,
group_id=group_id,
artifact_id=artifact_id,
limit=limit,
offset=offset,
)

schemas_list = [
Schema(
hrn=schema["hrn"],
configuration=SchemaConfiguration.from_dict(json_dict=schema),
platform=self,
)
for schema in schemas["items"]
]
return schemas_list

[docs]
def get_schema(self, schema_hrn: str) -> Schema:
"""
Return the information about schema (hrn, groupId, artifactId, version) and related
artifact and variants for the given HRN.

:param schema_hrn: The HRN of the schema.
:return: response from the API.
"""
schema_details = self.artifact_api.get_schema(schema_hrn=schema_hrn)
return Schema(
hrn=schema_details["schema"]["hrn"],
configuration=SchemaConfiguration.from_dict(json_dict=schema_details["schema"]),
platform=self,
)

[docs]
def delete_schema(self, schema_hrn: str) -> bool:
"""
Delete the schema and related artifacts by given HRN.

:param schema_hrn: The HRN of the schema.
:return: response from the API.
"""
return self.artifact_api.delete_schema(schema_hrn=schema_hrn)

[docs]
def get_all_subscriptions(
self, limit: Optional[int] = None
) -> Iterator[InteractiveMapSubscription]:
"""
Lists all subscriptions that your account has access to.

:param limit: number of records to limit per fetch.
:yields: A :List of InteractiveMapSubscription class object
"""

page_token = ""

while True:
filters: Dict[Any, Any] = dict()
if limit:
filters["limit"] = limit

if page_token:
filters["pageToken"] = page_token

subscriptions = self.data_config_api.list_subscriptions(**filters)

for item in subscriptions["items"]:
yield InteractiveMapSubscription(item)

page_token = subscriptions["nextPageToken"] if "nextPageToken" in subscriptions else ""

if not page_token:
break

[docs]
def subscription_exists(self, subscription_hrn: str) -> bool:
"""
Checks whether a subscription with the specified HRN exists.

:param subscription_hrn: The HERE Resource Name (HRN) of subscription
:return: True if subscription exists.
"""

return self.data_config_api.check_subscription_exists(subscription_hrn=subscription_hrn)

[docs]
def get_subscription(self, subscription_hrn: str) -> InteractiveMapSubscription:
"""
To get configuration of the subscription associated with the HRN.

:param subscription_hrn: The HERE Resource Name (HRN) of subscription
:return: InteractiveMapSubscription object.
"""

subscription_resp = self.data_config_api.get_subscription(
subscription_hrn=subscription_hrn
)
logger.debug(f"Get Subscription Response : {subscription_resp}")
return InteractiveMapSubscription(subscription_resp)

[docs]
def subscribe(
self,
subscription_name: str,
description: str,
source_catalog_hrn: str,
source_layer: str,
destination_catalog_hrn: str,
destination_layer_id: str,
interactive_map_subscription_type: InteractiveMapSubscriptionType,
) -> InteractiveMapSubscription:
"""
Method to Subscribe to a Stream Layer from Layer's Catalog HRN.
Source Layer is the current layer and Source Catalog is Layer's Catalog which it belongs.

:param subscription_name: Name of the subscription.
:param description: Description of the subscription.
:param source_catalog_hrn: Catalog HRN of the source Catalog.
:param source_layer: Layer id of the source Interactive Map Layer.
:param destination_catalog_hrn: Catalog HRN of the destination Catalog.
:param destination_layer_id: Layer id of the destination Stream Layer.
:param interactive_map_subscription_type: InteractiveMapSubscriptionType containing type of
subscription.
:raises KeyError: in case statusToken in Response of createSubscription.
:raises ValueError: in case Created Subscription Status is NOT Active after
multiple retry till max retry time.
:return: InteractiveMapSubscription object containing details of the created subscription.
"""

configuration_object: dict = dict(
{"subscriptionName": subscription_name, "description": description, "sourceCatalog": source_catalog_hrn, "sourceLayer": source_layer, "destinationCatalog": destination_catalog_hrn, "destinationLayer": destination_layer_id, "interactiveMapSubscription": {"type": interactive_map_subscription_type.value},}
)

subscription_status_resp = dict({"status": ""})
start_time = time.time()

create_subscription_resp: dict = self.data_config_api.create_subscription(
configuration_object
)

logger.debug(f"Create Subscription Response : {create_subscription_resp}")

if "statusToken" not in create_subscription_resp:
raise KeyError("Missing statusToken in Response of createSubscription.")

time.sleep(min(0.1, self._polling_wait))

while time.time() - start_time < self._retry_max_time:
subscription_status_resp = self.data_config_api.get_subscription_status(
status_token=create_subscription_resp["statusToken"]
)
logger.debug(f"Subscription Status Response : {subscription_status_resp}")

if subscription_status_resp["status"] == "active":
break

time.sleep(self._polling_wait)

logger.debug(
f"Total wait time for Subscription Status Response : {time.time() - start_time}"
)

if "status" in subscription_status_resp and subscription_status_resp["status"] != "active":
raise ValueError(
f"Created Subscription Status is NOT Active "
f"for statusToken {create_subscription_resp['statusToken']}."
)

subscription_resp: dict = self.data_config_api.get_subscription(
subscription_hrn=subscription_status_resp["subscriptionHrn"]
)

subscription_resp.update(create_subscription_resp)

return InteractiveMapSubscription(subscription_resp)

[docs]
def subscription_status(self, status_token: str) -> InteractiveMapSubscriptionStatus:
"""
Get the status of the subscription.

:param status_token: Status token from create/delete subscription response.
:return: InteractiveMapSubscriptionStatus object containing status.
"""

subscription_status_resp = self.data_config_api.get_subscription_status(
status_token=status_token
)
logger.debug(f"Subscription Status Response : {subscription_status_resp}")
return InteractiveMapSubscriptionStatus(subscription_status_resp)

[docs]
def unsubscribe(self, subscription_hrn: str) -> InteractiveMapUnsubscribe:
"""
Deletes a subscription associated with the HRN.

:param subscription_hrn: The HERE Resource Name (HRN) of subscription.
:return: InteractiveMapUnsubscribe object containing details.
"""
subscription_status = dict({"status": ""})
start_time = time.time()

delete_subscription_resp: dict = self.data_config_api.delete_subscription(subscription_hrn)

logger.debug(f"Delete Subscription response: {delete_subscription_resp}")
time.sleep(min(0.1, self._polling_wait))

while time.time() - start_time < self._retry_max_time:
subscription_status = self.data_config_api.get_subscription_status(
status_token=delete_subscription_resp["statusToken"]
)

logger.debug(f"Delete Subscription Status response: {subscription_status}")

if subscription_status["status"] == "deleted":
break

time.sleep(self._polling_wait)

delete_subscription_resp.update(subscription_status)

return InteractiveMapUnsubscribe(delete_subscription_resp)