feat(workspace): add method to expose configuration peroperty from workspace

Author: nazarfil

openhexa/graphql/graphql_client/client.py

@@ -558,6 +558,7 @@ def workspace(self, slug: str, **kwargs: Any) -> Optional[WorkspaceWorkspace]:
                 name
                 description
                 dockerImage
+                configuration
                 countries {
                   code
                   flag

openhexa/graphql/graphql_client/workspace.py

@@ -1,7 +1,7 @@
 # Generated by ariadne-codegen
 # Source: openhexa/graphql/queries.graphql
 
-from typing import List, Optional
+from typing import Any, List, Optional
 
 from pydantic import Field
 
@@ -17,6 +17,7 @@ class WorkspaceWorkspace(BaseModel):
     name: str
     description: Optional[str]
     docker_image: Optional[str] = Field(alias="dockerImage")
+    configuration: Optional[Any]
     countries: List["WorkspaceWorkspaceCountries"]
     permissions: "WorkspaceWorkspacePermissions"
 

openhexa/graphql/queries.graphql

@@ -200,6 +200,7 @@ query Workspace($slug: String!) {
     name
     description
     dockerImage
+    configuration
     countries {
       code
       flag

openhexa/graphql/schema.generated.graphql

@@ -4090,6 +4090,8 @@ Represents a workspace. A workspace is a shared environment where users can coll
 type Workspace {
   """File storage of the workspace represented as a bucket"""
   bucket: Bucket!
+  """JSON field containing workspace configuration as key-value pairs"""
+  configuration: JSON
   connections: [Connection!]!
   countries: [Country!]!
   createdAt: DateTime!

openhexa/sdk/workspaces/current_workspace.py

@@ -68,6 +68,60 @@ def countries(self) -> list[WorkspaceWorkspaceCountries]:
         except KeyError:
             raise WorkspaceConfigError("The workspace countries are not available in this environment.")
 
+    @property
+    def configuration(self) -> dict[str, str | dict] | None:
+        """The workspace configuration as a dictionary.
+
+        Returns a dictionary containing workspace configuration as key-value pairs,
+        where keys are strings and values can be either strings or JSON objects.
+        Returns None if no configuration is available or if not connected to the API.
+        """
+        if not self._connected:
+            return None
+        try:
+            workspace_data = OpenHexaClient().workspace(slug=self.slug)
+            return workspace_data.configuration
+        except KeyError:
+            raise WorkspaceConfigError("The workspace configuration is not available in this environment.")
+
+    def get_config(self, key: str, default=None) -> str | dict | None:
+        """Get a specific configuration value by key.
+
+        Parameters
+        ----------
+        key : str
+            The configuration key to retrieve
+        default : any, optional
+            Default value to return if key is not found
+
+        Returns
+        -------
+        str | dict | None
+            The configuration value, or default if key is not found
+        """
+        config = self.configuration
+        if config is None:
+            return default
+        return config.get(key, default)
+
+    def has_config(self, key: str) -> bool:
+        """Check if a configuration key exists.
+
+        Parameters
+        ----------
+        key : str
+            The configuration key to check
+
+        Returns
+        -------
+        bool
+            True if the key exists in configuration, False otherwise
+        """
+        config = self.configuration
+        if config is None:
+            return False
+        return key in config
+
     @property
     def database_host(self) -> str:
         """The workspace database host."""

tests/test_workspace.py

@@ -543,3 +543,48 @@ def test_connection_various_case(self, workspace):
             assert workspace.dhis2_connection("polio-123").url == url
             assert workspace.dhis2_connection("Polio-123").url == url
             assert workspace.dhis2_connection("POLIO-123").url == url
+
+    def test_workspace_configuration(self, workspace):
+        mock_config = {
+            "api_url": "https://api.example.com",
+            "debug_mode": "true",
+            "database_config": {"host": "localhost", "port": 5432},
+        }
+
+        mock_workspace_data = mock.Mock()
+        mock_workspace_data.configuration = mock_config
+
+        with mock.patch("openhexa.sdk.workspaces.current_workspace.OpenHexaClient") as mock_client:
+            mock_client.return_value.workspace.return_value = mock_workspace_data
+            assert workspace.configuration == mock_config
+
+    def test_workspace_get_config(self, workspace):
+        mock_config = {
+            "api_url": "https://api.example.com",
+            "debug_mode": "true",
+            "database_config": {"host": "localhost", "port": 5432},
+        }
+
+        mock_workspace_data = mock.Mock()
+        mock_workspace_data.configuration = mock_config
+
+        with mock.patch("openhexa.sdk.workspaces.current_workspace.OpenHexaClient") as mock_client:
+            mock_client.return_value.workspace.return_value = mock_workspace_data
+
+            assert workspace.get_config("api_url") == "https://api.example.com"
+            assert workspace.get_config("database_config") == {"host": "localhost", "port": 5432}
+            assert workspace.get_config("missing_key", "default_value") == "default_value"
+            assert workspace.get_config("missing_key") is None
+
+    def test_workspace_has_config(self, workspace):
+        mock_config = {"api_url": "https://api.example.com", "debug_mode": "true"}
+
+        mock_workspace_data = mock.Mock()
+        mock_workspace_data.configuration = mock_config
+
+        with mock.patch("openhexa.sdk.workspaces.current_workspace.OpenHexaClient") as mock_client:
+            mock_client.return_value.workspace.return_value = mock_workspace_data
+
+            assert workspace.has_config("api_url") is True
+            assert workspace.has_config("debug_mode") is True
+            assert workspace.has_config("missing_key") is False