Merge pull request #575 from atlanhq/APP-5956

APP-5956 : Add custom `AtlanError` for the `DataProduct.get_assets` method

Author: Aryamanz29

pyatlan/errors.py

@@ -629,6 +629,13 @@ class ErrorCode(Enum):
         "Set `test` to `True` or remove `username` and `password`.",
         InvalidRequestError,
     )
+    MISSING_DATA_PRODUCT_ASSET_DSL = (
+        400,
+        "ATLAN-PYTHON-400-072",
+        "Missing value for `data_product_assets_d_s_l`, which is required to retrieve DataProduct assets.",
+        "Ensure your product instance has a valid `data_product_assets_d_s_l` value before making the request.",
+        InvalidRequestError,
+    )
     AUTHENTICATION_PASSTHROUGH = (
         401,
         "ATLAN-PYTHON-401-000",

pyatlan/generator/templates/methods/asset/data_product.jinja2

@@ -90,16 +90,17 @@
         Reterieves list of all assets linked to the provided data product.
 
         :param client: connectivity to an Atlan tenant (optional). If not provided, the default client will be used.
-
+        :raises InvalidRequestError: if DataProduct asset DSL cannot be found (does not exist) in Atlan
         :raises AtlanError: if there is an issue interacting with the API
         :returns: instance of `IndexSearchResults` with list of all assets linked to the provided data product
-
         """
         from pyatlan.client.atlan import AtlanClient
 
         client = AtlanClient.get_current_client() if not client else client
         dp_dsl = self.data_product_assets_d_s_l
-        json_object = json.loads(dp_dsl) if dp_dsl else {}
+        if not dp_dsl:
+            raise ErrorCode.MISSING_DATA_PRODUCT_ASSET_DSL.exception_with_parameters()
+        json_object = json.loads(dp_dsl)
         request = IndexSearchRequest(**json_object.get("query", {}))
         response = client.asset.search(request)
         return response

pyatlan/model/assets/core/data_product.py

@@ -11,6 +11,7 @@
 
 from pydantic.v1 import Field, StrictStr, validator
 
+from pyatlan.errors import ErrorCode
 from pyatlan.model.data_mesh import DataProductsAssetsDSL
 from pyatlan.model.enums import (
     DataProductCriticality,
@@ -128,16 +129,17 @@ def get_assets(self, client: Optional[AtlanClient] = None):
         Reterieves list of all assets linked to the provided data product.
 
         :param client: connectivity to an Atlan tenant (optional). If not provided, the default client will be used.
-
+        :raises InvalidRequestError: if DataProduct asset DSL cannot be found (does not exist) in Atlan
         :raises AtlanError: if there is an issue interacting with the API
         :returns: instance of `IndexSearchResults` with list of all assets linked to the provided data product
-
         """
         from pyatlan.client.atlan import AtlanClient
 
         client = AtlanClient.get_current_client() if not client else client
         dp_dsl = self.data_product_assets_d_s_l
-        json_object = json.loads(dp_dsl) if dp_dsl else {}
+        if not dp_dsl:
+            raise ErrorCode.MISSING_DATA_PRODUCT_ASSET_DSL.exception_with_parameters()
+        json_object = json.loads(dp_dsl)
         request = IndexSearchRequest(**json_object.get("query", {}))
         response = client.asset.search(request)
         return response

tests/unit/model/data_product_test.py

@@ -3,6 +3,8 @@
 
 import pytest
 
+from pyatlan.client.atlan import AtlanClient
+from pyatlan.errors import InvalidRequestError
 from pyatlan.model.assets import AtlasGlossary, DataProduct
 from pyatlan.model.enums import CertificateStatus, DataProductStatus
 from pyatlan.model.fluent_search import CompoundQuery, FluentSearch
@@ -21,6 +23,26 @@
 ASSETS_PLAYBOOK_FILTER = '{"condition":"AND","isGroupLocked":false,"rules":[]}'
 
 
+@pytest.fixture(autouse=True)
+def set_env(monkeypatch):
+    monkeypatch.setenv("ATLAN_BASE_URL", "https://test.atlan.com")
+    monkeypatch.setenv("ATLAN_API_KEY", "test-api-key")
+
+
+@pytest.fixture()
+def client():
+    return AtlanClient()
+
+
+@pytest.fixture()
+def current_client(client, monkeypatch):
+    monkeypatch.setattr(
+        AtlanClient,
+        "get_current_client",
+        lambda: client,
+    )
+
+
 def load_json(respones_dir, filename):
     with (respones_dir / filename).open() as input_file:
         return load(input_file)
@@ -144,6 +166,21 @@ def test_create_for_modification():
     _assert_product(test_product)
 
 
+def test_get_assets_with_missing_dp_asset_dsl(client: AtlanClient):
+    data_product = DataProduct()
+    data_product.attributes.name = DATA_PRODUCT_NAME
+    data_product.parent_domain_qualified_name = DATA_PRODUCT_QUALIFIED_NAME
+    data_product.data_product_assets_d_s_l = None
+    with pytest.raises(
+        InvalidRequestError,
+        match=(
+            "Missing value for `data_product_assets_d_s_l`, "
+            "which is required to retrieve DataProduct assets."
+        ),
+    ):
+        data_product.get_assets(client=client)
+
+
 def test_trim_to_required():
     test_product = DataProduct.create_for_modification(
         qualified_name=DATA_PRODUCT_QUALIFIED_NAME,