from typing import TYPE_CHECKING, List, Optional
from tdw_catalog._client import _Client
from tdw_catalog.errors import CatalogInvalidArgumentException, \
_convert_error
from tdw_catalog.utils import ListOrganizationsFilter
if TYPE_CHECKING:
from tdw_catalog.user import User
import tdw_catalog.organization as organization
[docs]class Catalog(_Client):
"""
A :class:`.Catalog` is the primary client object for a ThinkData Catalog
Parameters
----------
api_key : Optional[str])
An optional api key for the Catalog platform. This parameter must be supplied,
containing your personal API key for the Catalog platform, and can only be
omitted when supplied via an environment variable `CATALOG_API_KEY` instead.
auth_url : Optional[str])
An optional auth url for the Catalog platform. This parameter must only be
supplied when connecting to a dedicated Catalog deployment, and can be
populated via an envrionment variable `CATALOG_AUTH_URL` instead. Defaults
to the auth url for the ThinkData Works SaaS Catalog platform
(https://account.ee.namara.io).
api_url : Optional[str])
An optional API url for the Catalog platform. This parameter must only be
supplied when connecting to a dedicated Catalog deployment, and can be
populated via an envrionment variable `CATALOG_API_URL` instead. Defaults
to the API url for the ThinkData Works SaaS Catalog platform
(https://api.ee.namara.io).
"""
def __init__(self, *args, **kwargs) -> None:
super().__init__(self, **kwargs)
def __clone_with_org_context(self, org_id: str) -> 'Catalog':
c = Catalog(api_key=self.api_key,
auth_url=self.base_auth_url,
api_url=self.base_api_url,
org_context=org_id)
return c
# TODO listing all users does not currently work
def _list_users(self,
user_ids: Optional[List[str]] = None) -> 'List[User]':
"""
Retrieve all :class:`.User`\\ s registered with the :class:`.Catalog` server
Parameters
----------
user_ids : List[str]
An optional list of :class:`.User` IDs to retrieve (retrieves all if not specified)
Returns
-------
list[User]
The :class:`.User`s which are a registered with the :class:`.Catalog` server
Raises
------
CatalogPermissionDeniedException
If the caller is not allowed to list :class:`.Catalog` :class:`.User`\\ s
CatalogException
If call to the :class:`.Catalog` server fails
"""
from tdw_catalog.user import User
try:
users_details_url = f'{self.base_auth_url}/users/details.json'
if (user_ids != None):
users_details_url += f'?user_ids={",".join(user_ids)}'
res = self._make_catalog_request(users_details_url)
identity_users = res.json()
return list(
map(
lambda identity_user: User(
client=self,
id=identity_user["id"],
name=identity_user["name"],
email=identity_user["email"],
),
identity_users,
))
except Exception as e:
raise _convert_error(e)
def _get_user(self, id: str) -> 'User':
"""
Retrieve the a specific :class:`.Catalog` :class:`.User`
Parameters
----------
id : str
The unique ID of the :class:`.User`
Returns
-------
User
The :class:`.User` with the given ID
Raises
------
CatalogPermissionDeniedException
If the caller is not allowed to fetch :class:`.Catalog` :class:`.User`\\ s
CatalogInvalidArgumentException
If the given :class:`.User` does not exist
CatalogException
If call to the :class:`.Catalog` server fails
"""
from tdw_catalog.user import User
try:
user_details_url = f'{self.base_auth_url}/users/details.json?user_ids={id}'
res = self._make_catalog_request(user_details_url)
identity_users = res.json()
if (not identity_users or len(identity_users) == 0):
raise CatalogInvalidArgumentException(
message=f"User {id} does not exist.")
return User(
client=self,
user_id=identity_users[0]["id"],
name=identity_users[0]["name"],
email=identity_users[0]["email"],
)
except Exception as e:
raise _convert_error(e)
[docs] def list_organizations(
self,
filter: 'Optional[ListOrganizationsFilter]' = None
) -> 'List[organization.Organization]':
"""
Retrieve the list of :class:`.Organization`\\ s to which the caller belongs.
Parameters
----------
filter : ListOrganizationsFilter
An optional filter on the returned :class:`.Organization`\\ s (`None` by default).
Returns
-------
list[Organization]
The list of :class:`.Organization`\\ s to which the caller belongs, ordered by title (ascending).
Raises
------
CatalogException
If there is an issue communicating with the :class:`.Catalog` server, or an issue with the server itself
"""
import tdw_catalog.organization as organization
try:
rpc_results = self._list_organizations(filter=filter)
return list(
map(
lambda o: organization.Organization(
client=self.__clone_with_org_context(o["id"]), **o),
rpc_results))
except Exception as e:
raise _convert_error(e)
[docs] def get_organization(self, id: str) -> 'organization.Organization':
"""
Retrieve a specific :class:`.Organization`
Parameters
----------
id : str
The UUID of the :class:`.Organization`
Returns
-------
Organization
The :class:`.Organization` which has the provided id, if it exists and the caller is a member of it
Raises
------
CatalogPermissionDeniedException
If the caller is not a member of the given :class:`.Organization`, or if it does not exist
CatalogException
If there is an issue communicating with the :class:`.Catalog` server, or an issue with the server itself
"""
import tdw_catalog.organization as organization
return organization.Organization.get(
client=self.__clone_with_org_context(id), id=id)
[docs] def create_organization(self, title: str) -> 'organization.Organization':
"""
Creates an :class:`.Organization`
Parameters
----------
title : str
The title for the new :class:`.Organization`
Returns
-------
Organization
The created :class:`.Organization`
Raises
------
CatalogException
If there is an issue communicating with the :class:`.Catalog` server, or an issue with the server itself
"""
import tdw_catalog.organization as organization
try:
rpc_result = self._create_organization(
organization={"title": title})
return organization.Organization(
client=self.__clone_with_org_context(rpc_result["id"]),
**rpc_result)
except Exception as e:
raise _convert_error(e)