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
5 changes: 5 additions & 0 deletions packages/gooddata-sdk/src/gooddata_sdk/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -264,6 +264,11 @@
CatalogDependentEntitiesResponse,
CatalogEntityIdentifier,
)
from gooddata_sdk.catalog.workspace.entity_model.resolved_llm_provider import (
CatalogResolvedLlmModel,
CatalogResolvedLlmProvider,
CatalogResolvedLlms,
)
from gooddata_sdk.catalog.workspace.entity_model.user_data_filter import (
CatalogUserDataFilter,
CatalogUserDataFilterAttributes,
Expand Down
Original file line number Diff line number Diff line change
@@ -0,0 +1,78 @@
# (C) 2026 GoodData Corporation
from __future__ import annotations

from typing import Any

import attrs
from gooddata_api_client.model.resolved_llm_provider import ResolvedLlmProvider
from gooddata_api_client.model.resolved_llms import ResolvedLlms

from gooddata_sdk.catalog.base import Base
from gooddata_sdk.utils import safeget


@attrs.define(kw_only=True)
class CatalogResolvedLlmModel(Base):
"""A single LLM model available from a resolved LLM provider."""

id: str
family: str | None = None

@staticmethod
def client_class() -> type:
from gooddata_api_client.model.llm_model import LlmModel

return LlmModel

@classmethod
def from_api(cls, data: dict[str, Any]) -> CatalogResolvedLlmModel:
return cls(
id=data["id"],
family=data.get("family"),
)


@attrs.define(kw_only=True)
class CatalogResolvedLlmProvider(Base):
"""Resolved LLM provider configuration for a workspace."""

id: str
title: str
models: list[CatalogResolvedLlmModel] = attrs.field(factory=list)

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

@classmethod
def from_api(cls, data: dict[str, Any]) -> CatalogResolvedLlmProvider:
models = [CatalogResolvedLlmModel.from_api(m) for m in (data.get("models") or [])]
return cls(
id=data["id"],
title=data["title"],
models=models,
)


@attrs.define(kw_only=True)
class CatalogResolvedLlms(Base):
"""Container for resolved LLM configuration for a workspace.

Wraps the response from the resolveLlmProviders endpoint.
The ``data`` field holds the active LLM provider, or ``None`` if
no provider is configured for the workspace.
"""

data: CatalogResolvedLlmProvider | None = None

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

@classmethod
def from_api(cls, response: Any) -> CatalogResolvedLlms:
response_dict = response.to_dict() if hasattr(response, "to_dict") else response
data_dict = safeget(response_dict, ["data"])
if data_dict is None:
return cls(data=None)
return cls(data=CatalogResolvedLlmProvider.from_api(data_dict))
Original file line number Diff line number Diff line change
Expand Up @@ -34,6 +34,7 @@
CatalogFilterView,
CatalogFilterViewDocument,
)
from gooddata_sdk.catalog.workspace.entity_model.resolved_llm_provider import CatalogResolvedLlms
from gooddata_sdk.catalog.workspace.entity_model.user_data_filter import (
CatalogUserDataFilter,
CatalogUserDataFilterDocument,
Expand Down Expand Up @@ -239,6 +240,27 @@ def resolve_workspace_settings(self, workspace_id: str, settings: list) -> dict:
]
return {setting["type"]: setting for setting in resolved_workspace_settings}

def resolve_llm_providers(self, workspace_id: str) -> CatalogResolvedLlms:
"""Resolve the active LLM provider configuration for a workspace.

Returns the active LLM provider and its associated models configured
for the given workspace. When the ENABLE_LLM_ENDPOINT_REPLACEMENT
feature flag is enabled on the server the response contains an LLM
Provider object; otherwise it falls back to the legacy LLM Endpoint
representation.

Args:
workspace_id: Workspace ID

Returns:
CatalogResolvedLlms: Resolved LLM provider configuration
"""
response = self._client.actions_api.resolve_llm_providers(
workspace_id,
_check_return_type=False,
)
return CatalogResolvedLlms.from_api(response)

# Declarative methods - workspaces

def get_declarative_workspaces(self, exclude: list[str] | None = None) -> CatalogDeclarativeWorkspaces:
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -18,6 +18,9 @@
CatalogDependsOn,
CatalogDependsOnDateFilter,
CatalogEntityIdentifier,
CatalogResolvedLlmModel,
CatalogResolvedLlmProvider,
CatalogResolvedLlms,
CatalogValidateByItem,
CatalogWorkspace,
DataSourceValidator,
Expand Down Expand Up @@ -502,3 +505,50 @@ 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)


def test_resolved_llms_from_api_with_provider():
"""Unit test: CatalogResolvedLlms.from_api deserializes a provider response correctly."""
response_dict = {
"data": {
"id": "my-provider",
"title": "My LLM Provider",
"models": [
{"id": "gpt-4o", "family": "OPENAI"},
{"id": "gpt-4o-mini", "family": "OPENAI"},
],
}
}

class _FakeResponse:
def to_dict(self):
return response_dict

result = CatalogResolvedLlms.from_api(_FakeResponse())
assert result.data is not None
assert isinstance(result.data, CatalogResolvedLlmProvider)
assert result.data.id == "my-provider"
assert result.data.title == "My LLM Provider"
assert len(result.data.models) == 2
assert isinstance(result.data.models[0], CatalogResolvedLlmModel)
assert result.data.models[0].id == "gpt-4o"
assert result.data.models[0].family == "OPENAI"


def test_resolved_llms_from_api_no_data():
"""Unit test: CatalogResolvedLlms.from_api handles null data field."""
response_dict = {"data": None}

class _FakeResponse:
def to_dict(self):
return response_dict

result = CatalogResolvedLlms.from_api(_FakeResponse())
assert result.data is None


@gd_vcr.use_cassette(str(_fixtures_dir / "test_resolve_llm_providers.yaml"))
def test_resolve_llm_providers_integration(test_config):
sdk = GoodDataSdk.create(host_=test_config["host"], token_=test_config["token"])
result = sdk.catalog_workspace.resolve_llm_providers(test_config["workspace"])
assert isinstance(result, CatalogResolvedLlms)
Loading