[generator] Aligned generator templates with the latest SDK changes for IndistinctAsset fallback model

Author: Aryamanz29

pyatlan/generator/class_generator.py

@@ -938,6 +938,9 @@ def filter_attributes_of_custom_entity_type():
     for file in (ASSETS_DIR).glob("*.py"):
         file.unlink()
     for file in (CORE_ASSETS_DIR).glob("*.py"):
+        # Remove all files in the core assets directory except `indistinct.py`
+        if file.name == "indistinct_asset.py":
+            continue
         file.unlink()
     generator = Generator()
     EnumDefInfo.create(type_defs.enum_defs)

pyatlan/generator/templates/core/init.jinja2

@@ -6,6 +6,7 @@
 {{ asset_import }}
 {% endif %}
 {% endfor %}
+from .indistinct_asset import IndistinctAsset  # noqa: F401
 
 # Update asset forward references:
 localns = locals()

pyatlan/generator/templates/init.jinja2

@@ -9,6 +9,7 @@ __PYATLAN_ASSETS__ = {
                 "{{ asset.name }}"{% if not loop.last %},{% endif %}
             {% endif %}
         {% endfor %}
+        "IndistinctAsset"
     ],
     {% for asset in assets -%}
         {% if not asset.is_core_asset and asset.name not in asset._CORE_ASSETS  %}

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

@@ -104,6 +104,8 @@
 
     @classmethod
     def _convert_to_real_type_(cls, data):
+        from .indistinct_asset import IndistinctAsset
+
         """Convert raw asset data into the appropriate asset type."""
         if isinstance(data, Asset):
             return data
@@ -123,7 +125,9 @@
             sys.modules.get("pyatlan.model.assets", {}), data_type, None
         )
 
-        return sub_type(**data) if sub_type else None  # Gracefully handle missing types
+        return (
+            sub_type(**data) if sub_type else IndistinctAsset(**data)
+        )  # If no subtype found, return IndistinctAsset
 
     if TYPE_CHECKING:
         from pyatlan.model.lineage import FluentLineage

pyatlan/generator/templates/mypy_init.jinja2

@@ -10,7 +10,8 @@ __all__ = [
         {% if not asset.is_core_asset %}
             "{{ asset.name }}"{% if not loop.last %},{% endif %}
         {% endif %}
-    {% endfor %}
+    {% endfor %},
+    "IndistinctAsset"
 ]
 
 {% for asset in assets -%}
@@ -20,3 +21,4 @@ from .core.{{ asset.module_name }} import {{ asset.name }}
 from .{{ asset.module_name }} import {{ asset.name }}
 {% endif %}
 {% endfor %}
+from .core.indistinct_asset import IndistinctAsset

pyatlan/model/structs.py

@@ -45,6 +45,19 @@ def flatten_structs_attributes(cls, values: Dict[str, Any]) -> Dict[str, Any]:
         return values
 
 
+class MCRuleSchedule(AtlanObject):
+    """Description"""
+
+    mc_rule_schedule_type: Optional[str] = Field(default=None, description="")
+    mc_rule_schedule_interval_in_minutes: Optional[int] = Field(
+        default=None, description=""
+    )
+    mc_rule_schedule_start_time: Optional[datetime] = Field(
+        default=None, description=""
+    )
+    mc_rule_schedule_crontab: Optional[str] = Field(default=None, description="")
+
+
 class DbtJobRun(AtlanObject):
     """Description"""
 
@@ -61,19 +74,6 @@ class DbtJobRun(AtlanObject):
     dbt_compiled_code: Optional[str] = Field(default=None, description="")
 
 
-class MCRuleSchedule(AtlanObject):
-    """Description"""
-
-    mc_rule_schedule_type: Optional[str] = Field(default=None, description="")
-    mc_rule_schedule_interval_in_minutes: Optional[int] = Field(
-        default=None, description=""
-    )
-    mc_rule_schedule_start_time: Optional[datetime] = Field(
-        default=None, description=""
-    )
-    mc_rule_schedule_crontab: Optional[str] = Field(default=None, description="")
-
-
 class AwsCloudWatchMetric(AtlanObject):
     """Description"""
 
@@ -104,13 +104,6 @@ class ColumnValueFrequencyMap(AtlanObject):
     column_value_frequency: Optional[int] = Field(default=None, description="")
 
 
-class SourceTagAttachmentValue(AtlanObject):
-    """Description"""
-
-    tag_attachment_key: Optional[str] = Field(default=None, description="")
-    tag_attachment_value: Optional[str] = Field(default=None, description="")
-
-
 class BadgeCondition(AtlanObject):
     """Description"""
 
@@ -143,6 +136,13 @@ def create(
     badge_condition_colorhex: Optional[str] = Field(default=None, description="")
 
 
+class SourceTagAttachmentValue(AtlanObject):
+    """Description"""
+
+    tag_attachment_key: Optional[str] = Field(default=None, description="")
+    tag_attachment_value: Optional[str] = Field(default=None, description="")
+
+
 class StarredDetails(AtlanObject):
     """Description"""
 
@@ -444,10 +444,10 @@ class SourceTagAttribute(AtlanObject):
     )
 
 
-DbtJobRun.update_forward_refs()
-
 MCRuleSchedule.update_forward_refs()
 
+DbtJobRun.update_forward_refs()
+
 AwsCloudWatchMetric.update_forward_refs()
 
 Histogram.update_forward_refs()
@@ -456,10 +456,10 @@ class SourceTagAttribute(AtlanObject):
 
 ColumnValueFrequencyMap.update_forward_refs()
 
-SourceTagAttachmentValue.update_forward_refs()
-
 BadgeCondition.update_forward_refs()
 
+SourceTagAttachmentValue.update_forward_refs()
+
 StarredDetails.update_forward_refs()
 
 AwsTag.update_forward_refs()