[refactor] Make client layer natively support both Pydantic and msgspec models

- Replace isinstance checks in client/common/asset.py with _is_model_instance
  for dual-model compatibility (9 call sites)
- Replace isinstance in client/asset.py and aio/batch.py for AtlasGlossaryTerm
- Make BulkRequest.process_attributes skip msgspec models (they handle
  relationship categorization in their own serialization pipeline)
- Use _is_model_instance in BulkRequest.process_relationship_attributes
- Register msgspec JSON encoder in pyatlan/model/core.py using model's own
  to_json(nested=True) for proper nested API format serialization
- Make Asset._convert_to_real_type_ accept v9 msgspec models via _is_model_instance
- Remove all monkey-patches from tests_v9/unit/conftest.py (Patch 1-4 no longer
  needed — dual-model support is now in production code)

All tests pass: 5798 legacy + 1540 v9

Author: Aryamanz29

pyatlan/client/aio/batch.py

@@ -4,7 +4,7 @@
 
 from typing import TYPE_CHECKING, Dict, List, Optional, cast
 
-from pyatlan.validate import validate_arguments
+from pyatlan.validate import _is_model_instance, validate_arguments
 
 from pyatlan.client.asset import (
     AssetCreationHandling,
@@ -410,7 +410,7 @@ def _track_response(self, response: AssetMutationResponse, sent: list[Asset]):
 
     @staticmethod
     def __track(tracker: List[Asset], candidate: Asset):
-        if isinstance(candidate, AtlasGlossaryTerm):
+        if _is_model_instance(candidate, AtlasGlossaryTerm):
             # trim_to_required for AtlasGlossaryTerm requires anchor
             # which is not include in AssetMutationResponse
             asset = cast(Asset, AtlasGlossaryTerm.ref_by_guid(candidate.guid))

pyatlan/client/asset.py

@@ -32,7 +32,7 @@
     parse_obj_as,
 )
 
-from pyatlan.validate import validate_arguments
+from pyatlan.validate import _is_model_instance, validate_arguments
 from tenacity import (
     RetryError,
     retry,
@@ -2571,7 +2571,7 @@ def _track_response(self, response: AssetMutationResponse, sent: list[Asset]):
 
     @staticmethod
     def __track(tracker: List[Asset], candidate: Asset):
-        if isinstance(candidate, AtlasGlossaryTerm):
+        if _is_model_instance(candidate, AtlasGlossaryTerm):
             # trim_to_required for AtlasGlossaryTerm requires anchor
             # which is not include in AssetMutationResponse
             asset = cast(Asset, AtlasGlossaryTerm.ref_by_guid(candidate.guid))

pyatlan/client/common/asset.py

@@ -59,6 +59,7 @@
     with_active_term,
 )
 from pyatlan.utils import unflatten_custom_metadata_for_entity
+from pyatlan.validate import _is_model_instance
 
 if TYPE_CHECKING:
     from pyatlan.client.aio import AsyncAtlanClient
@@ -331,7 +332,7 @@ def process_response(
                 assets := [
                     asset
                     for asset in (search_results.current_page() or search_results)
-                    if isinstance(asset, asset_type)
+                    if _is_model_instance(asset, asset_type)
                 ]
             )
         ):
@@ -535,7 +536,7 @@ def process_fluent_search_response(
         """
         if search_results and search_results.current_page():
             first_result = search_results.current_page()[0]
-            if isinstance(first_result, asset_type):
+            if _is_model_instance(first_result, asset_type):
                 return first_result
             else:
                 raise ErrorCode.ASSET_NOT_FOUND_BY_NAME.exception_with_parameters(
@@ -564,7 +565,7 @@ def process_direct_api_response(
                 asset_type.__name__, qualified_name
             )
         asset = GetByQualifiedName.handle_relationships(raw_json)
-        if not isinstance(asset, asset_type):
+        if not _is_model_instance(asset, asset_type):
             raise ErrorCode.ASSET_NOT_FOUND_BY_NAME.exception_with_parameters(
                 asset_type.__name__, qualified_name
             )
@@ -644,7 +645,7 @@ def process_fluent_search_response(
         """
         if search_results and search_results.current_page():
             first_result = search_results.current_page()[0]
-            if isinstance(first_result, asset_type):
+            if _is_model_instance(first_result, asset_type):
                 return first_result
             else:
                 raise ErrorCode.ASSET_NOT_TYPE_REQUESTED.exception_with_parameters(
@@ -667,7 +668,7 @@ def process_direct_api_response(
         :raises NotFoundError: if asset not found or wrong type
         """
         asset = GetByQualifiedName.handle_relationships(raw_json)
-        if not isinstance(asset, asset_type):
+        if not _is_model_instance(asset, asset_type):
             raise ErrorCode.ASSET_NOT_TYPE_REQUESTED.exception_with_parameters(
                 guid, asset_type.__name__
             )
@@ -1205,7 +1206,9 @@ def handle_glossary_anchor(
         :param glossary_guid: GUID of the glossary
         :raises AtlanError: if glossary_guid is required but missing
         """
-        if isinstance(asset, (AtlasGlossaryTerm, AtlasGlossaryCategory)):
+        if _is_model_instance(asset, AtlasGlossaryTerm) or _is_model_instance(
+            asset, AtlasGlossaryCategory
+        ):
             if not glossary_guid:
                 raise ErrorCode.MISSING_GLOSSARY_GUID.exception_with_parameters(
                     asset_type_name
@@ -1561,7 +1564,7 @@ def validate_search_results(
         """
         if results and results.current_page():
             first_result = results.current_page()[0]
-            if not isinstance(first_result, asset_type):
+            if not _is_model_instance(first_result, asset_type):
                 if guid is None:
                     raise ErrorCode.ASSET_NOT_FOUND_BY_NAME.exception_with_parameters(
                         asset_type.__name__, qualified_name
@@ -1665,7 +1668,7 @@ def process_search_results(
                 assets := [
                     asset
                     for asset in (results.current_page() or results)
-                    if isinstance(asset, asset_type)
+                    if _is_model_instance(asset, asset_type)
                 ]
             )
         ):
@@ -1706,13 +1709,13 @@ async def process_async_search_results(
             if current_page:
                 # Use current page if available
                 assets = [
-                    asset for asset in current_page if isinstance(asset, asset_type)
+                    asset for asset in current_page if _is_model_instance(asset, asset_type)
                 ]
             else:
                 # Otherwise, collect from async iterator
                 assets = []
                 async for asset in results:
-                    if isinstance(asset, asset_type):
+                    if _is_model_instance(asset, asset_type):
                         assets.append(asset)
 
             if assets:
@@ -1897,7 +1900,7 @@ def process_search_results(response, glossary):
         category_dict = {}
 
         for category in filter(
-            lambda a: isinstance(a, AtlasGlossaryCategory), response
+            lambda a: _is_model_instance(a, AtlasGlossaryCategory), response
         ):
             guid = category.guid
             category_dict[guid] = category
@@ -1934,13 +1937,13 @@ async def process_async_search_results(response, glossary):
             categories = [
                 asset
                 for asset in response.current_page()
-                if isinstance(asset, AtlasGlossaryCategory)
+                if _is_model_instance(asset, AtlasGlossaryCategory)
             ]
         else:
             # Collect from async iterator
             categories = []
             async for asset in response:
-                if isinstance(asset, AtlasGlossaryCategory):
+                if _is_model_instance(asset, AtlasGlossaryCategory):
                     categories.append(asset)
 
         for category in categories:

pyatlan/model/assets/core/asset.py

@@ -40,6 +40,7 @@
     StarredDetails,
 )
 from pyatlan.utils import init_guid, validate_required_fields
+from pyatlan.validate import _is_model_instance
 
 from .referenceable import Referenceable
 
@@ -163,6 +164,11 @@ def _convert_to_real_type_(cls, data):
         if isinstance(data, Asset):
             return data
 
+        # Accept v9 msgspec models as-is — they share the same class name
+        # hierarchy and are compatible via the dual-model support layer.
+        if _is_model_instance(data, Asset):
+            return data
+
         if isinstance(data, list):  # Recursively process lists
             return [cls._convert_to_real_type_(item) for item in data]
 

pyatlan/model/core.py

@@ -6,10 +6,36 @@
 from abc import ABC
 from typing import TYPE_CHECKING
 
+import msgspec
 import yaml  # type: ignore[import-untyped]
 from pydantic.v1 import BaseModel, Extra, Field, root_validator, validator
 
+from pydantic.v1.json import ENCODERS_BY_TYPE
+
 from pyatlan.model.utils import encoders, to_camel_case
+from pyatlan.validate import _is_model_instance
+
+# ---------------------------------------------------------------------------
+# Register msgspec.Struct in Pydantic's JSON encoder so that Pydantic-based
+# models (e.g. BulkRequest, AtlanRequest) can serialise v9 msgspec entities
+# when calling .json().  This enables dual-model compatibility.
+#
+# If the v9 model has its own ``to_json`` method (all v9 assets do), we
+# use it — it produces the correct nested API format with ``attributes``,
+# ``relationshipAttributes``, ``appendRelationshipAttributes``, etc.
+# For simpler structs without custom serialization we fall back to the
+# generic ``msgspec.to_builtins``.
+# ---------------------------------------------------------------------------
+
+
+def _encode_msgspec_struct(obj):
+    if hasattr(obj, "to_json"):
+        return json.loads(obj.to_json(nested=True))
+    return msgspec.to_builtins(obj)
+
+
+if msgspec.Struct not in ENCODERS_BY_TYPE:
+    ENCODERS_BY_TYPE[msgspec.Struct] = _encode_msgspec_struct
 
 if TYPE_CHECKING:
     from dataclasses import dataclass
@@ -447,6 +473,12 @@ class BulkRequest(AtlanObject, GenericModel, Generic[T]):
     def process_attributes(cls, asset):
         from pyatlan.model.assets import Asset
 
+        # v9 msgspec models handle relationship categorization
+        # in their own serialization pipeline (categorize_relationships),
+        # so return them as-is — the JSON encoder will handle the rest.
+        if isinstance(asset, msgspec.Struct):
+            return asset
+
         if not isinstance(asset, Asset):
             return asset
 
@@ -503,7 +535,7 @@ def process_relationship_attributes(cls, asset, attribute):
         # Process list of relationship attributes
         if attribute_value and isinstance(attribute_value, list):
             for value in attribute_value:
-                if value and isinstance(value, Asset):
+                if value and _is_model_instance(value, Asset):
                     if value.semantic == SaveSemantic.REMOVE:
                         remove_attributes.append(value)
                     elif value.semantic == SaveSemantic.APPEND:
@@ -532,7 +564,7 @@ def process_relationship_attributes(cls, asset, attribute):
                 exclude_attributes.add(attribute_name)
 
         # Process single relationship attribute
-        elif attribute_value and isinstance(attribute_value, Asset):
+        elif attribute_value and _is_model_instance(attribute_value, Asset):
             if attribute_value.semantic == SaveSemantic.REMOVE:
                 # Add the replace attribute to the set to exclude it
                 # from the "attributes" property in the request payload.

tests_v9/unit/conftest.py

@@ -4,134 +4,16 @@
 """
 PyTest configuration and fixtures for pyatlan_v9 unit tests.
 These fixtures provide test utilities for msgspec-based models.
-
-This module also applies compatibility patches that allow v9 msgspec models
-to work seamlessly with the legacy Pydantic-based client layer.  The patches
-MUST run at module load time (before any client classes are imported) so that
-Pydantic's cached validators pick up the patched behaviour.
 """
 
-# ---------------------------------------------------------------------------
-# v9 ↔ legacy compatibility patches  (executed at import time)
-# ---------------------------------------------------------------------------
-
-import msgspec
-from pydantic.v1.json import ENCODERS_BY_TYPE
-from pydantic.v1.main import ModelMetaclass
-
-from pyatlan.model.assets.core.asset import Asset as _LegacyAsset
-from pyatlan_v9.model.assets import Asset as _V9Asset
-
-# ---------- Patch 1: Pydantic isinstance ---------------------------------
-# Pydantic's ModelMetaclass (which extends ABCMeta) overrides
-# ``__instancecheck__`` and short-circuits with a
-# ``hasattr(instance, '__post_root_validators__')`` guard that always rejects
-# msgspec Structs.  We relax that guard so that a v9 model whose *class name*
-# appears in the MRO of the checked type is accepted.
-_original_instancecheck = ModelMetaclass.__instancecheck__
-
-
-def _v9_instancecheck(self, instance):
-    """Accept v9 msgspec models where MRO names match."""
-    if _original_instancecheck(self, instance):
-        return True
-    if isinstance(instance, msgspec.Struct):
-        v9_mro_names = {cls.__name__ for cls in type(instance).__mro__}
-        if self.__name__ in v9_mro_names:
-            return True
-    return False
-
-
-ModelMetaclass.__instancecheck__ = _v9_instancecheck
-
-# ---------- Patch 2: Pydantic JSON encoder --------------------------------
-# Register ``msgspec.Struct`` so that Pydantic's json-serialisation path
-# (used by BulkRequest.json() and AtlanRequest.json()) can serialise v9
-# models without raising ``TypeError: not JSON serializable``.
-ENCODERS_BY_TYPE[msgspec.Struct] = lambda o: msgspec.to_builtins(o)
-
-# ---------- Patch 3: Asset._convert_to_real_type_ -------------------------
-# The legacy @validate_arguments decorator calls
-# ``Asset._convert_to_real_type_(data)`` when validating Union[Asset, …]
-# parameters.  This patch makes it accept v9 Struct instances as-is.
-_original_convert = _LegacyAsset._convert_to_real_type_.__func__
-
-
-@classmethod  # type: ignore[misc]
-def _convert_to_real_type_v9_compat(cls, data):
-    """Accept v9 msgspec models in legacy Pydantic validation."""
-    if isinstance(data, _V9Asset):
-        return data
-    return _original_convert(cls, data)
-
-
-_LegacyAsset._convert_to_real_type_ = _convert_to_real_type_v9_compat  # type: ignore[assignment]
-
-# ---------- Patch 4: BulkRequest.process_attributes -----------------------
-# The legacy BulkRequest validator tries to access Pydantic-specific
-# attributes (``remove_relationship_attributes``, ``attributes.__fields_set__``,
-# ``.dict()``) on every entity.  v9 Struct models don't have those.
-# We monkey-patch the validator's *code object* so that any reference captured
-# in Pydantic's lambda closures automatically gets the updated behaviour.
-from pyatlan.model import core as _core_module  # noqa: E402
-
-_core_module.msgspec = msgspec  # inject into module globals
-
-from pyatlan.model.core import BulkRequest  # noqa: E402
-
-_original_process_func = BulkRequest.process_attributes.__func__
-
-
-def _new_process_attributes(cls, asset):
-    """BulkRequest validator that skips v9 models."""
-    if isinstance(asset, msgspec.Struct):
-        return asset
-    # --- original legacy logic (inlined) ---
-    from pyatlan.model.assets import Asset  # noqa: F811
-
-    if not isinstance(asset, Asset):
-        return asset
-
-    exclude_attributes = set()
-    asset.remove_relationship_attributes = {}
-    asset.append_relationship_attributes = {}
-    for attribute in asset.attributes.__fields_set__:
-        exclude_attributes.update(cls.process_relationship_attributes(asset, attribute))
-    exclude_relationship_attributes = {
-        key: True
-        for key in [
-            "remove_relationship_attributes",
-            "append_relationship_attributes",
-        ]
-        if not getattr(asset, key)
-    }
-    if exclude_attributes:
-        exclude_relationship_attributes = {
-            **{"attributes": exclude_attributes},
-            **exclude_relationship_attributes,
-        }
-    return asset.__class__(
-        **asset.dict(
-            by_alias=True,
-            exclude_unset=True,
-            exclude=exclude_relationship_attributes,
-        )
-    )
-
-
-_original_process_func.__code__ = _new_process_attributes.__code__
-
-# ---------------------------------------------------------------------------
-# NOW import the client (triggers AssetClient import with @validate_arguments)
-# ---------------------------------------------------------------------------
-from json import load  # noqa: E402
-from pathlib import Path  # noqa: E402
-from unittest.mock import patch  # noqa: E402
-
-import pytest  # noqa: E402
-
-from pyatlan.client.atlan import AtlanClient  # noqa: E402
-from pyatlan_v9.model.serde import Serde, get_serde  # noqa: E402
+from json import load
+from pathlib import Path
+from unittest.mock import patch
+
+import pytest
+
+from pyatlan.client.atlan import AtlanClient
+from pyatlan_v9.model.serde import Serde, get_serde
 
 # Use the same test data directory as the original tests
 TEST_DATA_DIR = Path(__file__).parent.parent.parent / "tests" / "unit" / "data"