[refactor] Improved MISSING_DATA_PRODUCT_ASSET_DSL custom error

Author: Aryamanz29

pyatlan/errors.py

@@ -629,12 +629,12 @@ class ErrorCode(Enum):
         "Set `test` to `True` or remove `username` and `password`.",
         InvalidRequestError,
     )
-    MISSING_DP_Asset_DSL = (
+    MISSING_DATA_PRODUCT_ASSET_DSL = (
         400,
         "ATLAN-PYTHON-400-072",
-        "No DataProduct Aseet DSL was found.",
-        "You must provide a DataProduct asset DSL when retrieving DataProduct assets.",
-        NotFoundError,
+        "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,

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

@@ -90,20 +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 NotFoundError: if DataProduct asset DSL cannot be found (does not exist) in Atlan
+        :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
-        if dp_dsl:
-            json_object = json.loads(dp_dsl) if dp_dsl else {}
-            request = IndexSearchRequest(**json_object.get("query", {}))
-            response = client.asset.search(request)
-        else:
-            raise ErrorCode.MISSING_DP_Asset_DSL.exception_with_parameters()
+        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

@@ -129,22 +129,19 @@ 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 NotFoundError: if DataProduct asset DSL cannot be found (does not exist) in Atlan
+        :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
-        if dp_dsl:
-            json_object = json.loads(dp_dsl) if dp_dsl else {}
-            request = IndexSearchRequest(**json_object.get("query", {}))
-            response = client.asset.search(request)
-        else:
-            raise ErrorCode.MISSING_DP_Asset_DSL.exception_with_parameters()
+        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
 
     type_name: str = Field(default="DataProduct", allow_mutation=False)

tests/integration/data_mesh_test.py

@@ -6,7 +6,6 @@
 
 from pyatlan.client.asset import IndexSearchResults
 from pyatlan.client.atlan import AtlanClient
-from pyatlan.errors import NotFoundError
 from pyatlan.model.assets import (
     Asset,
     AtlasGlossary,
@@ -503,26 +502,6 @@ def test_product_get_assets(client: AtlanClient, product: DataProduct):
     assert isinstance(asset_list, IndexSearchResults)
 
 
-@pytest.mark.order(after="test_retrieve_product")
-def test_dp_dsl_missing(client: AtlanClient, product: DataProduct):
-    assert product.guid
-    test_product = (
-        FluentSearch.select()
-        .where(Asset.GUID.eq(product.guid))
-        .where(Asset.TYPE_NAME.eq("DataProduct"))
-        .to_request()
-    )
-    retrieved_dp = client.asset.search(test_product).current_page()[0]
-    assert retrieved_dp
-    assert retrieved_dp.data_product_assets_d_s_l is None  # type: ignore
-
-    with pytest.raises(
-        NotFoundError,
-        match="No DataProduct Aseet DSL was found. Suggestion: You must provide a DataProduct asset DSL when retrieving DataProduct assets.",
-    ):
-        retrieved_dp.get_assets(client)  # type: ignore
-
-
 @pytest.mark.order(after="test_retrieve_contract")
 def test_delete_contract(client: AtlanClient, contract: DataContract):
     response = client.asset.purge_by_guid(contract.guid)

tests/unit/model/data_product_test.py

@@ -4,7 +4,7 @@
 import pytest
 
 from pyatlan.client.atlan import AtlanClient
-from pyatlan.errors import NotFoundError
+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
@@ -23,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)
@@ -146,17 +166,19 @@ def test_create_for_modification():
     _assert_product(test_product)
 
 
-def test_get_assets_with_missing_dp_asset_dsl_raise_not_found_error():
+def test_get_assets_with_missing_dp_asset_dsl(client: AtlanClient):
     data_product = DataProduct()
-    client = AtlanClient()
     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(
-        NotFoundError,
-        match="No DataProduct Aseet DSL was found. Suggestion: You must provide a DataProduct asset DSL when retrieving DataProduct assets.",
+        InvalidRequestError,
+        match=(
+            "Missing value for `data_product_assets_d_s_l`, "
+            "which is required to retrieve DataProduct assets."
+        ),
     ):
-        data_product.get_assets(client)
+        data_product.get_assets(client=client)
 
 
 def test_trim_to_required():