Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 1 addition & 0 deletions packages/gooddata-sdk/src/gooddata_sdk/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -249,6 +249,7 @@
CatalogDeclarativeWorkspaceModel,
CatalogDeclarativeWorkspaces,
)
from gooddata_sdk.catalog.workspace.entity_model.certification import CatalogSetCertificationRequest
from gooddata_sdk.catalog.workspace.entity_model.content_objects.dataset import (
CatalogAttribute,
CatalogDataset,
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -20,6 +20,7 @@
)
from gooddata_sdk.catalog.workspace.declarative_model.workspace.logical_model.ldm import CatalogDeclarativeModel
from gooddata_sdk.catalog.workspace.declarative_model.workspace.workspace import LAYOUT_WORKSPACES_DIR
from gooddata_sdk.catalog.workspace.entity_model.certification import CatalogSetCertificationRequest
from gooddata_sdk.catalog.workspace.entity_model.content_objects.dataset import (
CatalogAggregatedFact,
CatalogAttribute,
Expand Down Expand Up @@ -176,6 +177,42 @@ def get_metrics_catalog(self, workspace_id: str) -> list[CatalogMetric]:
catalog_metrics = [CatalogMetric.from_api(metric) for metric in metrics.data]
return catalog_metrics

def set_metric_certification(
self,
workspace_id: str,
metric_id: str,
*,
message: str | None = None,
status: str | None = "CERTIFIED",
) -> None:
"""Set or clear the certification status of a metric.

Args:
workspace_id (str):
Workspace identification string e.g. "demo"
metric_id (str):
ID of the metric to certify or decertify.
message (str | None):
Optional message to associate with the certification.
status (str | None):
Certification status. Use ``"CERTIFIED"`` (default) to certify,
or ``None`` to remove an existing certification.

Returns:
None
"""
request = CatalogSetCertificationRequest(
id=metric_id,
type="metric",
message=message,
status=status,
)
self._actions_api.set_certification(
workspace_id=workspace_id,
set_certification_request=request.as_api_model(),
_check_return_type=False,
)

def get_facts_catalog(self, workspace_id: str) -> list[CatalogFact]:
"""Retrieve all facts in a given workspace.

Expand Down
Original file line number Diff line number Diff line change
@@ -0,0 +1,43 @@
# (C) 2024 GoodData Corporation
from __future__ import annotations

from typing import Any, Literal

import attrs
from gooddata_api_client.model.set_certification_request import SetCertificationRequest

from gooddata_sdk.catalog.base import Base

CertificationStatus = Literal["CERTIFIED"]


@attrs.define(kw_only=True)
class CatalogSetCertificationRequest(Base):
"""Request to set or clear the certification status of an analytics object.

Pass ``status=None`` to remove the certification from the object.
Pass ``status="CERTIFIED"`` (the default) to certify it.
"""

id: str
type: str
message: str | None = None
status: str | None = "CERTIFIED"

@staticmethod
def client_class() -> type[SetCertificationRequest]:
return SetCertificationRequest

def as_api_model(self) -> SetCertificationRequest:
kwargs: dict[str, Any] = {}
if self.message is not None:
kwargs["message"] = self.message
# status=None means "clear certification" — always forward it so the
# caller can explicitly remove a previously set status.
kwargs["status"] = self.status
return SetCertificationRequest(
id=self.id,
type=self.type,
_check_type=False,
**kwargs,
)
Original file line number Diff line number Diff line change
Expand Up @@ -25,5 +25,20 @@ def format(self) -> str | None:
def is_hidden(self) -> bool | None:
return safeget(self.json_api_attributes, ["isHidden"])

@property
def certification(self) -> str | None:
"""Certification status of the metric (e.g. 'CERTIFIED'), or None if not certified."""
return safeget(self.json_api_attributes, ["certification"])

@property
def certification_message(self) -> str | None:
"""Optional message associated with the certification."""
return safeget(self.json_api_attributes, ["certificationMessage"])

@property
def certified_at(self) -> str | None:
"""ISO-8601 datetime string of when the certification was set, or None."""
return safeget(self.json_api_attributes, ["certifiedAt"])

def as_computable(self) -> Metric:
return SimpleMetric(local_id=self.id, item=self.obj_id)
Original file line number Diff line number Diff line change
Expand Up @@ -18,6 +18,7 @@
CatalogDependsOn,
CatalogDependsOnDateFilter,
CatalogEntityIdentifier,
CatalogSetCertificationRequest,
CatalogValidateByItem,
CatalogWorkspace,
DataSourceValidator,
Expand Down Expand Up @@ -502,3 +503,62 @@ def test_export_definition_analytics_layout(test_config):
assert deep_eq(analytics_o.analytics.export_definitions, analytics_e.analytics.export_definitions)
finally:
safe_delete(_refresh_workspaces, sdk)


# --- Certification unit tests ---


def test_set_certification_request_as_api_model_certified():
"""CatalogSetCertificationRequest serialises correctly for a CERTIFIED request."""
req = CatalogSetCertificationRequest(id="my-metric", type="metric", message="Approved", status="CERTIFIED")
api_model = req.as_api_model()
assert api_model.id == "my-metric"
assert api_model.type == "metric"
assert api_model.message == "Approved"
assert api_model.status == "CERTIFIED"


def test_set_certification_request_as_api_model_clear():
"""CatalogSetCertificationRequest serialises correctly when clearing certification (status=None)."""
req = CatalogSetCertificationRequest(id="my-metric", type="metric", status=None)
api_model = req.as_api_model()
assert api_model.id == "my-metric"
assert api_model.type == "metric"
assert api_model.status is None


def test_set_certification_request_defaults():
"""Default status is 'CERTIFIED' and message defaults to None."""
req = CatalogSetCertificationRequest(id="m1", type="metric")
assert req.status == "CERTIFIED"
assert req.message is None


def test_set_metric_certification_calls_api():
"""set_metric_certification forwards the correct request to the actions API."""
from unittest.mock import MagicMock

from gooddata_sdk.catalog.workspace.content_service import CatalogWorkspaceContentService

mock_api_client = MagicMock()
mock_api_client.entities_api = MagicMock()
mock_api_client.layout_api = MagicMock()
mock_api_client.actions_api = MagicMock()
mock_api_client.user_management_api = MagicMock()

service = CatalogWorkspaceContentService(mock_api_client)
service.set_metric_certification(
workspace_id="demo",
metric_id="total-revenue",
message="Verified by data team",
status="CERTIFIED",
)

mock_api_client.actions_api.set_certification.assert_called_once()
call_kwargs = mock_api_client.actions_api.set_certification.call_args
assert call_kwargs.kwargs["workspace_id"] == "demo"
api_req = call_kwargs.kwargs["set_certification_request"]
assert api_req.id == "total-revenue"
assert api_req.type == "metric"
assert api_req.message == "Verified by data team"
assert api_req.status == "CERTIFIED"
Loading