Refactored BusinessAttributes to CustomeMetadata

Author: ErnestoLoma

pyatlan/cache/custom_metadata_cache.py

@@ -5,7 +5,7 @@
 
 from pyatlan.client.atlan import AtlanClient
 from pyatlan.error import LogicError, NotFoundError
-from pyatlan.model.core import BusinessAttributes
+from pyatlan.model.core import CustomMetadata
 from pyatlan.model.enums import AtlanTypeCategory
 from pyatlan.model.typedef import AttributeDef, CustomMetadataDef
 
@@ -36,7 +36,7 @@ class CustomMetadataCache:
 
     @classmethod
     def _refresh_cache(cls) -> None:
-        from pyatlan.model.core import BusinessAttributes, to_snake_case
+        from pyatlan.model.core import CustomMetadata, to_snake_case
 
         client = AtlanClient.get_default_client()
         if client is None:
@@ -59,7 +59,7 @@ def _refresh_cache(cls) -> None:
                 cls.map_attr_name_to_id[type_id] = {}
                 meta_name = cm.display_name.replace(" ", "")
                 attribute_class_name = f"Attributes_{meta_name}"
-                attrib_type = type(attribute_class_name, (BusinessAttributes,), {})
+                attrib_type = type(attribute_class_name, (CustomMetadata,), {})
                 attrib_type._meta_data_type_id = type_id  # type: ignore
                 attrib_type._meta_data_type_name = type_name  # type: ignore
                 cls.map_id_to_type[type_id] = attrib_type
@@ -210,7 +210,7 @@ def get_custom_metadata(
         name: str,
         asset_type: type,
         business_attributes: Optional[dict[str, Any]] = None,
-    ) -> BusinessAttributes:
+    ) -> CustomMetadata:
         type_name = asset_type.__name__
         ba_id = cls.get_id_for_name(name)
         if ba_id is None:

pyatlan/generator/templates/entity.jinja2

@@ -9,7 +9,7 @@ from typing import Any, ClassVar, Dict, List, Optional, TypeVar
 
 from pydantic import Field, StrictStr, root_validator, validator
 
-from pyatlan.model.core import Announcement, AtlanObject, BusinessAttributes, Classification, Meaning
+from pyatlan.model.core import Announcement, AtlanObject, CustomMetadata, Classification, Meaning
 from pyatlan.model.enums import (
     ADLSAccessTier,
     ADLSAccountStatus,
@@ -226,7 +226,7 @@ class {{ entity_def.name }}({{super_classes[0]}} {%- if "Asset" in super_classes
         if not self.create_time or self.created_by:
             self.attributes.validate_required()
 
-    def get_business_attributes(self, name: str) -> BusinessAttributes:
+    def get_custom_metadata(self, name: str) -> CustomMetadata:
         from pyatlan.cache.custom_metadata_cache import CustomMetadataCache
 
         ba_id = CustomMetadataCache.get_id_for_name(name)
@@ -240,7 +240,7 @@ class {{ entity_def.name }}({{super_classes[0]}} {%- if "Asset" in super_classes
                 break
         else:
             raise ValueError(
-                f"Business attributes {name} are not applicable to {self.type_name}"
+                f"Custom metadata attributes {name} are not applicable to {self.type_name}"
             )
         if ba_type := CustomMetadataCache.get_type_for_id(ba_id):
             return (
@@ -250,27 +250,27 @@ class {{ entity_def.name }}({{super_classes[0]}} {%- if "Asset" in super_classes
             )
         else:
             raise ValueError(
-                f"Business attributes {name} are not applicable to {self.type_name}"
+                f"Custom metadata attributes {name} are not applicable to {self.type_name}"
             )
 
-    def set_business_attribute(self, business_attributes: BusinessAttributes) -> None:
+    def set_custom_metadata(self, custom_metadata: CustomMetadata) -> None:
         from pyatlan.cache.custom_metadata_cache import CustomMetadataCache
 
-        if not isinstance(business_attributes, BusinessAttributes):
+        if not isinstance(custom_metadata, CustomMetadata):
             raise ValueError(
-                "business_attributes must be an instance of BusinessAttributes"
+                "business_attributes must be an instance of CustomMetadata"
             )
         if (
-            type(business_attributes)
+            type(custom_metadata)
             not in CustomMetadataCache.types_by_asset[self.type_name]
         ):
             raise ValueError(
-                f"Business attributes {business_attributes._meta_data_type_name} are not applicable to {self.type_name}"
+                f"Business attributes {custom_metadata._meta_data_type_name} are not applicable to {self.type_name}"
             )
-        ba_dict = dict(business_attributes)
+        ba_dict = dict(custom_metadata)
         if not self.business_attributes:
             self.business_attributes = {}
-        self.business_attributes[business_attributes._meta_data_type_id] = ba_dict
+        self.business_attributes[custom_metadata._meta_data_type_id] = ba_dict
 
 {%- else %}
     {%- if entity_def.name == "Asset"  %}

pyatlan/model/assets.py

@@ -12,8 +12,8 @@
 from pyatlan.model.core import (
     Announcement,
     AtlanObject,
-    BusinessAttributes,
     Classification,
+    CustomMetadata,
     Meaning,
 )
 from pyatlan.model.enums import (
@@ -246,12 +246,12 @@ def validate_required(self):
         if not self.create_time or self.created_by:
             self.attributes.validate_required()
 
-    def get_business_attributes(self, name: str) -> BusinessAttributes:
+    def get_custom_metadata(self, name: str) -> CustomMetadata:
         from pyatlan.cache.custom_metadata_cache import CustomMetadataCache
 
         ba_id = CustomMetadataCache.get_id_for_name(name)
         if ba_id is None:
-            raise ValueError(f"No business attributes with the name: {name} exist")
+            raise ValueError(f"No custom metadata with the name: {name} exist")
         for a_type in CustomMetadataCache.types_by_asset[self.type_name]:
             if (
                 hasattr(a_type, "_meta_data_type_name")
@@ -260,7 +260,7 @@ def get_business_attributes(self, name: str) -> BusinessAttributes:
                 break
         else:
             raise ValueError(
-                f"Business attributes {name} are not applicable to {self.type_name}"
+                f"Custom metadata attributes {name} are not applicable to {self.type_name}"
             )
         if ba_type := CustomMetadataCache.get_type_for_id(ba_id):
             return (
@@ -270,27 +270,27 @@ def get_business_attributes(self, name: str) -> BusinessAttributes:
             )
         else:
             raise ValueError(
-                f"Business attributes {name} are not applicable to {self.type_name}"
+                f"Custom metadata attributes {name} are not applicable to {self.type_name}"
             )
 
-    def set_business_attribute(self, business_attributes: BusinessAttributes) -> None:
+    def set_custom_metadata(self, custom_metadata: CustomMetadata) -> None:
         from pyatlan.cache.custom_metadata_cache import CustomMetadataCache
 
-        if not isinstance(business_attributes, BusinessAttributes):
+        if not isinstance(custom_metadata, CustomMetadata):
             raise ValueError(
-                "business_attributes must be an instance of BusinessAttributes"
+                "business_attributes must be an instance of CustomMetadata"
             )
         if (
-            type(business_attributes)
+            type(custom_metadata)
             not in CustomMetadataCache.types_by_asset[self.type_name]
         ):
             raise ValueError(
-                f"Business attributes {business_attributes._meta_data_type_name} are not applicable to {self.type_name}"
+                f"Business attributes {custom_metadata._meta_data_type_name} are not applicable to {self.type_name}"
             )
-        ba_dict = dict(business_attributes)
+        ba_dict = dict(custom_metadata)
         if not self.business_attributes:
             self.business_attributes = {}
-        self.business_attributes[business_attributes._meta_data_type_id] = ba_dict
+        self.business_attributes[custom_metadata._meta_data_type_id] = ba_dict
 
 
 class Asset(Referenceable):

pyatlan/model/core.py

@@ -189,7 +189,7 @@ class BulkRequest(AtlanObject, GenericModel, Generic[T]):
     entities: list[T]
 
 
-class BusinessAttributes(dict):
+class CustomMetadata(dict):
     _meta_data_type_name = ""
     _meta_data_type_id = ""
 

tests/integration/custom_metadata_test.py

@@ -87,7 +87,7 @@ def test_003_purge_custom_metadata(client: AtlanClient):
 
 def test_custom_metadata_has_human_readable_properties(client: AtlanClient):
     table = client.get_asset_by_guid("5f47cfb9-1313-4f03-9213-9913fff3c878", Table)
-    anomalo = table.get_business_attributes("Anomalo")
+    anomalo = table.get_custom_metadata("Anomalo")
     assert hasattr(anomalo, "data_volume")
     assert anomalo.data_volume is None
     assert hasattr(anomalo, "data_volume_details")
@@ -114,7 +114,7 @@ def test_custom_metadata_has_human_readable_properties(client: AtlanClient):
     assert anomalo.validation_rules is None
     assert hasattr(anomalo, "validation_rules_details")
     assert anomalo.validation_rules_details is None
-    monte_carlo = table.get_business_attributes("Monte Carlo")
+    monte_carlo = table.get_custom_metadata("Monte Carlo")
     assert monte_carlo is not None
 
 

tests/unit/test_model.py

@@ -1031,9 +1031,9 @@ def test_get_business_attributes_when_name_not_valid_raises_value_error(
 ):
     mock_client.return_value.get_typedefs.return_value = type_def_response
     with pytest.raises(
-        ValueError, match="No business attributes with the name: Zoro exist"
+        ValueError, match="No custom metadata with the name: Zoro exist"
     ):
-        table.get_business_attributes("Zoro")
+        table.get_custom_metadata("Zoro")
 
 
 @patch("pyatlan.cache.custom_metadata_cache.AtlanClient")
@@ -1043,9 +1043,9 @@ def test_get_business_attributes_when_name_not_appropriate_for_asset_raises_valu
     mock_client.get_default_client.return_value = mock_client
     mock_client.get_typedefs.return_value = type_def_response
     with pytest.raises(
-        ValueError, match="Business attributes Moon are not applicable to Table"
+        ValueError, match="Custom metadata attributes Moon are not applicable to Table"
     ):
-        table.get_business_attributes("Moon")
+        table.get_custom_metadata("Moon")
 
 
 @patch("pyatlan.cache.custom_metadata_cache.AtlanClient")
@@ -1054,7 +1054,7 @@ def test_get_business_attributes_with_valid_name_returns_empty_attribute_when_ta
 ):
     mock_client.return_value.get_typedefs.return_value = type_def_response
 
-    monte_carlo = table.get_business_attributes("Monte Carlo")
+    monte_carlo = table.get_custom_metadata("Monte Carlo")
 
     assert monte_carlo is not None
     assert monte_carlo.freshness is None
@@ -1074,7 +1074,7 @@ def test_get_business_attributes_with_valid_name_returns_attribute_when_table_ha
         }
     }
 
-    monte_carlo = table.get_business_attributes("Monte Carlo")
+    monte_carlo = table.get_custom_metadata("Monte Carlo")
 
     assert monte_carlo is not None
     assert monte_carlo.freshness == "pass"
@@ -1086,9 +1086,9 @@ def test_set_busines_attributes_with_non_business_attributes_object_raises_value
 ):
     with pytest.raises(
         ValueError,
-        match="business_attributes must be an instance of BusinessAttributes",
+        match="business_attributes must be an instance of CustomMetadata",
     ):
-        table.set_business_attribute({})
+        table.set_custom_metadata({})
 
 
 def test_set_business_attributes_with_non_appropriate_meta_data_type_name_raises_value_error(
@@ -1097,9 +1097,9 @@ def test_set_business_attributes_with_non_appropriate_meta_data_type_name_raises
 
     with pytest.raises(
         ValueError,
-        match="business_attributes must be an instance of BusinessAttributes",
+        match="business_attributes must be an instance of CustomMetadata",
     ):
-        table.set_business_attribute({})
+        table.set_custom_metadata({})
 
 
 @patch("pyatlan.cache.custom_metadata_cache.AtlanClient")
@@ -1113,7 +1113,7 @@ def test_assigning_to_invalid_business_attribute_raises_attribute_error(
             TABLE_URL: "https://getmontecarlo.com/catalog/",
         }
     }
-    business_attributes = table.get_business_attributes("Monte Carlo")
+    business_attributes = table.get_custom_metadata("Monte Carlo")
     with pytest.raises(AttributeError, match="Attribute bogus does not exist"):
         business_attributes.bogus = "123"
 
@@ -1130,7 +1130,7 @@ def test_set_business_attributes_with_business_attribute_not_appropriate_to_asse
         ValueError,
         match="Business attributes Moon are not applicable to Table",
     ):
-        table.set_business_attribute(moon)
+        table.set_custom_metadata(moon)
 
 
 @patch("pyatlan.cache.custom_metadata_cache.AtlanClient")
@@ -1145,11 +1145,11 @@ def test_set_business_attributes_with_appropriate_business_attribute_updates_dic
             TABLE_URL: "https://getmontecarlo.com/catalog/",
         }
     }
-    monte_carlo = table.get_business_attributes("Monte Carlo")
+    monte_carlo = table.get_custom_metadata("Monte Carlo")
 
     monte_carlo.freshness = "fail"
     monte_carlo.table_url = "http://anywhere.com"
-    table.set_business_attribute(monte_carlo)
+    table.set_custom_metadata(monte_carlo)
 
     monte_carlo_dict = table.business_attributes[MONTE_CARLO]
     assert monte_carlo_dict[FRESHNESS] == monte_carlo.freshness