Added generator files

Author: vaibhavatlan

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

@@ -0,0 +1,49 @@
+
+    @overload
+    @classmethod
+    def creator(
+        cls,
+        *,
+        name: str,
+        ai_application_version: str,
+        ai_application_development_stage: AIApplicationDevelopmentStage,
+    ) -> AIApplication: ...
+
+    @overload
+    @classmethod
+    def creator(
+        cls,
+        *,
+        name: str,
+        ai_application_version: str,
+        ai_application_development_stage: AIApplicationDevelopmentStage,
+        owner_groups: Set[str],
+        owner_users: Set[str],
+        models: List[AIModel],
+    ) -> AIApplication: ...
+
+    @classmethod
+    @init_guid
+    def creator(
+        cls,
+        *,
+        name: str,
+        ai_application_version: str,
+        ai_application_development_stage: AIApplicationDevelopmentStage,
+        owner_groups: Optional[Set[str]] = set(),
+        owner_users: Optional[Set[str]] = set(),
+        models: Optional[List[AIModel]] = [],
+    ) -> AIApplication:
+        validate_required_fields(
+            ["name", "ai_application_version", "ai_application_development_stage"],
+            [name, ai_application_version, ai_application_development_stage],
+        )
+        attributes = AIApplication.Attributes.creator(
+            name=name,
+            ai_application_version=ai_application_version,
+            ai_application_development_stage=ai_application_development_stage,
+            owner_groups=owner_groups,
+            owner_users=owner_users,
+            models=models,
+        )
+        return cls(attributes=attributes)

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

@@ -0,0 +1,110 @@
+
+    @overload
+    @classmethod
+    def creator(
+        cls,
+        *,
+        name: str,
+        ai_model_status: AIModelStatus,
+    ) -> AIModel: ...
+
+    @overload
+    @classmethod
+    def creator(
+        cls,
+        *,
+        name: str,
+        ai_model_status: AIModelStatus,
+        owner_groups: Set[str],
+        owner_users: Set[str],
+        ai_model_version: str,
+    ) -> AIModel: ...
+
+    @classmethod
+    @init_guid
+    def creator(
+        cls,
+        *,
+        name: str,
+        ai_model_status: AIModelStatus,
+        owner_groups: Optional[Set[str]] = set(),
+        owner_users: Optional[Set[str]] = set(),
+        ai_model_version: Optional[str] = None,
+    ) -> AIModel:
+        validate_required_fields(
+            ["name", "ai_model_status"],
+            [name, ai_model_status],
+        )
+        attributes = AIModel.Attributes.creator(
+            name=name,
+            ai_model_status=ai_model_status,
+            owner_groups=owner_groups,
+            owner_users=owner_users,
+            ai_model_version=ai_model_version,
+        )
+        return cls(attributes=attributes)
+
+    @classmethod
+    def processes_creator(
+        cls,
+        ai_model: AIModel,
+        dataset_dict: Dict[AIDatasetType, list],
+    ) -> List[Process]:
+        """
+        Creates a list of Process objects representing the relationships between an AI model and its datasets.
+
+        :param ai_model: the AI model for which to create processes
+        :param dataset_dict: dictionary mapping AI dataset types to lists of assets
+        :returns: list of Process objects representing the AI model's data lineage
+        :raises ValueError: when the AI model is missing required attributes (guid or name)
+        """
+        if not ai_model.guid or not ai_model.name:
+            raise ValueError("AI model must have both guid and name attributes")
+        process_list = []
+        for key, value_list in dataset_dict.items():
+            for value in value_list:
+                asset_type = Asset._convert_to_real_type_(value)
+                if key == AIDatasetType.OUTPUT:
+                    process_name = f"{ai_model.name} -> {value.name}"
+                    process_created = Process.creator(
+                        name=process_name,
+                        connection_qualified_name="default/ai/dataset",
+                        inputs=[AIModel.ref_by_guid(guid=ai_model.guid)],
+                        outputs=[asset_type.ref_by_guid(guid=value.guid)],  # type: ignore
+                        extra_hash_params={key.value},
+                    )
+                    process_created.ai_dataset_type = key
+                else:
+                    process_name = f"{value.name} -> {ai_model.name}"
+                    process_created = Process.creator(
+                        name=process_name,
+                        connection_qualified_name="default/ai/dataset",
+                        inputs=[asset_type.ref_by_guid(guid=value.guid)],  # type: ignore
+                        outputs=[AIModel.ref_by_guid(guid=ai_model.guid)],
+                        extra_hash_params={key.value},
+                    )
+                    process_created.ai_dataset_type = key
+                process_list.append(process_created)
+
+        return process_list
+
+    @classmethod
+    def processes_batch_save(cls, client, process_list: List[Process]) -> List:
+        """
+        Saves a list of Process objects to Atlan in batches to optimize performance.
+        We save the processes in batches of 20.
+
+        :param client: Atlan client instance for making API calls
+        :param process_list: list of Process objects to save
+        :returns: list of API responses from each batch save operation
+        """
+        batch_size = 20
+        total_processes = len(process_list)
+        responses = []
+
+        for i in range(0, total_processes, batch_size):
+            batch = process_list[i : i + batch_size]
+            response = client.asset.save(batch)
+            responses.append(response)
+
+        return responses

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

@@ -9,6 +9,7 @@
         outputs: List["Catalog"],
         process_id: Optional[str] = None,
         parent: Optional[Process] = None,
+        extra_hash_params: Optional[Set[str]] = set(),
     ) -> Process:
         return Process(
             attributes=Process.Attributes.create(
@@ -18,6 +19,7 @@
                 inputs=inputs,
                 outputs=outputs,
                 parent=parent,
+                extra_hash_params=extra_hash_params,
             )
         )
 

pyatlan/generator/templates/methods/attribute/a_i_application.jinja2

@@ -0,0 +1,30 @@
+
+        @classmethod
+        @init_guid
+        def creator(
+            cls,
+            *,
+            name: str,
+            ai_application_version: str,
+            ai_application_development_stage: AIApplicationDevelopmentStage,
+            owner_groups: Optional[Set[str]] = set(),
+            owner_users: Optional[Set[str]] = set(),
+            models: Optional[List[AIModel]] = [],
+        ) -> AIApplication.Attributes:
+            validate_required_fields(
+                ["name", "ai_application_version", "ai_application_development_stage"],
+                [name, ai_application_version, ai_application_development_stage],
+            )
+            name_camel_case = to_camel_case(name)
+            return AIApplication.Attributes(
+                name=name,
+                qualified_name=f"default/ai/aiapplication/{name_camel_case}",
+                connector_name=AtlanConnectorType.AI.value,
+                ai_application_version=ai_application_version,
+                ai_application_development_stage=ai_application_development_stage,
+                owner_groups=owner_groups,
+                owner_users=owner_users,
+                models=models,
+                certificate_status=CertificateStatus.DRAFT,
+                asset_cover_image="/assets/default-product-cover-DeQonY47.webp",
+            )

pyatlan/generator/templates/methods/attribute/a_i_model.jinja2

@@ -0,0 +1,27 @@
+
+        @classmethod
+        @init_guid
+        def creator(
+            cls,
+            *,
+            name: str,
+            ai_model_status: AIModelStatus,
+            owner_groups: Optional[Set[str]] = set(),
+            owner_users: Optional[Set[str]] = set(),
+            ai_model_version: Optional[str] = None,
+        ) -> AIModel.Attributes:
+            validate_required_fields(
+                ["name", "ai_model_status"],
+                [name, ai_model_status],
+            )
+            name_camel_case = to_camel_case(name)
+            return AIModel.Attributes(
+                name=name,
+                qualified_name=f"default/ai/aiapplication/{name_camel_case}",
+                connector_name=AtlanConnectorType.AI.value,
+                ai_model_status=ai_model_status,
+                ai_model_version=ai_model_version,
+                owner_groups=owner_groups,
+                owner_users=owner_users,
+                asset_cover_image="/assets/default-product-cover-DeQonY47.webp",
+            )

pyatlan/generator/templates/methods/attribute/process.jinja2

@@ -7,6 +7,7 @@
             outputs: List["Catalog"],
             parent: Optional["Process"] = None,
             process_id: Optional[str] = None,
+            extra_hash_params: Optional[Set[str]] = set(),
         ) -> str:
             def append_relationship(output: StringIO, relationship: Asset):
                 if relationship.guid:
@@ -29,6 +30,11 @@
                 append_relationship(buffer, parent)
             append_relationships(buffer, inputs)
             append_relationships(buffer, outputs)
+            # Handles edge case where identical name, connection, input, and output caused hash collisions,
+            # resulting in duplicate qualified names and backend skipping process creation.
+            if extra_hash_params:
+                for param in extra_hash_params:
+                    buffer.write(param)
             ret_value = hashlib.md5(  # noqa: S303, S324
                 buffer.getvalue().encode()
             ).hexdigest()
@@ -45,6 +51,7 @@
             outputs: List["Catalog"],
             process_id: Optional[str] = None,
             parent: Optional[Process] = None,
+            extra_hash_params: Optional[Set[str]] = set(),
         ) -> Process.Attributes:
             qualified_name = Process.Attributes.generate_qualified_name(
                 name=name,
@@ -53,6 +60,7 @@
                 inputs=inputs,
                 outputs=outputs,
                 parent=parent,
+                extra_hash_params=extra_hash_params,
             )
             connector_name = connection_qualified_name.split("/")[1]
             return Process.Attributes(