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
6 changes: 6 additions & 0 deletions packages/gooddata-sdk/src/gooddata_sdk/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -99,6 +99,11 @@
CatalogSectionSlideTemplate,
)
from gooddata_sdk.catalog.organization.common.widget_slides_template import CatalogWidgetSlidesTemplate
from gooddata_sdk.catalog.organization.entity_model.custom_geo_collection import (
CatalogCustomGeoCollection,
CatalogCustomGeoCollectionAttributes,
CatalogCustomGeoCollectionDocument,
)
from gooddata_sdk.catalog.organization.entity_model.directive import CatalogCspDirective
from gooddata_sdk.catalog.organization.entity_model.export_template import (
CatalogExportTemplate,
Expand All @@ -125,6 +130,7 @@
)
from gooddata_sdk.catalog.organization.entity_model.organization import CatalogOrganization
from gooddata_sdk.catalog.organization.entity_model.setting import CatalogOrganizationSetting
from gooddata_sdk.catalog.organization.layout.custom_geo_collection import CatalogDeclarativeCustomGeoCollection
from gooddata_sdk.catalog.organization.layout.export_template import (
CatalogDeclarativeExportTemplate,
)
Expand Down
Original file line number Diff line number Diff line change
@@ -0,0 +1,52 @@
# (C) 2026 GoodData Corporation
from __future__ import annotations

from attrs import define
from gooddata_api_client.model.json_api_custom_geo_collection_in import JsonApiCustomGeoCollectionIn
from gooddata_api_client.model.json_api_custom_geo_collection_in_attributes import (
JsonApiCustomGeoCollectionInAttributes,
)
from gooddata_api_client.model.json_api_custom_geo_collection_in_document import JsonApiCustomGeoCollectionInDocument

from gooddata_sdk.catalog.base import Base


@define(kw_only=True)
class CatalogCustomGeoCollectionDocument(Base):
data: CatalogCustomGeoCollection

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


@define(kw_only=True)
class CatalogCustomGeoCollection(Base):
id: str
attributes: CatalogCustomGeoCollectionAttributes | None = None

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

@classmethod
def init(
cls,
collection_id: str,
name: str | None = None,
description: str | None = None,
) -> CatalogCustomGeoCollection:
return cls(
id=collection_id,
attributes=CatalogCustomGeoCollectionAttributes(name=name, description=description),
)


@define(kw_only=True)
class CatalogCustomGeoCollectionAttributes(Base):
description: str | None = None
name: str | None = None

@staticmethod
def client_class() -> type[JsonApiCustomGeoCollectionInAttributes]:
return JsonApiCustomGeoCollectionInAttributes
Original file line number Diff line number Diff line change
@@ -0,0 +1,18 @@
# (C) 2026 GoodData Corporation
from __future__ import annotations

from attrs import define
from gooddata_api_client.model.declarative_custom_geo_collection import DeclarativeCustomGeoCollection

from gooddata_sdk.catalog.base import Base


@define(kw_only=True)
class CatalogDeclarativeCustomGeoCollection(Base):
id: str
description: str | None = None
name: str | None = None

@staticmethod
def client_class() -> type[DeclarativeCustomGeoCollection]:
return DeclarativeCustomGeoCollection
Original file line number Diff line number Diff line change
Expand Up @@ -5,9 +5,11 @@
from typing import Any

from gooddata_api_client.exceptions import NotFoundException
from gooddata_api_client.model.declarative_custom_geo_collections import DeclarativeCustomGeoCollections
from gooddata_api_client.model.declarative_export_templates import DeclarativeExportTemplates
from gooddata_api_client.model.declarative_notification_channels import DeclarativeNotificationChannels
from gooddata_api_client.model.json_api_csp_directive_in_document import JsonApiCspDirectiveInDocument
from gooddata_api_client.model.json_api_custom_geo_collection_in_document import JsonApiCustomGeoCollectionInDocument
from gooddata_api_client.model.json_api_export_template_in_document import JsonApiExportTemplateInDocument
from gooddata_api_client.model.json_api_export_template_post_optional_id_document import (
JsonApiExportTemplatePostOptionalIdDocument,
Expand All @@ -20,6 +22,7 @@

from gooddata_sdk import CatalogDeclarativeExportTemplate, CatalogExportTemplate
from gooddata_sdk.catalog.catalog_service_base import CatalogServiceBase
from gooddata_sdk.catalog.organization.entity_model.custom_geo_collection import CatalogCustomGeoCollection
from gooddata_sdk.catalog.organization.entity_model.directive import CatalogCspDirective
from gooddata_sdk.catalog.organization.entity_model.identity_provider import CatalogIdentityProvider
from gooddata_sdk.catalog.organization.entity_model.jwk import CatalogJwk, CatalogJwkDocument
Expand All @@ -30,6 +33,7 @@
CatalogLlmProviderPatchDocument,
)
from gooddata_sdk.catalog.organization.entity_model.setting import CatalogOrganizationSetting
from gooddata_sdk.catalog.organization.layout.custom_geo_collection import CatalogDeclarativeCustomGeoCollection
from gooddata_sdk.catalog.organization.layout.identity_provider import CatalogDeclarativeIdentityProvider
from gooddata_sdk.catalog.organization.layout.notification_channel import CatalogDeclarativeNotificationChannel
from gooddata_sdk.client import GoodDataApiClient
Expand Down Expand Up @@ -698,3 +702,110 @@ def switch_active_identity_provider(self, identity_provider_id: str) -> None:
)
except Exception as e:
raise ValueError(f"Error switching active identity provider: {str(e)}")

# Custom Geo Collection - Entity API

def list_custom_geo_collections(self) -> list[CatalogCustomGeoCollection]:
"""Returns a list of all custom geo collections in the current organization.

Returns:
list[CatalogCustomGeoCollection]:
List of custom geo collections in the current organization.
"""
get_custom_geo_collections = functools.partial(
self._entities_api.get_all_entities_custom_geo_collections,
_check_return_type=False,
)
custom_geo_collections = load_all_entities(get_custom_geo_collections)
return [CatalogCustomGeoCollection.from_api(collection) for collection in custom_geo_collections.data]

def get_custom_geo_collection(self, collection_id: str) -> CatalogCustomGeoCollection:
"""Get an individual custom geo collection.

Args:
collection_id (str):
Custom geo collection identification string.

Returns:
CatalogCustomGeoCollection:
Catalog custom geo collection object.
"""
collection_api = self._entities_api.get_entity_custom_geo_collections(
id=collection_id, _check_return_type=False
).data
return CatalogCustomGeoCollection.from_api(collection_api)

def create_or_update_custom_geo_collection(self, collection: CatalogCustomGeoCollection) -> None:
"""Create a new custom geo collection or overwrite an existing one with the same id.

Args:
collection (CatalogCustomGeoCollection):
Catalog custom geo collection object to be created or updated.

Returns:
None
"""
document_api = JsonApiCustomGeoCollectionInDocument(data=collection.to_api())
try:
self.get_custom_geo_collection(collection.id)
self._entities_api.update_entity_custom_geo_collections(
id=collection.id,
json_api_custom_geo_collection_in_document=document_api,
_check_return_type=False,
)
except NotFoundException:
self._entities_api.create_entity_custom_geo_collections(
json_api_custom_geo_collection_in_document=document_api,
_check_return_type=False,
)

def delete_custom_geo_collection(self, collection_id: str) -> None:
"""Delete a custom geo collection.

Args:
collection_id (str):
Custom geo collection identification string.

Returns:
None

Raises:
ValueError:
Custom geo collection does not exist.
"""
try:
self._entities_api.delete_entity_custom_geo_collections(collection_id)
except NotFoundException:
raise ValueError(
f"Can not delete {collection_id} custom geo collection. This custom geo collection does not exist."
)

# Custom Geo Collection - Layout API

def get_declarative_custom_geo_collections(self) -> list[CatalogDeclarativeCustomGeoCollection]:
"""Get all declarative custom geo collections in the current organization.

Returns:
list[CatalogDeclarativeCustomGeoCollection]:
List of declarative custom geo collections.
"""
result = self._layout_api.get_custom_geo_collections_layout(_check_return_type=False)
custom_geo_collections = result.custom_geo_collections if hasattr(result, "custom_geo_collections") else []
return [CatalogDeclarativeCustomGeoCollection.from_api(c) for c in custom_geo_collections]

def put_declarative_custom_geo_collections(
self, custom_geo_collections: list[CatalogDeclarativeCustomGeoCollection]
) -> None:
"""Put declarative custom geo collections in the current organization.

Args:
custom_geo_collections (list[CatalogDeclarativeCustomGeoCollection]):
List of declarative custom geo collections.

Returns:
None
"""
api_collections = [c.to_api() for c in custom_geo_collections]
self._layout_api.set_custom_geo_collections(
declarative_custom_geo_collections=DeclarativeCustomGeoCollections(custom_geo_collections=api_collections)
)
Original file line number Diff line number Diff line change
Expand Up @@ -3,9 +3,13 @@

from pathlib import Path

import pytest
from gooddata_api_client.exceptions import NotFoundException
from gooddata_sdk import (
CatalogCspDirective,
CatalogCustomGeoCollection,
CatalogCustomGeoCollectionAttributes,
CatalogDeclarativeCustomGeoCollection,
CatalogDeclarativeNotificationChannel,
CatalogJwk,
CatalogOrganization,
Expand Down Expand Up @@ -523,6 +527,71 @@ def test_layout_notification_channels(test_config, snapshot_notification_channel
# assert patched_identity_provider.attributes.oauth_issuer_location == oauth_issuer_location
# finally:
# sdk.catalog_organization.delete_identity_provider(identity_provider_id)
# Unit tests for CatalogCustomGeoCollection and CatalogDeclarativeCustomGeoCollection


@pytest.mark.parametrize(
"collection_id, name, description",
[
("geo-1", "My Collection", "A test geo collection"),
("geo-2", None, None),
("geo-3", "Named Only", None),
("geo-4", None, "Desc Only"),
],
)
def test_catalog_custom_geo_collection_init(collection_id, name, description):
collection = CatalogCustomGeoCollection.init(collection_id=collection_id, name=name, description=description)
assert collection.id == collection_id
assert collection.attributes is not None
assert collection.attributes.name == name
assert collection.attributes.description == description


def test_catalog_custom_geo_collection_to_api():
collection = CatalogCustomGeoCollection.init(
collection_id="geo-test",
name="Test Collection",
description="Test description",
)
api_model = collection.to_api()
assert api_model.id == "geo-test"
assert api_model.attributes.name == "Test Collection"
assert api_model.attributes.description == "Test description"


def test_catalog_custom_geo_collection_attributes():
attrs = CatalogCustomGeoCollectionAttributes(name="My Geo", description="My Desc")
assert attrs.name == "My Geo"
assert attrs.description == "My Desc"


@pytest.mark.parametrize(
"geo_id, name, description",
[
("declarative-geo-1", "Declarative Geo", "A declarative geo collection"),
("declarative-geo-2", None, None),
("declarative-geo-3", "Named", None),
],
)
def test_catalog_declarative_custom_geo_collection(geo_id, name, description):
declarative = CatalogDeclarativeCustomGeoCollection(id=geo_id, name=name, description=description)
assert declarative.id == geo_id
assert declarative.name == name
assert declarative.description == description


def test_catalog_declarative_custom_geo_collection_to_api():
declarative = CatalogDeclarativeCustomGeoCollection(
id="declarative-geo",
name="Declarative Collection",
description="Declarative description",
)
api_model = declarative.to_api()
assert api_model.id == "declarative-geo"
assert api_model.name == "Declarative Collection"
assert api_model.description == "Declarative description"


# assert len(sdk.catalog_organization.list_identity_providers()) == 0
#
#
Expand Down
Loading