fix: add sync get/set_custom_metadata to v9 Asset and fix custom metadata integration tests

- Add sync get_custom_metadata and set_custom_metadata methods to v9 Asset (overlay and generated)
- Fix flush_custom_metadata to be a no-op (set_custom_metadata already writes to business_attributes)
- Handle msgspec.UNSET in CustomMetadataProxy.__init__ (not just None)
- Fix t.attributes.anchor → t.anchor in tests (v9 flat model)

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

Author: Aryamanz29

pyatlan_v9/model/assets/_overlays/asset.py

@@ -128,12 +128,47 @@ def validate_required(self) -> None:
 
     def flush_custom_metadata(self, client=None) -> None:
         """
-        Flush (clear) all custom metadata on this asset.
+        Flush custom metadata to business_attributes.
+
+        In v9, set_custom_metadata already writes to business_attributes directly,
+        so this is a no-op for compatibility with the legacy API.
+
+        Args:
+            client: AtlanClient instance (for compatibility with legacy API)
+        """
+        pass
+
+    def get_custom_metadata(self, client, name: str):
+        """Get custom metadata by name.
+
+        Args:
+            client: AtlanClient instance
+            name: human-readable name of the custom metadata set
+
+        Returns:
+            CustomMetadataDict for the requested set
+        """
+        from pyatlan_v9.model.custom_metadata import CustomMetadataProxy
+
+        proxy = CustomMetadataProxy(
+            business_attributes=self.business_attributes, client=client
+        )
+        return proxy.get_custom_metadata(name=name)
+
+    def set_custom_metadata(self, custom_metadata, client=None) -> None:
+        """Set custom metadata on this asset.
 
         Args:
+            custom_metadata: CustomMetadataDict to set
             client: AtlanClient instance (for compatibility with legacy API)
         """
-        self.business_attributes = {}
+        from pyatlan_v9.model.custom_metadata import CustomMetadataProxy
+
+        proxy = CustomMetadataProxy(
+            business_attributes=self.business_attributes, client=client or custom_metadata._client
+        )
+        proxy.set_custom_metadata(custom_metadata=custom_metadata)
+        self.business_attributes = proxy.business_attributes
 
     async def get_custom_metadata_async(self, client, name: str):
         """Async: get custom metadata by name."""

pyatlan_v9/model/assets/asset.py

@@ -1178,12 +1178,47 @@ def validate_required(self) -> None:
 
     def flush_custom_metadata(self, client=None) -> None:
         """
-        Flush (clear) all custom metadata on this asset.
+        Flush custom metadata to business_attributes.
+
+        In v9, set_custom_metadata already writes to business_attributes directly,
+        so this is a no-op for compatibility with the legacy API.
+
+        Args:
+            client: AtlanClient instance (for compatibility with legacy API)
+        """
+        pass
+
+    def get_custom_metadata(self, client, name: str):
+        """Get custom metadata by name.
+
+        Args:
+            client: AtlanClient instance
+            name: human-readable name of the custom metadata set
+
+        Returns:
+            CustomMetadataDict for the requested set
+        """
+        from pyatlan_v9.model.custom_metadata import CustomMetadataProxy
+
+        proxy = CustomMetadataProxy(
+            business_attributes=self.business_attributes, client=client
+        )
+        return proxy.get_custom_metadata(name=name)
+
+    def set_custom_metadata(self, custom_metadata, client=None) -> None:
+        """Set custom metadata on this asset.
 
         Args:
+            custom_metadata: CustomMetadataDict to set
             client: AtlanClient instance (for compatibility with legacy API)
         """
-        self.business_attributes = {}
+        from pyatlan_v9.model.custom_metadata import CustomMetadataProxy
+
+        proxy = CustomMetadataProxy(
+            business_attributes=self.business_attributes, client=client or custom_metadata._client
+        )
+        proxy.set_custom_metadata(custom_metadata=custom_metadata)
+        self.business_attributes = proxy.business_attributes
 
     async def get_custom_metadata_async(self, client, name: str):
         """Async: get custom metadata by name."""

pyatlan_v9/model/custom_metadata.py

@@ -119,7 +119,10 @@ def __init__(
         self._metadata: Union[dict[str, CustomMetadataDict], None] = None
         self._business_attributes = business_attributes
         self._modified = False
-        if self._business_attributes is None:
+        if self._business_attributes is None or not isinstance(
+            self._business_attributes, dict
+        ):
+            self._business_attributes = None
             return
         self._metadata = {}
         for cm_id, cm_attributes in self._business_attributes.items():

tests_v9/integration/custom_metadata_test.py

@@ -849,7 +849,7 @@ def test_search_by_any_accountable(
         assert isinstance(t, AtlasGlossaryTerm)
         assert t.guid == term.guid
         assert t.qualified_name == term.qualified_name
-        anchor = t.attributes.anchor
+        anchor = t.anchor
         assert anchor
         assert anchor.name == glossary.name
         _validate_raci_attributes_replacement(
@@ -890,7 +890,7 @@ def test_search_by_specific_accountable(
         assert isinstance(t, AtlasGlossaryTerm)
         assert t.guid == term.guid
         assert t.qualified_name == term.qualified_name
-        anchor = t.attributes.anchor
+        anchor = t.anchor
         assert anchor
         assert anchor.name == glossary.name