feat(gooddata-sdk): [AUTO] Add IP allowlist policy CRUD endpoints to metadata API

yenkins-admin · yenkins-admin · commit 4e8844bcea08 · 2026-05-18T15:05:42.000Z
diff --git a/packages/gooddata-sdk/src/gooddata_sdk/__init__.py b/packages/gooddata-sdk/src/gooddata_sdk/__init__.py
@@ -109,6 +109,10 @@
     CatalogExportTemplate,
     CatalogExportTemplateAttributes,
 )
+from gooddata_sdk.catalog.organization.entity_model.ip_allowlist_policy import (
+    CatalogIpAllowlistPolicy,
+    CatalogIpAllowlistPolicyTargets,
+)
 from gooddata_sdk.catalog.organization.entity_model.jwk import (
     CatalogJwk,
     CatalogJwkAttributes,
diff --git a/packages/gooddata-sdk/src/gooddata_sdk/catalog/organization/entity_model/ip_allowlist_policy.py b/packages/gooddata-sdk/src/gooddata_sdk/catalog/organization/entity_model/ip_allowlist_policy.py
@@ -0,0 +1,50 @@
+# (C) 2026 GoodData Corporation
+from __future__ import annotations
+
+from typing import Any
+
+import attrs
+
+from gooddata_sdk.catalog.base import Base
+from gooddata_sdk.catalog.identifier import CatalogAssigneeIdentifier
+
+
+@attrs.define(kw_only=True)
+class CatalogIpAllowlistPolicy(Base):
+    """Represents an IP allowlist policy entity.
+
+    Attributes:
+        id: Unique identifier for the policy.
+        allowed_sources: IP addresses or CIDR ranges that are allowed by this policy.
+        users: Users this policy applies to.
+        user_groups: User groups this policy applies to.
+    """
+
+    id: str
+    allowed_sources: list[str] = attrs.field(factory=list)
+    users: list[CatalogAssigneeIdentifier] = attrs.field(factory=list)
+    user_groups: list[CatalogAssigneeIdentifier] = attrs.field(factory=list)
+
+    @classmethod
+    def from_api(cls, entity: dict[str, Any]) -> CatalogIpAllowlistPolicy:
+        attrs_data = entity.get("attributes") or {}
+        relationships = entity.get("relationships") or {}
+        users_rel = (relationships.get("users") or {}).get("data") or []
+        groups_rel = (relationships.get("userGroups") or {}).get("data") or []
+        return cls(
+            id=entity["id"],
+            allowed_sources=list(attrs_data.get("allowedSources") or []),
+            users=[CatalogAssigneeIdentifier(id=u["id"], type=u["type"]) for u in users_rel],
+            user_groups=[CatalogAssigneeIdentifier(id=g["id"], type=g["type"]) for g in groups_rel],
+        )
+
+
+@attrs.define(kw_only=True)
+class CatalogIpAllowlistPolicyTargets(Base):
+    """Target payload for ``addTargets`` / ``removeTargets`` action endpoints.
+
+    Attributes:
+        targets: List of assignee identifiers (users or user-groups) to add or remove.
+    """
+
+    targets: list[CatalogAssigneeIdentifier] = attrs.field(factory=list)
diff --git a/packages/gooddata-sdk/src/gooddata_sdk/catalog/organization/service.py b/packages/gooddata-sdk/src/gooddata_sdk/catalog/organization/service.py
@@ -2,7 +2,9 @@
 from __future__ import annotations
 
 import functools
+import json
 from typing import Any, Literal
+from urllib.parse import urlparse
 
 from gooddata_api_client.exceptions import NotFoundException
 from gooddata_api_client.model.declarative_export_templates import DeclarativeExportTemplates
@@ -22,6 +24,10 @@
 from gooddata_sdk.catalog.catalog_service_base import CatalogServiceBase
 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.ip_allowlist_policy import (
+    CatalogIpAllowlistPolicy,
+    CatalogIpAllowlistPolicyTargets,
+)
 from gooddata_sdk.catalog.organization.entity_model.jwk import CatalogJwk, CatalogJwkDocument
 from gooddata_sdk.catalog.organization.entity_model.llm_provider import (
     CatalogLlmProvider,
@@ -45,6 +51,10 @@
 HLL_TYPE_SETTING_ID = "hyperLogLogType"
 HLL_TYPE_SETTING_TYPE = "HLL_TYPE"
 
+_IP_ALLOWLIST_ENTITIES_BASE = "api/v1/entities/ipAllowlistPolicies"
+_IP_ALLOWLIST_ACTIONS_BASE = "api/v1/actions/ipAllowlistPolicies"
+_JSON_CONTENT_TYPE = "application/json"
+
 
 class CatalogOrganizationService(CatalogServiceBase):
     def __init__(self, api_client: GoodDataApiClient) -> None:
@@ -628,6 +638,150 @@ def delete_llm_provider(self, id: str) -> None:
         """
         self._entities_api.delete_entity_llm_providers(id, _check_return_type=False)
 
+    # IP Allowlist Policy APIs
+
+    @staticmethod
+    def _ip_policy_to_body(policy: CatalogIpAllowlistPolicy) -> bytes:
+        """Serialize a policy to a JSON:API request body."""
+        data: dict[str, Any] = {
+            "type": "ipAllowlistPolicy",
+            "id": policy.id,
+            "attributes": {"allowedSources": policy.allowed_sources},
+        }
+        rels: dict[str, Any] = {}
+        if policy.users:
+            rels["users"] = {"data": [{"id": u.id, "type": u.type} for u in policy.users]}
+        if policy.user_groups:
+            rels["userGroups"] = {"data": [{"id": g.id, "type": g.type} for g in policy.user_groups]}
+        if rels:
+            data["relationships"] = rels
+        return json.dumps({"data": data}).encode("utf-8")
+
+    def get_ip_allowlist_policy(self, policy_id: str) -> CatalogIpAllowlistPolicy:
+        """Get an IP allowlist policy by ID.
+
+        Args:
+            policy_id (str): Policy identifier.
+
+        Returns:
+            CatalogIpAllowlistPolicy: Retrieved policy.
+
+        Raises:
+            NotFoundException: Policy does not exist.
+        """
+        response = self._client._do_get_request(f"{_IP_ALLOWLIST_ENTITIES_BASE}/{policy_id}")
+        if response.status_code == 404:
+            raise NotFoundException(status=404, reason=response.reason)
+        response.raise_for_status()
+        return CatalogIpAllowlistPolicy.from_api(response.json()["data"])
+
+    def list_ip_allowlist_policies(self) -> list[CatalogIpAllowlistPolicy]:
+        """Return all IP allowlist policies in the organization.
+
+        Follows JSON:API ``links.next`` pagination transparently.
+
+        Returns:
+            list[CatalogIpAllowlistPolicy]: All policies.
+        """
+        policies: list[CatalogIpAllowlistPolicy] = []
+        endpoint = _IP_ALLOWLIST_ENTITIES_BASE
+        while True:
+            response = self._client._do_get_request(endpoint)
+            response.raise_for_status()
+            body = response.json()
+            policies.extend(CatalogIpAllowlistPolicy.from_api(item) for item in body["data"])
+            next_url = (body.get("links") or {}).get("next")
+            if not next_url:
+                break
+            parsed = urlparse(str(next_url))
+            endpoint = parsed.path.lstrip("/")
+            if parsed.query:
+                endpoint = f"{endpoint}?{parsed.query}"
+        return policies
+
+    def create_ip_allowlist_policy(self, policy: CatalogIpAllowlistPolicy) -> CatalogIpAllowlistPolicy:
+        """Create a new IP allowlist policy.
+
+        Args:
+            policy (CatalogIpAllowlistPolicy): Policy to create.
+
+        Returns:
+            CatalogIpAllowlistPolicy: Newly created policy as returned by the server.
+        """
+        response = self._client._do_post_request(
+            data=self._ip_policy_to_body(policy),
+            endpoint=_IP_ALLOWLIST_ENTITIES_BASE,
+            content_type=_JSON_CONTENT_TYPE,
+        )
+        response.raise_for_status()
+        return CatalogIpAllowlistPolicy.from_api(response.json()["data"])
+
+    def update_ip_allowlist_policy(self, policy: CatalogIpAllowlistPolicy) -> CatalogIpAllowlistPolicy:
+        """Replace an existing IP allowlist policy (full PUT).
+
+        Args:
+            policy (CatalogIpAllowlistPolicy): Updated policy object.
+
+        Returns:
+            CatalogIpAllowlistPolicy: Policy as returned by the server after update.
+
+        Raises:
+            NotFoundException: Policy does not exist.
+        """
+        response = self._client._do_put_request(
+            data=self._ip_policy_to_body(policy),
+            endpoint=f"{_IP_ALLOWLIST_ENTITIES_BASE}/{policy.id}",
+            content_type=_JSON_CONTENT_TYPE,
+        )
+        if response.status_code == 404:
+            raise NotFoundException(status=404, reason=response.reason)
+        response.raise_for_status()
+        return CatalogIpAllowlistPolicy.from_api(response.json()["data"])
+
+    def delete_ip_allowlist_policy(self, policy_id: str) -> None:
+        """Delete an IP allowlist policy.
+
+        Args:
+            policy_id (str): Policy identifier.
+
+        Raises:
+            ValueError: Policy does not exist.
+        """
+        response = self._client._do_delete_request(f"{_IP_ALLOWLIST_ENTITIES_BASE}/{policy_id}")
+        if response.status_code == 404:
+            raise ValueError(f"Cannot delete IP allowlist policy {policy_id!r}. This policy does not exist.")
+        response.raise_for_status()
+
+    def add_targets_to_ip_allowlist_policy(self, policy_id: str, targets: CatalogIpAllowlistPolicyTargets) -> None:
+        """Add users or user-groups to an existing IP allowlist policy.
+
+        Args:
+            policy_id (str): Policy identifier.
+            targets (CatalogIpAllowlistPolicyTargets): Targets to add.
+        """
+        body = json.dumps({"targets": [{"id": t.id, "type": t.type} for t in targets.targets]}).encode("utf-8")
+        response = self._client._do_post_request(
+            data=body,
+            endpoint=f"{_IP_ALLOWLIST_ACTIONS_BASE}/{policy_id}/addTargets",
+            content_type=_JSON_CONTENT_TYPE,
+        )
+        response.raise_for_status()
+
+    def remove_targets_from_ip_allowlist_policy(self, policy_id: str, targets: CatalogIpAllowlistPolicyTargets) -> None:
+        """Remove users or user-groups from an existing IP allowlist policy.
+
+        Args:
+            policy_id (str): Policy identifier.
+            targets (CatalogIpAllowlistPolicyTargets): Targets to remove.
+        """
+        body = json.dumps({"targets": [{"id": t.id, "type": t.type} for t in targets.targets]}).encode("utf-8")
+        response = self._client._do_post_request(
+            data=body,
+            endpoint=f"{_IP_ALLOWLIST_ACTIONS_BASE}/{policy_id}/removeTargets",
+            content_type=_JSON_CONTENT_TYPE,
+        )
+        response.raise_for_status()
+
     # Layout APIs
 
     def get_declarative_notification_channels(self) -> list[CatalogDeclarativeNotificationChannel]:
diff --git a/packages/gooddata-sdk/src/gooddata_sdk/client.py b/packages/gooddata-sdk/src/gooddata_sdk/client.py
@@ -105,7 +105,7 @@ def _do_post_request(
             content_type (str): The content type of the data being sent.
 
         Returns:
-            None
+            requests.Response
         """
         if not self._hostname.endswith("/"):
             endpoint = f"/{endpoint}"
@@ -121,6 +121,72 @@ 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 path (without leading slash) to call.
+
+        Returns:
+            requests.Response
+        """
+        if not self._hostname.endswith("/"):
+            endpoint = f"/{endpoint}"
+
+        return requests.get(
+            url=f"{self._hostname}{endpoint}",
+            headers={
+                "Authorization": f"Bearer {self._token}",
+            },
+        )
+
+    def _do_put_request(
+        self,
+        data: bytes,
+        endpoint: str,
+        content_type: str,
+    ) -> requests.Response:
+        """Perform a PUT request to a specified endpoint.
+
+        Args:
+            data (bytes): The data to be sent in the PUT request.
+            endpoint (str): The endpoint path (without leading slash) to call.
+            content_type (str): The content type of the data being sent.
+
+        Returns:
+            requests.Response
+        """
+        if not self._hostname.endswith("/"):
+            endpoint = f"/{endpoint}"
+
+        return requests.put(
+            url=f"{self._hostname}{endpoint}",
+            headers={
+                "Content-Type": content_type,
+                "Authorization": f"Bearer {self._token}",
+            },
+            data=data,
+        )
+
+    def _do_delete_request(self, endpoint: str) -> requests.Response:
+        """Perform a DELETE request to a specified endpoint.
+
+        Args:
+            endpoint (str): The endpoint path (without leading slash) to call.
+
+        Returns:
+            requests.Response
+        """
+        if not self._hostname.endswith("/"):
+            endpoint = f"/{endpoint}"
+
+        return requests.delete(
+            url=f"{self._hostname}{endpoint}",
+            headers={
+                "Authorization": f"Bearer {self._token}",
+            },
+        )
+
     def do_request(
         self,
         data: bytes,
diff --git a/packages/gooddata-sdk/tests/catalog/test_catalog_organization.py b/packages/gooddata-sdk/tests/catalog/test_catalog_organization.py
@@ -7,6 +7,7 @@
 from gooddata_sdk import (
     CatalogCspDirective,
     CatalogDeclarativeNotificationChannel,
+    CatalogIpAllowlistPolicy,
     CatalogJwk,
     CatalogOrganization,
     CatalogOrganizationSetting,
@@ -563,3 +564,39 @@ def test_layout_notification_channels(test_config, snapshot_notification_channel
 #         sdk.catalog_organization.put_declarative_identity_providers([])
 #         idps = sdk.catalog_organization.get_declarative_identity_providers()
 #         assert len(idps) == 0
+
+
+@gd_vcr.use_cassette(str(_fixtures_dir / "ip_allowlist_policy_crud.yaml"))
+def test_ip_allowlist_policy_crud(test_config):
+    """Integration test covering the full IP allowlist policy CRUD lifecycle."""
+    sdk = GoodDataSdk.create(host_=test_config["host"], token_=test_config["token"])
+    policy_id = "testIpPolicy"
+    policy = CatalogIpAllowlistPolicy(
+        id=policy_id,
+        allowed_sources=["192.168.0.0/16"],
+    )
+    updated_policy = CatalogIpAllowlistPolicy(
+        id=policy_id,
+        allowed_sources=["10.0.0.0/8", "192.168.0.0/16"],
+    )
+    try:
+        # Create
+        created = sdk.catalog_organization.create_ip_allowlist_policy(policy)
+        assert created.id == policy_id
+        assert created.allowed_sources == ["192.168.0.0/16"]
+
+        # Get
+        fetched = sdk.catalog_organization.get_ip_allowlist_policy(policy_id)
+        assert fetched.id == policy_id
+        assert fetched.allowed_sources == ["192.168.0.0/16"]
+
+        # List — policy must appear in the full list
+        policies = sdk.catalog_organization.list_ip_allowlist_policies()
+        assert any(p.id == policy_id for p in policies)
+
+        # Update
+        updated = sdk.catalog_organization.update_ip_allowlist_policy(updated_policy)
+        assert updated.id == policy_id
+        assert set(updated.allowed_sources) == {"10.0.0.0/8", "192.168.0.0/16"}
+    finally:
+        safe_delete(sdk.catalog_organization.delete_ip_allowlist_policy, policy_id)
diff --git a/packages/gooddata-sdk/tests/catalog/unit_tests/test_ip_allowlist_policy.py b/packages/gooddata-sdk/tests/catalog/unit_tests/test_ip_allowlist_policy.py
