Add jwk support to the gooddata-sdk

gooddata-sdk/gooddata_sdk/__init__.py

@@ -55,6 +55,12 @@
     ExportSettings,
 )
 from gooddata_sdk.catalog.identifier import CatalogDatasetWorkspaceDataFilterIdentifier, CatalogWorkspaceIdentifier
+from gooddata_sdk.catalog.organization.entity_model.jwk import (
+    CatalogJwk,
+    CatalogJwkAttributes,
+    CatalogJwkDocument,
+    CatalogRsaSpecification,
+)
 from gooddata_sdk.catalog.organization.entity_model.organization import CatalogOrganization
 from gooddata_sdk.catalog.organization.service import CatalogOrganizationService
 from gooddata_sdk.catalog.permission.declarative_model.dashboard_assignees import (

gooddata-sdk/gooddata_sdk/catalog/organization/entity_model/jwk.py

@@ -0,0 +1,64 @@
+# (C) 2022 GoodData Corporation
+from __future__ import annotations
+
+from typing import List, Optional, Type
+
+import attr
+
+from gooddata_api_client.model.json_api_jwk_in import JsonApiJwkIn
+from gooddata_api_client.model.json_api_jwk_in_attributes import JsonApiJwkInAttributes
+from gooddata_api_client.model.json_api_jwk_in_attributes_content import JsonApiJwkInAttributesContent
+from gooddata_api_client.model.json_api_jwk_in_document import JsonApiJwkInDocument
+from gooddata_sdk.catalog.base import Base
+
+
+@attr.s(auto_attribs=True, kw_only=True)
+class CatalogJwkDocument(Base):
+    data: CatalogJwk
+
+    @staticmethod
+    def client_class() -> Type[JsonApiJwkInDocument]:
+        return JsonApiJwkInDocument
+
+
+@attr.s(auto_attribs=True, kw_only=True)
+class CatalogJwk(Base):
+    id: str
+    attributes: Optional[CatalogJwkAttributes] = None
+
+    @staticmethod
+    def client_class() -> Type[JsonApiJwkIn]:
+        return JsonApiJwkIn
+
+    @classmethod
+    def init(
+        cls,
+        jwk_id: str,
+        rsa_spec: Optional[CatalogRsaSpecification] = None,
+    ) -> CatalogJwk:
+        return cls(id=jwk_id, attributes=CatalogJwkAttributes(content=rsa_spec))
+
+
+@attr.s(auto_attribs=True, kw_only=True)
+class CatalogJwkAttributes(Base):
+    content: Optional[CatalogRsaSpecification] = None
+
+    @staticmethod
+    def client_class() -> Type[JsonApiJwkInAttributes]:
+        return JsonApiJwkInAttributes
+
+
+@attr.s(auto_attribs=True, kw_only=True)
+class CatalogRsaSpecification(Base):
+    alg: str
+    e: str
+    kid: str
+    kty: str
+    n: str
+    use: str
+    x5c: List[str]
+    x5t: str
+
+    @staticmethod
+    def client_class() -> Type[JsonApiJwkInAttributesContent]:
+        return JsonApiJwkInAttributesContent

gooddata-sdk/gooddata_sdk/catalog/organization/service.py

@@ -1,11 +1,15 @@
 # (C) 2022 GoodData Corporation
 from __future__ import annotations
 
-from typing import Optional
+import functools
+from typing import List, Optional
 
+from gooddata_api_client.exceptions import NotFoundException
 from gooddata_sdk.catalog.catalog_service_base import CatalogServiceBase
+from gooddata_sdk.catalog.organization.entity_model.jwk import CatalogJwk, CatalogJwkDocument
 from gooddata_sdk.catalog.organization.entity_model.organization import CatalogOrganizationDocument
 from gooddata_sdk.client import GoodDataApiClient
+from gooddata_sdk.utils import load_all_entities
 
 
 class CatalogOrganizationService(CatalogServiceBase):
@@ -60,3 +64,70 @@ def update_name(self, name: str) -> None:
         organization.attributes.name = name
         organization_document = CatalogOrganizationDocument(data=organization)
         self._entities_api.update_entity_organizations(organization.id, organization_document.to_api())
+
+    def create_or_update_jwk(self, jwk: CatalogJwk) -> None:
+        """Create a new jwk or overwrite an existing jwk with the same id.
+
+        Args:
+            jwk (CatalogJwk):
+                Catalog Jwk object to be created or updated.
+
+        Returns:
+            None
+
+        Raises:
+            ValueError: Jwk can not be updated.
+        """
+        jwk_document = CatalogJwkDocument(data=jwk)
+        try:
+            self.get_jwk(jwk.id)
+            self._entities_api.update_entity_jwks(id=jwk.id, json_api_jwk_in_document=jwk_document.to_api())
+        except NotFoundException:
+            self._entities_api.create_entity_jwks(json_api_jwk_in_document=jwk_document.to_api())
+
+    def get_jwk(self, jwk_id: str) -> CatalogJwk:
+        """Get an individual jwk.
+
+        Args:
+            jwk_id (str):
+                Jwk identification string e.g. "demo"
+
+        Returns:
+            CatalogJwk:
+                Catalog jwk object containing structure of the jwk.
+        """
+        jwk_api = self._entities_api.get_entity_jwks(id=jwk_id).data
+        return CatalogJwk.from_api(jwk_api)
+
+    def delete_jwk(self, jwk_id: str) -> None:
+        """Delete a jwk.
+
+        Args:
+            jwk_id (str):
+                Jwk identification string e.g. "demo"
+
+        Returns:
+            None
+
+        Raises:
+            ValueError:
+                Jwk does not exist.
+        """
+        try:
+            self._entities_api.delete_entity_jwks(jwk_id)
+        except NotFoundException:
+            raise ValueError(f"Can not delete {jwk_id} jwk. This jwk does not exist.")
+
+    def list_jwks(self) -> List[CatalogJwk]:
+        """Returns a list of all jwks in current organization.
+
+        Args:
+            List[CatalogJwk]
+
+        Returns:
+            List[CatalogJwk]:
+                List of jwks in the current organization.
+        """
+        get_jwks = functools.partial(self._entities_api.get_all_entities_jwks, _check_return_type=False)
+        jwks = load_all_entities(get_jwks)
+        return [CatalogJwk.from_api(jwk) for jwk in jwks.data]