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 project abstraction."""
import logging
from typing import TYPE_CHECKING, List, Optional

from here.platform.api.aaa_authorization_api import AAAAuthorizationApi
from here.platform.api.base_api import BaseApi
from here.platform.schema import SchemaRegistry

logger = logging.getLogger(name)

if TYPE_CHECKING:
from here.platform.platform import Platform

[docs]
class ProjectResource:
"""
Represents how a resource, identified by HRN, is used in a project.
"""

def init(self, hrn: str, type: str, relation: str):
"""
Create a reference to a project resource
:param hrn: the HRN of the resource
:param type: the type of the resource
:param relation: the relationship of the resource with the project
"""
self.hrn = hrn
self.type = type
self.relation = relation

[docs]
class Project:
"""HERE platform project abstraction."""

def init(
self,
hrn: str,
platform: "Platform",
):
"""
Instantiate project given some HRN, API instance, and schema registry.

:param hrn: HERE resource name
:param platform: the instance of Platform
"""
self.hrn = hrn
self.api: BaseApi = platform.base_api # type: ignore
self.aaa_auth_api: AAAAuthorizationApi = platform.aaa_auth_api # type: ignore
self.platform = platform
self.schema_registry: SchemaRegistry = platform.schema_registry # type: ignore
self.data_config_api = platform.data_config_api # type: ignore

[docs]
def get_details(self) -> dict:
"""
Get project details.

:return: a dictionary with project details.
"""
return dict(self.aaa_auth_api.get_project(project=self.hrn))

[docs]
def add_member(self, member_hrn: str):
"""
Add a member to the project.

:param member_hrn: hrn of the member who needs to be added to the project
"""
self.aaa_auth_api.add_project_member(self.hrn, member_hrn)

[docs]
def assign_member_role(self, member_hrn: str, role_name: str):
"""
Add and grant access to a project to an entity.

:param member_hrn: HRN identifying the user, app or group.
:param role_name: role which needs to be assigned to a entity
"""
self.aaa_auth_api.add_project_member(project=self.hrn, member=member_hrn)
roles = self.aaa_auth_api.get_roles(resource=self.hrn)
role_hrn = ""
for role in roles["data"]:
if role_name == role["name"]:
role_hrn = role["hrn"]
self.aaa_auth_api.add_role_entity(entity=member_hrn, role=str(role_hrn))

[docs]
def list_resources(
self,
resource_type: str,
resource_limit: Optional[int] = None,
resource_relation: str = "home",

page_token: Optional[str] = None,

) -> List[ProjectResource]:
"""
Get the list of resources in the requested project or referenced in the project

:param resource_type: Type of the resource, i.e. catalog, pipeline,
pipeline-template, schema, artifact, ...
:param resource_limit: Number of entries to be returned in the response
:param resource_relation: The relation (home, reference) of the resource
:return: a list of project resources
"""
response = self.aaa_auth_api.list_project_resources(
self.hrn,
type=resource_type,
relation=resource_relation,
limit=str(resource_limit) if resource_limit else None,

page_token=page_token,

)

return [
ProjectResource(r["resource"], r["type"], r["relation"]) for r in response["items"]
]

[docs]
def delete_member(self, member_hrn: str):
"""
Remove the member from the specified Project.

:param member_hrn: HRN identifying the user, app or group.
"""
self.aaa_auth_api.delete_project_member(self.hrn, member_hrn)

[docs]
def list_members(
self,
limit: Optional[int] = None,
only_include_identities: bool = False,

page_token: Optional[str] = None,

) -> List[str]:
"""
Get the list of members of the Project.

:param limit: Number of entries to be returned in the response
:param only_include_identities: If true, returns an effective project members list
containing only user and app identities, including those that are members
of the project indirectly via a group. It will also return users who are project
admins of the project, and Resource Managers for the realm. If false, returns users,
apps, and groups that are direct members of the project, excluding any users
and apps that only have membership via a group.

:return: list of HRN of members of the project
"""
response = self.aaa_auth_api.list_project_members(
project=self.hrn,
limit=str(limit) if limit else None,
only_include_identities=only_include_identities,

page_token=page_token,

)
return [m["member"] for m in response["items"]]