from tdw_catalog.entity import Entity, Property, EntityBase
from tdw_catalog.relations import _OrganizationRelation
from tdw_catalog.utils import LegacyFilter
from dataclasses import dataclass, field
from datetime import datetime
from typing import TYPE_CHECKING, Dict, List, Optional
import tdw_catalog.user as user
import tdw_catalog.team as team
from tdw_catalog.errors import CatalogException, _convert_error, _raise_error
from tdw_catalog import Catalog
if TYPE_CHECKING:
import tdw_catalog.organization as organization
[docs]@dataclass
class OrganizationMemberRoles:
"""
:class:`.OrganizationMemberRoles` defines the roles which an :class:`.OrganizationMember` has within an :class:`.Organization`
Attributes
----------
role_data_uploader : bool
Whether this :class:`.OrganizationMember` is allowed to upload to the :class:`.Organization`
role_data_viewer : bool
Whether this :class:`.OrganizationMember` is allowed to view Datasets within the :class:`.Organization`
role_data_editor : bool
Whether this :class:`.OrganizationMember` is allowed to modify Datasets within the :class:`.Organization`
role_member_manager : bool
Whether this :class:`.OrganizationMember` is allowed to manage members within the :class:`.Organization`
role_organization_manager : bool
Whether this :class:`.OrganizationMember` is allowed to manage the :class:`.Organization`
role_admin : bool
Whether this :class:`.OrganizationMember` is an :class:`.Organization` administrator
role_topic_manager : bool
Whether this :class:`.OrganizationMember` is allowed to manage :class:`.Topic`\\ s within the :class:`.Organization`
role_field_template_manager : bool
Whether this :class:`.OrganizationMember` is allowed to manage :class:`.MetadataTemplate`\\ s
"""
role_data_uploader: bool = field(default=False)
role_data_viewer: bool = field(default=False)
role_data_editor: bool = field(default=False)
role_member_manager: bool = field(default=False)
role_organization_manager: bool = field(default=False)
role_admin: bool = field(default=False)
role_topic_manager: bool = field(default=False)
role_field_template_manager: bool = field(default=False)
def deserialize(self, data: dict) -> None:
self.role_data_uploader = data.get("role_data_uploader", False)
self.role_data_viewer = data.get("role_data_viewer", False)
self.role_data_editor = data.get("role_data_editor", False)
self.role_member_manager = data.get("role_member_manager", False)
self.role_organization_manager = data.get("role_organization_manager",
False)
self.role_admin = data.get("role_admin", False)
self.role_topic_manager = data.get("role_topic_manager", False)
self.role_field_template_manager = data.get(
"role_field_template_manager", False)
def serialize(self) -> dict:
return {
"role_data_uploader": self.role_data_uploader,
"role_data_viewer": self.role_data_viewer,
"role_data_editor": self.role_data_editor,
"role_member_manager": self.role_member_manager,
"role_organization_manager": self.role_organization_manager,
"role_admin": self.role_admin,
"role_topic_manager": self.role_topic_manager,
"role_field_template_manager": self.role_field_template_manager
}
[docs]@Entity([
Property("organization_id",
str,
relation="tdw_catalog.organization.Organization",
serialize=True),
Property("roles", OrganizationMemberRoles, writable=True, serialize=False),
Property("created_at", datetime),
Property("updated_at", datetime)
])
class OrganizationMember(user.User, _OrganizationRelation):
"""
An :class:`.OrganizationMember` reflects a relationship between :class:`.User` and :class:`.Organization`, where the :class:`.User` has been invited to the :class:`.Organization` and been granted specific privileges within the :class:`.Organization`.
Attributes
----------
user_id : str
The unique user ID of the :class:`.OrganizationMember`
organization : organization.Organization
The :class:`.Organization` object that relates to the `organization_id` of this model
organization_id : str
The unique ID of the :class:`.Organization` to which this :class:`.OrganizationMember` belongs
roles: OrganizationMemberRoles
The roles this Member has within their :class:`.Organization`
created_at : datetime
The datetime at which this :class:`.OrganizationMember` was added to the :class:`.Organization`
updated_at : datetime
The datetime at which this :class:`.OrganizationMember` was last updated
"""
_client: 'Catalog'
organization: 'organization.Organization'
organization_id: str
roles: OrganizationMemberRoles
created_at: datetime
updated_at: datetime
[docs] @classmethod
def get(cls, client: 'Catalog', organization_id: str, id: str):
"""
Retrieve an :class:`.OrganizationMember` belonging to this :class:`.Organization`
Parameters
----------
client : Catalog
The :class:`.Catalog` client of the :class:`.Organization` containing the :class:`.OrganizationMember`
organization_id : str
The unique ID of the :class:`.Organization`
id : str
The unique ID of the :class:`.OrganizationMember`
Returns
-------
OrganizationMember
The :class:`.OrganizationMember` associated with the given ID
Raises
------
CatalogInternalException
If call to the :class:`.Catalog` server fails
CatalogNotFoundException
If no :class:`.OrganizationMember` is found matching the provided ID
CatalogPermissionDeniedException
If the caller is not allowed to retrieve this :class:`.OrganizationMember`
"""
try:
res = client._get_organization_member(
organization_id=organization_id, user_id=id)
return OrganizationMember(client, **res)
except Exception as e:
err = _raise_error(
e, "Unable to fetch OrganizationMember {}".format(id))
def __init__(self, client, **kwargs):
super().__init__(client, **kwargs)
if self.roles is None:
self.roles = OrganizationMemberRoles()
def __str__(self) -> str:
return f"<OrganizationMember user={self.user_id}>"
def __eq__(self, other) -> bool:
return isinstance(other,
self.__class__) and self.user_id == other.user_id
def serialize(self) -> dict:
result = super().serialize()
return {**result, **self.roles.serialize()}
def deserialize(self, data):
super().deserialize(data)
# If no roles have been specified we can create the defaults
if self.roles is None:
self.roles = OrganizationMemberRoles()
# If roles is in data, deserializing would overwrite, so we can just
# leave it as is. Otherwise, we can deserialize the roles off the resp
if "roles" not in data:
self.roles.deserialize(data)
[docs] def save(self) -> None:
"""
Update this :class:`.OrganizationMember`, saving any changes to its roles
Parameters
----------
None
Returns
-------
None
Raises
------
CatalogPermissionDeniedException
If the caller is not allowed to update this :class:`.OrganizationMember`
CatalogAlreadyExistsException
If the caller is attempting to invite a member which is already a member of the :class:`.Organization`
CatalogException
If call to the :class:`.Catalog` server fails
"""
try:
res = self._client._update_organization_member(
member=self.serialize())
self.deserialize(res)
except Exception as e:
raise _convert_error(e)
[docs] def delete(self) -> None:
"""
Remove this :class:`.OrganizationMember` from the :class:`.Organization`. This :class:`.OrganizationMember` object
should not be used after `delete()` returns successfully.
Parameters
----------
None
Returns
-------
None
Raises
------
CatalogPermissionDeniedException
If the caller is not allowed to delete this :class:`.OrganizationMember`, or if the caller is attempting to delete themselves
CatalogException
If call to the :class:`.Catalog` server fails
"""
try:
self._client._remove_organization_member(member=self.serialize())
except Exception as e:
raise _convert_error(e)
[docs] def get_teams(self,
filter: Optional[LegacyFilter] = None) -> List[team.Team]:
"""
Retrieve the :class:`.Team`\\ s to which this :class:`.OrganizationMember` belongs
Parameters
----------
filter : Optional[LegacyFilter]
An optional filter on the returned :class:`.Team` list, useful for pagination of results
Returns
-------
List[Team]
The list of :class:`.Team`\\ s to which this :class:`.OrganizationMember` belongs
Raises
------
CatalogPermissionDeniedException
If the caller is not allowed to retrieve :class:`.Team`\\ s from this :class:`.Organization`
CatalogException
If call to the :class:`.Catalog` server fails
"""
try:
org_member_teams = self._client._list_organization_member_groups(
user_id=self.user_id,
organization_id=self.organization_id,
filter=filter)
return list(
map(
lambda org_member_team: team.Team(
client=self._client,
id=org_member_team["id"],
organization_id=org_member_team["organization_id"],
title=org_member_team["title"]),
org_member_teams,
))
except Exception as e:
raise _convert_error(e)