Update jinja templates

pyatlan/generator/templates/entity.jinja2

@@ -10,15 +10,10 @@ from io import StringIO
 from typing import Any, ClassVar, Dict, List, Optional, TypeVar
 from urllib.parse import quote, unquote
 
-from pydantic import Field, StrictStr, root_validator, validator
+from pydantic import Field, PrivateAttr, StrictStr, root_validator, validator
 
-from pyatlan.model.core import (
-    Announcement,
-    AtlanObject,
-    Classification,
-    CustomMetadata,
-    Meaning,
-)
+from pyatlan.model.core import Announcement, AtlanObject, Classification, Meaning
+from pyatlan.model.custom_metadata import CustomMetadataDict, CustomMetadataProxy
 from pyatlan.model.enums import (
     ADLSAccessTier,
     ADLSAccountStatus,
@@ -90,9 +85,16 @@ SelfAsset = TypeVar("SelfAsset", bound="Asset")
 class {{ entity_def.name }}({{super_classes[0]}} {%- if "Asset" in super_classes  %}, type_name='{{ entity_def.name }}'{% endif %}):
     """Description"""
 {% if entity_def.name == "Referenceable" %}
-    def __init__(__pydantic_self__, **data:Any)->None:
+    def __init__(__pydantic_self__, **data: Any) -> None:
         super().__init__(**data)
         __pydantic_self__.__fields_set__.update(["attributes", "type_name"])
+        __pydantic_self__._metadata_proxy = CustomMetadataProxy(
+            __pydantic_self__.business_attributes
+        )
+
+    def json(self, *args, **kwargs) -> str:
+        self.business_attributes = self._metadata_proxy.business_attributes
+        return super().json(**kwargs)
 {% endif %}
     def __setattr__(self, name, value):
             if name in {{ entity_def.name }}._convience_properties:
@@ -117,6 +119,7 @@ class {{ entity_def.name }}({{super_classes[0]}} {%- if "Asset" in super_classes
         def validate_required(self):
             pass
 
+    _metadata_proxy: CustomMetadataProxy = PrivateAttr()
     attributes: '{{entity_def.name}}.Attributes' = Field(
         default_factory = lambda : {{entity_def.name}}.Attributes(),
         description='Map of attributes in the instance and their values. The specific keys of this map will vary '
@@ -219,51 +222,15 @@ 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_custom_metadata(self, name: str) -> CustomMetadata:
-        from pyatlan.cache.custom_metadata_cache import CustomMetadataCache
+    def get_custom_metadata(self, name: str) -> CustomMetadataDict:
+        return self._metadata_proxy.get_custom_metadata(name=name)
 
-        ba_id = CustomMetadataCache.get_id_for_name(name)
-        if ba_id is None:
-            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")
-                and a_type._meta_data_type_name == name
-            ):
-                break
-        else:
-            raise ValueError(
-                f"Custom metadata attributes {name} are not applicable to {self.type_name}"
-            )
-        if ba_type := CustomMetadataCache.get_type_for_id(ba_id):
-            return (
-                ba_type(self.business_attributes[ba_id])
-                if self.business_attributes and ba_id in self.business_attributes
-                else ba_type()
-            )
-        else:
-            raise ValueError(
-                f"Custom metadata attributes {name} are not applicable to {self.type_name}"
-            )
+    def set_custom_metadata(self, custom_metadata: CustomMetadataDict):
+        return self._metadata_proxy.set_custom_metadata(custom_metadata=custom_metadata)
 
-    def set_custom_metadata(self, custom_metadata: CustomMetadata) -> None:
-        from pyatlan.cache.custom_metadata_cache import CustomMetadataCache
+    def flush_custom_metadata(self):
+        self.business_attributes = self._metadata_proxy.business_attributes
 
-        if not isinstance(custom_metadata, CustomMetadata):
-            raise ValueError(
-                "business_attributes must be an instance of CustomMetadata"
-            )
-        if (
-            type(custom_metadata)
-            not in CustomMetadataCache.types_by_asset[self.type_name]
-        ):
-            raise ValueError(
-                f"Business attributes {custom_metadata._meta_data_type_name} are not applicable to {self.type_name}"
-            )
-        ba_dict = dict(custom_metadata)
-        if not self.business_attributes:
-            self.business_attributes = {}
-        self.business_attributes[custom_metadata._meta_data_type_id] = ba_dict
 
 {%- else %}
     {%- if entity_def.name == "Asset"  %}

pyatlan/generator/templates/structs.jinja2

@@ -43,6 +43,6 @@ class {{struct.name}}(AtlanObject):
     {% endif %}
     {%- for attribute_def in struct.attribute_defs %}
     {%- set type = attribute_def.type_name | get_type %}
-    {{attribute_def.name | to_snake_case }}: {% if attribute_def.is_optional %}Optional[{% endif %}{{type}}{% if attribute_def.is_optional %}]{% endif %} = Field(None, description='' , alias='{{attribute_def.name}}')
+    {{attribute_def.name | to_snake_case }}: {% if attribute_def.is_optional %}Optional[{% endif %}'{{type}}'{% if attribute_def.is_optional %}]{% endif %} = Field(None, description='' , alias='{{attribute_def.name}}')
     {%- endfor %}
 {% endfor %}

pyatlan/model/assets.py

@@ -90,15 +90,15 @@ def __init__(__pydantic_self__, **data: Any) -> None:
             __pydantic_self__.business_attributes
         )
 
+    def json(self, *args, **kwargs) -> str:
+        self.business_attributes = self._metadata_proxy.business_attributes
+        return super().json(**kwargs)
+
     def __setattr__(self, name, value):
         if name in Referenceable._convience_properties:
             return object.__setattr__(self, name, value)
         super().__setattr__(name, value)
 
-    def json(self, *args, **kwargs) -> str:
-        self.business_attributes = self._metadata_proxy.business_attributes
-        return super().json(**kwargs)
-
     _convience_properties: ClassVar[list[str]] = [
         "qualified_name",
         "replicated_from",
@@ -6456,7 +6456,7 @@ class Attributes(ObjectStore.Attributes):
     )
 
 
-class ADLS(ObjectStore):
+class ADLS(Azure):
     """Description"""
 
     def __setattr__(self, name, value):
@@ -6470,6 +6470,8 @@ def __setattr__(self, name, value):
         "azure_location",
         "adls_account_secondary_location",
         "azure_tags",
+        "input_to_processes",
+        "output_from_processes",
     ]
 
     @property
@@ -6526,6 +6528,26 @@ def azure_tags(self, azure_tags: Optional[list[AzureTag]]):
             self.attributes = self.Attributes()
         self.attributes.azure_tags = azure_tags
 
+    @property
+    def input_to_processes(self) -> Optional[list[Process]]:
+        return self.attributes.input_to_processes
+
+    @input_to_processes.setter
+    def input_to_processes(self, input_to_processes: Optional[list[Process]]):
+        if self.attributes is None:
+            self.attributes = self.Attributes()
+        self.attributes.input_to_processes = input_to_processes
+
+    @property
+    def output_from_processes(self) -> Optional[list[Process]]:
+        return self.attributes.output_from_processes
+
+    @output_from_processes.setter
+    def output_from_processes(self, output_from_processes: Optional[list[Process]]):
+        if self.attributes is None:
+            self.attributes = self.Attributes()
+        self.attributes.output_from_processes = output_from_processes
+
     type_name: str = Field("ADLS", allow_mutation=False)
 
     @validator("type_name")
@@ -6534,7 +6556,7 @@ def validate_type_name(cls, v):
             raise ValueError("must be ADLS")
         return v
 
-    class Attributes(ObjectStore.Attributes):
+    class Attributes(Azure.Attributes):
         adls_account_qualified_name: Optional[str] = Field(
             None, description="", alias="adlsAccountQualifiedName"
         )
@@ -6550,6 +6572,12 @@ class Attributes(ObjectStore.Attributes):
         azure_tags: Optional[list[AzureTag]] = Field(
             None, description="", alias="azureTags"
         )
+        input_to_processes: Optional[list[Process]] = Field(
+            None, description="", alias="inputToProcesses"
+        )  # relationship
+        output_from_processes: Optional[list[Process]] = Field(
+            None, description="", alias="outputFromProcesses"
+        )  # relationship
 
     attributes: "ADLS.Attributes" = Field(
         default_factory=lambda: ADLS.Attributes(),