Skip to content
Closed
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 @@ -272,6 +272,7 @@
from gooddata_sdk.catalog.workspace.entity_model.workspace import CatalogWorkspace
from gooddata_sdk.client import GoodDataApiClient
from gooddata_sdk.compute.compute_to_sdk_converter import ComputeToSdkConverter
from gooddata_sdk.compute.model.ai_chat import ConversationFeedback, ConversationResponseList, ConversationTurnResponse
from gooddata_sdk.compute.model.attribute import Attribute
from gooddata_sdk.compute.model.base import ExecModelEntity, ObjId
from gooddata_sdk.compute.model.execution import (
Expand Down
58 changes: 57 additions & 1 deletion packages/gooddata-sdk/src/gooddata_sdk/client.py
Original file line number Diff line number Diff line change
Expand Up @@ -103,6 +103,60 @@ def _do_post_request(

return response

def _do_get_request(
self,
endpoint: str,
) -> requests.Response:
"""Perform a GET request to a specified endpoint.

Args:
endpoint (str): The endpoint URL to which the request is made.

Returns:
requests.Response: The response from the HTTP GET request.
"""
if not self._hostname.endswith("/"):
endpoint = f"/{endpoint}"

response = requests.get(
url=f"{self._hostname}{endpoint}",
headers={
"Authorization": f"Bearer {self._token}",
},
)

return response

def _do_patch_request(
self,
data: bytes,
endpoint: str,
content_type: str,
) -> requests.Response:
"""Perform a PATCH request to a specified endpoint.

Args:
data (bytes): The data to be sent in the PATCH request.
endpoint (str): The endpoint URL to which the request is made.
content_type (str): The content type of the data being sent.

Returns:
requests.Response: The response from the HTTP PATCH request.
"""
if not self._hostname.endswith("/"):
endpoint = f"/{endpoint}"

response = requests.patch(
url=f"{self._hostname}{endpoint}",
headers={
"Content-Type": content_type,
"Authorization": f"Bearer {self._token}",
},
data=data,
)

return response

def do_request(
self,
data: bytes,
Expand All @@ -126,8 +180,10 @@ def do_request(
"""
if method == HttpMethod.POST:
return self._do_post_request(data, endpoint, content_type)
elif method == HttpMethod.PATCH:
return self._do_patch_request(data, endpoint, content_type)
else:
raise NotImplementedError("Currently only supports the POST method.")
raise NotImplementedError("Currently only supports the POST and PATCH methods.")

@staticmethod
def _set_default_headers(headers: dict) -> None:
Expand Down
72 changes: 72 additions & 0 deletions packages/gooddata-sdk/src/gooddata_sdk/compute/model/ai_chat.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,72 @@
# (C) 2026 GoodData Corporation
from __future__ import annotations

from typing import Any

import attrs


@attrs.define(kw_only=True)
class ConversationFeedback:
"""Represents feedback for a conversation turn response.

Corresponds to ``FeedbackDto`` in the gen-ai OpenAPI spec.
"""

type: str
"""Feedback type. One of ``'POSITIVE'`` or ``'NEGATIVE'``."""

text: str | None = None
"""Optional free-form feedback comment."""

@classmethod
def from_api(cls, data: dict[str, Any]) -> ConversationFeedback:
return cls(
type=data["type"],
text=data.get("text"),
)

def to_api_dict(self) -> dict[str, Any]:
result: dict[str, Any] = {"type": self.type}
if self.text is not None:
result["text"] = self.text
return result


@attrs.define(kw_only=True)
class ConversationTurnResponse:
"""Represents a single conversation turn response.

Corresponds to ``ConversationTurnResponseDto`` in the gen-ai OpenAPI spec.
"""

response_id: str
created_at: str
updated_at: str
feedback: ConversationFeedback | None = None

@classmethod
def from_api(cls, data: dict[str, Any]) -> ConversationTurnResponse:
feedback_data = data.get("feedback")
return cls(
response_id=data["responseId"],
created_at=data["createdAt"],
updated_at=data["updatedAt"],
feedback=ConversationFeedback.from_api(feedback_data) if feedback_data is not None else None,
)


@attrs.define(kw_only=True)
class ConversationResponseList:
"""Represents a list of conversation turn responses.

Corresponds to ``ConversationResponseListDto`` in the gen-ai OpenAPI spec.
"""

responses: list[ConversationTurnResponse] = attrs.field(factory=list)

@classmethod
def from_api(cls, data: dict[str, Any]) -> ConversationResponseList:
return cls(
responses=[ConversationTurnResponse.from_api(r) for r in data.get("responses", [])],
)
50 changes: 50 additions & 0 deletions packages/gooddata-sdk/src/gooddata_sdk/compute/service.py
Original file line number Diff line number Diff line change
Expand Up @@ -17,6 +17,7 @@
from gooddata_api_client.model.search_result import SearchResult

from gooddata_sdk.client import GoodDataApiClient
from gooddata_sdk.compute.model.ai_chat import ConversationFeedback, ConversationResponseList
from gooddata_sdk.compute.model.execution import (
Execution,
ExecutionDefinition,
Expand Down Expand Up @@ -350,3 +351,52 @@ def sync_metadata(self, workspace_id: str, async_req: bool = False) -> None:
None
"""
self._actions_api.metadata_sync(workspace_id, async_req=async_req, _check_return_type=False)

def get_conversation_responses(
self,
workspace_id: str,
conversation_id: str,
) -> ConversationResponseList:
"""
Get conversation turn responses for a specific conversation.

Args:
workspace_id (str): workspace identifier
conversation_id (str): conversation identifier

Returns:
ConversationResponseList: List of conversation turn responses with optional feedback
"""
endpoint = f"api/v1/ai/workspaces/{workspace_id}/chat/conversations/{conversation_id}/responses"
response = self._api_client._do_get_request(endpoint)
response.raise_for_status()
return ConversationResponseList.from_api(response.json())

def set_conversation_response_feedback(
self,
workspace_id: str,
conversation_id: str,
response_id: str,
feedback_type: str,
*,
feedback_text: str | None = None,
) -> None:
"""
Set feedback for a specific conversation turn response.

Args:
workspace_id (str): workspace identifier
conversation_id (str): conversation identifier
response_id (str): response identifier
feedback_type (str): feedback type (``'POSITIVE'`` or ``'NEGATIVE'``)
feedback_text (str | None): optional free-form feedback comment. Defaults to None.
"""
feedback = ConversationFeedback(type=feedback_type, text=feedback_text)
body: dict[str, Any] = {"feedback": feedback.to_api_dict()}
endpoint = f"api/v1/ai/workspaces/{workspace_id}/chat/conversations/{conversation_id}/responses/{response_id}"
raw_response = self._api_client._do_patch_request(
data=json.dumps(body).encode("utf-8"),
endpoint=endpoint,
content_type="application/json",
)
raw_response.raise_for_status()
110 changes: 110 additions & 0 deletions packages/gooddata-sdk/tests/compute/test_ai_chat_models.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,110 @@
# (C) 2026 GoodData Corporation
from __future__ import annotations

import pytest
from gooddata_sdk.compute.model.ai_chat import ConversationFeedback, ConversationResponseList, ConversationTurnResponse


class TestConversationFeedback:
def test_from_api_positive(self) -> None:
data = {"type": "POSITIVE"}
feedback = ConversationFeedback.from_api(data)
assert feedback.type == "POSITIVE"
assert feedback.text is None

def test_from_api_negative_with_text(self) -> None:
data = {"type": "NEGATIVE", "text": "The answer was wrong"}
feedback = ConversationFeedback.from_api(data)
assert feedback.type == "NEGATIVE"
assert feedback.text == "The answer was wrong"

def test_to_api_dict_without_text(self) -> None:
feedback = ConversationFeedback(type="POSITIVE")
result = feedback.to_api_dict()
assert result == {"type": "POSITIVE"}
assert "text" not in result

def test_to_api_dict_with_text(self) -> None:
feedback = ConversationFeedback(type="NEGATIVE", text="Not helpful")
result = feedback.to_api_dict()
assert result == {"type": "NEGATIVE", "text": "Not helpful"}

@pytest.mark.parametrize("feedback_type", ["POSITIVE", "NEGATIVE"])
def test_roundtrip(self, feedback_type: str) -> None:
original = ConversationFeedback(type=feedback_type, text="some comment")
restored = ConversationFeedback.from_api(original.to_api_dict())
assert restored.type == original.type
assert restored.text == original.text


class TestConversationTurnResponse:
def test_from_api_without_feedback(self) -> None:
data = {
"responseId": "resp-123",
"createdAt": "2026-01-01T00:00:00Z",
"updatedAt": "2026-01-01T00:01:00Z",
}
turn = ConversationTurnResponse.from_api(data)
assert turn.response_id == "resp-123"
assert turn.created_at == "2026-01-01T00:00:00Z"
assert turn.updated_at == "2026-01-01T00:01:00Z"
assert turn.feedback is None

def test_from_api_with_feedback(self) -> None:
data = {
"responseId": "resp-456",
"createdAt": "2026-02-01T10:00:00Z",
"updatedAt": "2026-02-01T10:05:00Z",
"feedback": {"type": "POSITIVE", "text": "Great answer!"},
}
turn = ConversationTurnResponse.from_api(data)
assert turn.response_id == "resp-456"
assert turn.feedback is not None
assert turn.feedback.type == "POSITIVE"
assert turn.feedback.text == "Great answer!"

def test_from_api_with_null_feedback(self) -> None:
data = {
"responseId": "resp-789",
"createdAt": "2026-03-01T08:00:00Z",
"updatedAt": "2026-03-01T08:00:30Z",
"feedback": None,
}
turn = ConversationTurnResponse.from_api(data)
assert turn.feedback is None


class TestConversationResponseList:
def test_from_api_empty(self) -> None:
data: dict = {"responses": []}
result = ConversationResponseList.from_api(data)
assert result.responses == []

def test_from_api_with_responses(self) -> None:
data = {
"responses": [
{
"responseId": "r1",
"createdAt": "2026-01-01T00:00:00Z",
"updatedAt": "2026-01-01T00:01:00Z",
},
{
"responseId": "r2",
"createdAt": "2026-01-02T00:00:00Z",
"updatedAt": "2026-01-02T00:02:00Z",
"feedback": {"type": "NEGATIVE"},
},
]
}
result = ConversationResponseList.from_api(data)
assert len(result.responses) == 2
assert result.responses[0].response_id == "r1"
assert result.responses[0].feedback is None
assert result.responses[1].response_id == "r2"
assert result.responses[1].feedback is not None
assert result.responses[1].feedback.type == "NEGATIVE"

def test_from_api_missing_responses_key(self) -> None:
# 'responses' key might be missing; default to empty list
result = ConversationResponseList.from_api({})
assert result.responses == []
Loading