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:ArtifactApi class to perform Artifact API operations.
"""

from enum import Enum
from typing import Optional
from urllib import parse

from here.platform.api.base_api import BaseApi
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.exceptions import PlatformException

[docs]
class AccessType(Enum):
"""
Access types enum used to list artifacts and schemas
"""

default = 0
orgAdmin = 1

[docs]
class ArtifactApi(BaseApi):
"""
This class provides access to HERE platform Artifact APIs.

The artifact service provides a way to store/consume schemas and artifacts. An artifact is
the package which can contain code, assets, binary files, and configuration data. Artifacts
are uniquely named, versioned, and immutable. Each schema and artifact are identified by group
ID, artifact ID and version.
"""

def init(
self,
base_url: str,
platform_config: PlatformConfig,
application_config: ApplicationConfig,
auth: Optional[Auth],
proxies: Optional[dict] = None,
):
"""
Instantiate ArtifactApi object.

:param base_url: base url
: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.
"""
super(ArtifactApi, self).init(
platform_config=platform_config,
application_config=application_config,
auth=auth,
proxies=proxies,
)
self.base_url = base_url

"""
Schema Operations
"""

[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,
) -> dict:
"""
Return the list of available schemas. If scoped token is used, then the list of schemas is
returned within the project the token belongs to. If unscoped token is used, then the list
of schemas is returned including the schemas belonging to the projects.

: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.
:raises PlatformException: If platform responds with an HTTP error.
"""
path = "/schema"
url = self.format_url(self.base_url, path)
params = {"access": access, "sort": sort, "order": order, "from": from_param, "groupId": group_id, "artifactId": artifact_id, "limit": limit, "offset": offset,}
resp = self.get(url, params=params)
if resp.status_code == 200:
resp_json: dict = resp.json()
return resp_json
else:
raise PlatformException(resp)

[docs]
def get_schema(self, schema_hrn: str) -> dict:
"""
Return the information about schema (hrn, groupId, artifactId, version) and related
artifact and variants for the given HRN. If scoped token is used, then the information
about schema is returned within the project the token belongs to.

:param schema_hrn: The HRN of the schema.
:return: response from the API.
:raises PlatformException: If platform responds with an HTTP error.
"""
path = f"/schema/{schema_hrn}"
url = self.format_url(self.base_url, path)
resp = self.get(url)
if resp.status_code == 200:
resp_json: dict = resp.json()
return resp_json
else:
raise PlatformException(resp)

[docs]
def delete_schema(self, schema_hrn: str) -> bool:
"""
Delete the schema and related artifacts by given HRN. If scoped token is used, then the
schema is deleted within the project the token belongs to.

:param schema_hrn: The HRN of the schema.
:return: response from the API.
:raises PlatformException: If platform responds with an HTTP error.
"""
path = f"/schema/{schema_hrn}"
url = self.format_url(self.base_url, path)
resp = self.delete(url)
if resp.status_code == 200:
return True
else:
raise PlatformException(resp)

[docs]
def get_schema_document(self, schema_hrn: str, file: str) -> str:
"""
Return the url to schema documentation.

:param schema_hrn: The HRN of the schema.
:param file: The path to schema file.
:return: response from the API.
:raises PlatformException: If platform responds with an HTTP error.
"""
path = f"/schema/{schema_hrn}/doc/{file}"
url = self.format_url(self.base_url, path)
resp = self.get(url)
if resp.status_code == 200:
return resp.text
else:
raise PlatformException(resp)

[docs]
def update_schema_permission(self, schema_hrn: str, body: dict) -> bool:
"""
Grant/revoke permissions for schema. READ grants 'readResource',
MODIFY grants 'modifyResource', SHARE grants 'shareResource'.

:param schema_hrn: The HRN of the schema.
:param body: a dictionary with permissions.
:return: response from the API.
:raises PlatformException: If platform responds with an HTTP error.
"""
path = f"/schema/{schema_hrn}/permission"
url = self.format_url(self.base_url, path)
resp = self.post(url, data=body)
if resp.status_code == 200:
return True
else:
raise PlatformException(resp)

[docs]
def link_schema_to_project(self, schema_hrn: str, body: dict) -> dict:
"""
Link schema to project request.

:param schema_hrn: The HRN of the Schema.
:param body: a dictionary with project HRN.
:return: response from the API.
:raises PlatformException: If platform responds with an HTTP error.
"""
path = f"/schema/{schema_hrn}/links"
url = self.format_url(self.base_url, path)
resp = self.post(url, data=body)
if resp.status_code == 200:
resp_json: dict = resp.json()
return resp_json
else:
raise PlatformException(resp)

[docs]
def make_schema_linkable(self, schema_hrn: str, body: dict) -> dict:
"""
Make schema and all attached artifacts available for linking to
a specific project or realm.

:param schema_hrn: The HRN of the Schema.
:param body: a dictionary with project or realm Hrn.
:return: response from the API.
:raises PlatformException: If platform responds with an HTTP error.
"""
path = f"/schema/{schema_hrn}/linkable"
url = self.format_url(self.base_url, path)
print(url)
resp = self.post(url, data=body)
if resp.status_code == 200:
resp_json: dict = resp.json()
return resp_json
else:
raise PlatformException(resp)

[docs]
def remove_schema_linkability(self, schema_hrn: str, target_hrn: str) -> dict:
"""
Makes schema and all attached artifacts not available for linking to a project or realm.

:param schema_hrn: The HRN of the schema.
:param target_hrn: The HRN of the target. Project HRN from which schema is to be unlinked
:return: response from the API.
:raises PlatformException: If platform responds with an HTTP error.
"""
path = f"/schema/{schema_hrn}/linkable/{parse.quote(target_hrn, safe=':')}"
url = self.format_url(self.base_url, path)
resp = self.delete(url)
if resp.status_code == 200:
resp_json: dict = resp.json()
return resp_json
else:
raise PlatformException(resp)

[docs]
def unlink_schema_from_project(self, schema_hrn: str, project_hrn: str) -> dict:
"""
Unlink schema and all attached Artifacts from the project it was previously linked. This
API works for tokens with or without a project scope.

:param schema_hrn: The HRN of the schema.
:param project_hrn: The HRN of the project.
:return: response from the API.
:raises PlatformException: If platform responds with an HTTP error.
"""
path = f"/schema/{schema_hrn}/links/{parse.quote(project_hrn, safe=':')}"
url = self.format_url(self.base_url, path)
resp = self.delete(url)
if resp.status_code == 200:
resp_json: dict = resp.json()
return resp_json
else:
raise PlatformException(resp)

[docs]
def get_schema_main_message(self, schema_hrn: str) -> bytes:
"""
Return the binary data of main message file from the schema.

:param schema_hrn: The HRN of the Schema.
:return: response from the API.
:raises PlatformException: If platform responds with an HTTP error.
"""
path = f"/schema/{schema_hrn}/message"
url = self.format_url(self.base_url, path)
resp = self.get(url)
if resp.status_code == 200:
return resp.content
else:
raise PlatformException(resp)

[docs]
def add_schema_to_resource_policy(self, schema_hrn: str, body: dict) -> dict:
"""
Add schema and all attached artifacts to a newly created resource policy. This API works
for tokens with or without a project scope. When the access token is project scoped, all
the resources added to a resource policy must either be homed or referenced in the project.

:param schema_hrn: The HRN of the Schema.
:param body: a dictionary with resource policy.
:return: response from the API.
:raises PlatformException: If platform responds with an HTTP error.
"""
path = f"/schema/{schema_hrn}/resourcePolicy"
url = self.format_url(self.base_url, path)
resp = self.post(url, data=body)
if resp.status_code == 202:
resp_json: dict = resp.json()
return resp_json
else:
raise PlatformException(resp)

"""
Artifact Operations
"""

[docs]
def register_artifact(self, artifact_id: str, group_id: str, body: dict) -> dict:
"""
Requires 'accessApi' permission. Create artifact with version 'NONE'. If scoped token
is used then the artifact is registered within the project the token belongs to otherwise
grant 'readResource', 'modifyResource' and 'shareResource' permissions to app which send
request and provided user id

:param group_id: The groupId of the artifact.
:param artifact_id: The ID of the artifact.
:param body: Contains the data you want to add or remove from the UserID
:return: response from the API.
:raises PlatformException: If platform responds with an HTTP error.
"""
path = f"/artifact/register/{group_id}/{artifact_id}"
url = self.format_url(self.base_url, path)
resp = self.put(url, data=body)
if resp.status_code in (200, 201):
resp_json: dict = resp.json()
return resp_json
else:
raise PlatformException(resp)

[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,
) -> dict:
"""
Return the list of available artifacts. Requires permission to 'apiAccess'.If scoped token
is used, then the list of artifacts is returned within the project the token belongs to.
If unscoped token is used, then the list of artifacts is returned including the artifacts
belonging to the projects.

: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.
:raises PlatformException: If platform responds with an HTTP error.
"""
path = "/artifact"
url = self.format_url(self.base_url, path)
params = {"access": access, "sort": sort, "order": order, "from": from_param, "groupId": group_id, "artifactId": artifact_id, "limit": limit, "offset": offset,}
resp = self.get(url, params=params)
if resp.status_code == 200:
resp_json: dict = resp.json()
return resp_json
else:
raise PlatformException(resp)

[docs]
def get_artifact_registration(self, artifact_id: str, group_id: str) -> dict:
"""
Return the information about registered artifact. Requires 'accessApi' permission and
'readResource' permission to specified artifact

:param group_id: The groupId of the artifact.
:param artifact_id: The ID of the artifact.
:return: response from the API.
:raises PlatformException: If platform responds with an HTTP error.
"""
path = f"/artifact/register/{group_id}/{artifact_id}"
url = self.format_url(self.base_url, path)
resp = self.get(url)
if resp.status_code == 200:
resp_json: dict = resp.json()
return resp_json
else:
raise PlatformException(resp)

[docs]
def get_artifact(self, artifact_hrn: str) -> dict:
"""
Return the information about artifact (hrn, groupId, artifactId, version) and linked
files. If scoped token is used, then the information about artifact is returned within
the project the token belongs to.

:param artifact_hrn: The HRN of the artifact.
:return: response from the API.
:raises PlatformException: If platform responds with an HTTP error.
"""
path = f"/artifact/{artifact_hrn}"
url = self.format_url(self.base_url, path)
resp = self.get(url)
if resp.status_code == 200:
resp_json: dict = resp.json()
return resp_json
else:
raise PlatformException(resp)

[docs]
def get_artifact_file(self, artifact_hrn: str, file_name: str) -> bytes:
"""
Return the binary data of file linked to artifact.

:param artifact_hrn: The HRN of the artifact
:param file_name: The path to artifact file
:return: response from the API.
:raises PlatformException: If platform responds with an HTTP error.
"""
path = f"/artifact/{artifact_hrn}/{file_name}"
url = self.format_url(self.base_url, path)
resp = self.get(url)
if resp.status_code == 200:
return resp.content
else:
raise PlatformException(resp)

[docs]
def put_artifact_file(self, artifact_hrn: str, file_name: str, body: bytes) -> bool:
"""
Upload the file and linked to artifact. Requires permission to 'modifyResource'
for the given HRN.

:param artifact_hrn: The HRN of the artifact.
:param body: Data to be uploaded
:param file_name: The path to artifact file
:return: response from the API.
:raises PlatformException: If platform responds with an HTTP error.
"""
path = f"/artifact/{artifact_hrn}/{file_name}"
url = self.format_url(self.base_url, path)
resp = self.put(url, data=body)
if resp.status_code == 201:
return True
else:
raise PlatformException(resp)

[docs]
def delete_artifact_file(self, artifact_hrn: str, file_name: str) -> bool:
"""
Delete file from the artifact. for the given HRN

:param artifact_hrn: The HRN of the artifact
:param file_name: The path to artifact file
:return: response from the API.
:raises PlatformException: If platform responds with an HTTP error.
"""
path = f"/artifact/{artifact_hrn}/{file_name}"
url = self.format_url(self.base_url, path)
resp = self.delete(url)
if resp.status_code == 200:
return True
else:
raise PlatformException(resp)

[docs]
def get_artifact_file_from_archive(
self, artifact_hrn: str, archive_file: str, file_in_archive: str
) -> bytes:
"""
Returns the binary data of a file stored in archive.

:param artifact_hrn: The HRN of the schema.
:param archive_file: The name of artifact archive. Supported archive types are zip and jar
:param file_in_archive: Relative path of file in archive
:return: response from the API.
:raises PlatformException: If platform responds with an HTTP error.
"""
path = f"/artifact/{artifact_hrn}/{archive_file}/files/{file_in_archive}"
url = self.format_url(self.base_url, path)
resp = self.get(url)
if resp.status_code == 200:
return resp.content
else:
raise PlatformException(resp)

[docs]
def delete_artifact(self, artifact_hrn: str, force: Optional[bool] = None) -> bool:
"""
Delete the artifact and related files by given HRN. If scoped token
is used, then the artifact is deleted within the project the token
belongs to. Partial HRN without version could be used. If more than
one version of artifact exists and partial HRN is passed then force
flag must be set to true.

: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.
:raises PlatformException: If platform responds with an HTTP error.
"""
path = f"/artifact/{artifact_hrn}"
url = self.format_url(self.base_url, path)
params =
if force is not None:
params = {"force": force}
resp = self.delete(url, params=params)
if resp.status_code == 200:
return True
else:
raise PlatformException(resp)

[docs]
def update_artifact_permission(self, artifact_hrn: str, body: dict) -> bool:
"""
READ grants 'readResource', MODIFY grants 'modifyResource', SHARE grants 'shareResource'.
Requires permission to 'shareResource' for the given HRN.

:param artifact_hrn: The HRN of the artifact.
:param body: a dictionary with permissions.
:return: response from the API.
:raises PlatformException: If platform responds with an HTTP error.
"""
path = f"/artifact/{artifact_hrn}/permission"
url = self.format_url(self.base_url, path)
resp = self.post(url, data=body)
if resp.status_code == 200:
return True
else:
raise PlatformException(resp)

[docs]
def get_reference_access_state(self, artifact_hrn: str) -> dict:
"""
Check that specified artifact has API key access enabled.
The calling principal must have "readResource" permission
to the artifact in order to use this endpoint.

:param artifact_hrn: The HRN of the artifact.
:return: response from the API.
:raises PlatformException: If platform responds with an HTTP error.
"""
path = f"/references/{artifact_hrn}"
url = self.format_url(self.base_url, path)
resp = self.get(url)
if resp.status_code == 200:
resp_json: dict = resp.json()
return resp_json
else:
raise PlatformException(resp)

[docs]
def grant_reference_access(self, artifact_hrn: str) -> dict:
"""
Grants API Key access to the artifact. This allows any principal that has "readResource"
permission to the artifact to use the "references" API to download artifact files using API
Access Key instead of the Bearer Token authorization. The calling principal must have
"shareResource" permission to the artifact in order to use this endpoint. This API works
for tokens with or without a project scope.

:param artifact_hrn: The HRN of the artifact.
:return: response from the API.
:raises PlatformException: If platform responds with an HTTP error.
"""
path = f"/references/{artifact_hrn}"
url = self.format_url(self.base_url, path)
resp = self.put(url)
if resp.status_code == 200:
resp_json: dict = resp.json()
return resp_json
else:
raise PlatformException(resp)

[docs]
def revoke_reference_access(self, artifact_hrn: str) -> dict:
"""
Revokes API Key access from the artifact. This removes the ability to use the "references"
API to download artifact files using API Access Key. The calling principal must have
"shareResource" permission to the artifact in order to use this endpoint. This API
works for tokens with or without a project scope.

:param artifact_hrn: The HRN of the artifact.
:return: response from the API.
:raises PlatformException: If platform responds with an HTTP error.
"""
path = f"/references/{artifact_hrn}"
url = self.format_url(self.base_url, path)
resp = self.delete(url)
if resp.status_code == 200:
resp_json: dict = resp.json()
return resp_json
else:
raise PlatformException(resp)

[docs]
def download_artifact_file_by_reference(
self, artifact_hrn: str, file_name: str, api_key: Optional[str] = None
) -> bytes:
"""
Downloads artifact file using API Key authentication. Reference access must
be granted to the artifact before this endpoint can be used. The calling
principal must have "readResource" permission to the artifact in order
to retrieve the file. In addition to API Key, this API supports tokens with
or without a project scope.

:param artifact_hrn: The HRN of the artifact.
:param file_name: The path to artifact file
:param api_key: The API key token used for request authentication
:return: response from the API.
:raises PlatformException: If platform responds with an HTTP error.
"""
path = f"/references/{artifact_hrn}/{file_name}"
url = self.format_url(self.base_url, path)
params = {"api_key": api_key}
resp = self.get(url, params=params)
if resp.status_code == 200:
return resp.content
else:
raise PlatformException(resp)