FND-3879: add pyatlan_v9 creators for the 2 Dataplex types

pyatlan/model/enums.py

@@ -268,6 +268,7 @@ def get_connector_name(
     API = ("api", AtlanConnectionCategory.API)
     DYNAMODB = ("dynamodb", AtlanConnectionCategory.WAREHOUSE)
     GCS = ("gcs", AtlanConnectionCategory.OBJECT_STORE)
+    DATAPLEX = ("dataplex", AtlanConnectionCategory.SAAS)
     HIVE = ("hive", AtlanConnectionCategory.WAREHOUSE)
     SAPHANA = ("sap-hana", AtlanConnectionCategory.WAREHOUSE)
     ADLS = ("adls", AtlanConnectionCategory.OBJECT_STORE)

pyatlan_v9/model/assets/_overlays/bigquery_routine.py

@@ -0,0 +1,75 @@
+# INTERNAL_IMPORT: from pyatlan.utils import init_guid, validate_required_fields
+
+    @classmethod
+    @init_guid
+    def creator(
+        cls,
+        *,
+        name: str,
+        definition: str,
+        schema_qualified_name: str,
+        schema_name: str | None = None,
+        database_name: str | None = None,
+        database_qualified_name: str | None = None,
+        connection_qualified_name: str | None = None,
+    ) -> "BigqueryRoutine":
+        """
+        Builds the minimal object necessary to create a BigqueryRoutine.
+
+        The qualifiedName is ``{schema_qualified_name}/_routines_/{name}``, the shape
+        BigqueryRoutine assets carry on tenants and that atlan-dataplex-app aspect-sync
+        parses. Current atlan-bigquery-app crawls emit ``Function`` / ``Procedure``
+        instead of BigqueryRoutine.
+        """
+        validate_required_fields(
+            ["name", "definition", "schema_qualified_name"],
+            [name, definition, schema_qualified_name],
+        )
+
+        fields = schema_qualified_name.split("/")
+        if len(fields) != 5:
+            raise ValueError(
+                f"Invalid schema_qualified_name: {schema_qualified_name}. "
+                "Expected format: default/bigquery/connection_id/project/dataset"
+            )
+
+        connector_name = fields[1]
+        connection_qn = (
+            connection_qualified_name or f"{fields[0]}/{fields[1]}/{fields[2]}"
+        )
+        db_name = database_name or fields[3]
+        sch_name = schema_name or fields[4]
+        db_qualified_name = database_qualified_name or f"{connection_qn}/{db_name}"
+
+        return cls(
+            name=name,
+            definition=definition,
+            qualified_name=f"{schema_qualified_name}/_routines_/{name}",
+            database_name=db_name,
+            database_qualified_name=db_qualified_name,
+            schema_name=sch_name,
+            schema_qualified_name=schema_qualified_name,
+            connector_name=connector_name,
+            connection_qualified_name=connection_qn,
+            atlan_schema=RelatedSchema(qualified_name=schema_qualified_name),
+        )
+
+    @classmethod
+    def updater(cls, *, qualified_name: str, name: str) -> "BigqueryRoutine":
+        """Create a BigqueryRoutine instance for update operations."""
+        validate_required_fields(["qualified_name", "name"], [qualified_name, name])
+        return cls(qualified_name=qualified_name, name=name)
+
+    def trim_to_required(self) -> "BigqueryRoutine":
+        """Return only fields required for update operations."""
+        return BigqueryRoutine.updater(
+            qualified_name=self.qualified_name, name=self.name
+        )
+
+    @classmethod
+    def create(cls, **kwargs) -> "BigqueryRoutine":
+        return cls.creator(**kwargs)
+
+    @classmethod
+    def create_for_modification(cls, **kwargs) -> "BigqueryRoutine":
+        return cls.updater(**kwargs)

pyatlan_v9/model/assets/_overlays/gcp_dataplex_aspect_type.py

@@ -0,0 +1,65 @@
+# INTERNAL_IMPORT: from pyatlan.utils import init_guid, validate_required_fields
+
+    @classmethod
+    @init_guid
+    def creator(
+        cls,
+        *,
+        resource_name: str,
+        connection_qualified_name: str,
+    ) -> "GCPDataplexAspectType":
+        """
+        Builds the minimal object necessary to create a GCPDataplexAspectType.
+
+        ``resource_name`` is the full GCP resource name,
+        ``projects/{project}/locations/{location}/aspectTypes/{id}``. The qualifiedName
+        is ``{connection_qualified_name}/aspect-type/{resource_name}`` and the name is
+        the aspect type id, as atlan-dataplex-app emits them.
+        """
+        validate_required_fields(
+            ["resource_name", "connection_qualified_name"],
+            [resource_name, connection_qualified_name],
+        )
+        parts = resource_name.split("/")
+        if (
+            len(parts) != 6
+            or parts[0] != "projects"
+            or parts[2] != "locations"
+            or parts[4] != "aspectTypes"
+            or not all(part.strip() for part in parts)
+        ):
+            raise ValueError(
+                f"Invalid resource_name: {resource_name}. "
+                "Expected format: projects/{project}/locations/{location}/aspectTypes/{id}"
+            )
+        fields = connection_qualified_name.split("/")
+        connector_name = fields[1] if len(fields) > 1 else None
+        return cls(
+            name=parts[5],
+            qualified_name=f"{connection_qualified_name}/aspect-type/{resource_name}",
+            connection_qualified_name=connection_qualified_name,
+            connector_name=connector_name,
+            gcp_dataplex_aspect_type_resource_name=resource_name,
+            gcp_dataplex_aspect_type_project=parts[1],
+            gcp_dataplex_aspect_type_location=parts[3],
+        )
+
+    @classmethod
+    def updater(cls, *, qualified_name: str, name: str) -> "GCPDataplexAspectType":
+        """Create a GCPDataplexAspectType instance for update operations."""
+        validate_required_fields(["qualified_name", "name"], [qualified_name, name])
+        return cls(qualified_name=qualified_name, name=name)
+
+    def trim_to_required(self) -> "GCPDataplexAspectType":
+        """Return only fields required for update operations."""
+        return GCPDataplexAspectType.updater(
+            qualified_name=self.qualified_name, name=self.name
+        )
+
+    @classmethod
+    def create(cls, **kwargs) -> "GCPDataplexAspectType":
+        return cls.creator(**kwargs)
+
+    @classmethod
+    def create_for_modification(cls, **kwargs) -> "GCPDataplexAspectType":
+        return cls.updater(**kwargs)

pyatlan_v9/model/assets/bigquery_routine.py

@@ -26,6 +26,7 @@
 )
 from pyatlan_v9.model.serde import Serde, get_serde
 from pyatlan_v9.model.transform import register_asset
+from pyatlan_v9.utils import init_guid, validate_required_fields
 
 from .airflow_related import RelatedAirflowTask
 from .anomalo_related import RelatedAnomaloCheck
@@ -509,9 +510,7 @@ def __post_init__(self) -> None:
     # SDK Methods
     # =========================================================================
 
-    _QUALIFIED_NAME_PATTERN: ClassVar[re.Pattern] = re.compile(
-        r"^.+/[^/]+/[^/]+/[^/]+$"
-    )
+    _QUALIFIED_NAME_PATTERN: ClassVar[re.Pattern] = re.compile(r"^.+/_routines_/[^/]+$")
 
     def validate(self, for_creation: bool = False) -> None:
         """
@@ -589,6 +588,80 @@ def relate(self) -> "RelatedBigqueryRoutine":
             return RelatedBigqueryRoutine(guid=self.guid)
         return RelatedBigqueryRoutine(qualified_name=self.qualified_name)
 
+    @classmethod
+    @init_guid
+    def creator(
+        cls,
+        *,
+        name: str,
+        definition: str,
+        schema_qualified_name: str,
+        schema_name: str | None = None,
+        database_name: str | None = None,
+        database_qualified_name: str | None = None,
+        connection_qualified_name: str | None = None,
+    ) -> "BigqueryRoutine":
+        """
+        Builds the minimal object necessary to create a BigqueryRoutine.
+
+        The qualifiedName is ``{schema_qualified_name}/_routines_/{name}``, the shape
+        BigqueryRoutine assets carry on tenants and that atlan-dataplex-app aspect-sync
+        parses. Current atlan-bigquery-app crawls emit ``Function`` / ``Procedure``
+        instead of BigqueryRoutine.
+        """
+        validate_required_fields(
+            ["name", "definition", "schema_qualified_name"],
+            [name, definition, schema_qualified_name],
+        )
+
+        fields = schema_qualified_name.split("/")
+        if len(fields) != 5:
+            raise ValueError(
+                f"Invalid schema_qualified_name: {schema_qualified_name}. "
+                "Expected format: default/bigquery/connection_id/project/dataset"
+            )
+
+        connector_name = fields[1]
+        connection_qn = (
+            connection_qualified_name or f"{fields[0]}/{fields[1]}/{fields[2]}"
+        )
+        db_name = database_name or fields[3]
+        sch_name = schema_name or fields[4]
+        db_qualified_name = database_qualified_name or f"{connection_qn}/{db_name}"
+
+        return cls(
+            name=name,
+            definition=definition,
+            qualified_name=f"{schema_qualified_name}/_routines_/{name}",
+            database_name=db_name,
+            database_qualified_name=db_qualified_name,
+            schema_name=sch_name,
+            schema_qualified_name=schema_qualified_name,
+            connector_name=connector_name,
+            connection_qualified_name=connection_qn,
+            atlan_schema=RelatedSchema(qualified_name=schema_qualified_name),
+        )
+
+    @classmethod
+    def updater(cls, *, qualified_name: str, name: str) -> "BigqueryRoutine":
+        """Create a BigqueryRoutine instance for update operations."""
+        validate_required_fields(["qualified_name", "name"], [qualified_name, name])
+        return cls(qualified_name=qualified_name, name=name)
+
+    def trim_to_required(self) -> "BigqueryRoutine":
+        """Return only fields required for update operations."""
+        return BigqueryRoutine.updater(
+            qualified_name=self.qualified_name, name=self.name
+        )
+
+    @classmethod
+    def create(cls, **kwargs) -> "BigqueryRoutine":
+        return cls.creator(**kwargs)
+
+    @classmethod
+    def create_for_modification(cls, **kwargs) -> "BigqueryRoutine":
+        return cls.updater(**kwargs)
+
     # =========================================================================
     # Optimized Serialization Methods (override Asset base class)
     # =========================================================================
@@ -1210,6 +1283,7 @@ def _bigquery_routine_to_nested(
         is_incomplete=bigquery_routine.is_incomplete,
         provenance_type=bigquery_routine.provenance_type,
         home_id=bigquery_routine.home_id,
+        delete_handler=bigquery_routine.delete_handler,
         depth=bigquery_routine.depth,
         immediate_upstream=bigquery_routine.immediate_upstream,
         immediate_downstream=bigquery_routine.immediate_downstream,
@@ -1258,6 +1332,7 @@ def _bigquery_routine_from_nested(nested: BigqueryRoutineNested) -> BigqueryRout
         "is_incomplete": nested.is_incomplete,
         "provenance_type": nested.provenance_type,
         "home_id": nested.home_id,
+        "delete_handler": nested.delete_handler,
         "depth": nested.depth,
         "immediate_upstream": nested.immediate_upstream,
         "immediate_downstream": nested.immediate_downstream,

pyatlan_v9/model/assets/gcp_dataplex_aspect_type.py

@@ -14,6 +14,7 @@
 
 from __future__ import annotations
 
+import re
 from typing import Any, ClassVar, Dict, List, Union
 
 from msgspec import UNSET, UnsetType
@@ -24,6 +25,7 @@
 )
 from pyatlan_v9.model.serde import Serde, get_serde
 from pyatlan_v9.model.transform import register_asset
+from pyatlan_v9.utils import init_guid, validate_required_fields
 
 from .anomalo_related import RelatedAnomaloCheck
 from .app_related import RelatedApplication, RelatedApplicationField
@@ -229,6 +231,10 @@ def __post_init__(self) -> None:
     # SDK Methods
     # =========================================================================
 
+    _QUALIFIED_NAME_PATTERN: ClassVar[re.Pattern] = re.compile(
+        r"^.+/aspect-type/projects/[^/]+/locations/[^/]+/aspectTypes/[^/]+$"
+    )
+
     def validate(self, for_creation: bool = False) -> None:
         """
         Dry-run validation of this GCPDataplexAspectType instance.
@@ -253,6 +259,11 @@ def validate(self, for_creation: bool = False) -> None:
             errors.append("name is required")
         if self.qualified_name is UNSET or self.qualified_name is None:
             errors.append("qualified_name is required")
+        elif not self._QUALIFIED_NAME_PATTERN.match(self.qualified_name):
+            errors.append(
+                f"qualified_name '{self.qualified_name}' does not match expected "
+                f"pattern: {self._QUALIFIED_NAME_PATTERN.pattern}"
+            )
         if errors:
             raise ValueError(f"GCPDataplexAspectType validation failed: {errors}")
 
@@ -285,6 +296,70 @@ def relate(self) -> "RelatedGCPDataplexAspectType":
             return RelatedGCPDataplexAspectType(guid=self.guid)
         return RelatedGCPDataplexAspectType(qualified_name=self.qualified_name)
 
+    @classmethod
+    @init_guid
+    def creator(
+        cls,
+        *,
+        resource_name: str,
+        connection_qualified_name: str,
+    ) -> "GCPDataplexAspectType":
+        """
+        Builds the minimal object necessary to create a GCPDataplexAspectType.
+
+        ``resource_name`` is the full GCP resource name,
+        ``projects/{project}/locations/{location}/aspectTypes/{id}``. The qualifiedName
+        is ``{connection_qualified_name}/aspect-type/{resource_name}`` and the name is
+        the aspect type id, as atlan-dataplex-app emits them.
+        """
+        validate_required_fields(
+            ["resource_name", "connection_qualified_name"],
+            [resource_name, connection_qualified_name],
+        )
+        parts = resource_name.split("/")
+        if (
+            len(parts) != 6
+            or parts[0] != "projects"
+            or parts[2] != "locations"
+            or parts[4] != "aspectTypes"
+            or not all(part.strip() for part in parts)
+        ):
+            raise ValueError(
+                f"Invalid resource_name: {resource_name}. "
+                "Expected format: projects/{project}/locations/{location}/aspectTypes/{id}"
+            )
+        fields = connection_qualified_name.split("/")
+        connector_name = fields[1] if len(fields) > 1 else None
+        return cls(
+            name=parts[5],
+            qualified_name=f"{connection_qualified_name}/aspect-type/{resource_name}",
+            connection_qualified_name=connection_qualified_name,
+            connector_name=connector_name,
+            gcp_dataplex_aspect_type_resource_name=resource_name,
+            gcp_dataplex_aspect_type_project=parts[1],
+            gcp_dataplex_aspect_type_location=parts[3],
+        )
+
+    @classmethod
+    def updater(cls, *, qualified_name: str, name: str) -> "GCPDataplexAspectType":
+        """Create a GCPDataplexAspectType instance for update operations."""
+        validate_required_fields(["qualified_name", "name"], [qualified_name, name])
+        return cls(qualified_name=qualified_name, name=name)
+
+    def trim_to_required(self) -> "GCPDataplexAspectType":
+        """Return only fields required for update operations."""
+        return GCPDataplexAspectType.updater(
+            qualified_name=self.qualified_name, name=self.name
+        )
+
+    @classmethod
+    def create(cls, **kwargs) -> "GCPDataplexAspectType":
+        return cls.creator(**kwargs)
+
+    @classmethod
+    def create_for_modification(cls, **kwargs) -> "GCPDataplexAspectType":
+        return cls.updater(**kwargs)
+
     # =========================================================================
     # Optimized Serialization Methods (override Asset base class)
     # =========================================================================
@@ -607,6 +682,7 @@ def _gcp_dataplex_aspect_type_to_nested(
         is_incomplete=gcp_dataplex_aspect_type.is_incomplete,
         provenance_type=gcp_dataplex_aspect_type.provenance_type,
         home_id=gcp_dataplex_aspect_type.home_id,
+        delete_handler=gcp_dataplex_aspect_type.delete_handler,
         depth=gcp_dataplex_aspect_type.depth,
         immediate_upstream=gcp_dataplex_aspect_type.immediate_upstream,
         immediate_downstream=gcp_dataplex_aspect_type.immediate_downstream,
@@ -657,6 +733,7 @@ def _gcp_dataplex_aspect_type_from_nested(
         "is_incomplete": nested.is_incomplete,
         "provenance_type": nested.provenance_type,
         "home_id": nested.home_id,
+        "delete_handler": nested.delete_handler,
         "depth": nested.depth,
         "immediate_upstream": nested.immediate_upstream,
         "immediate_downstream": nested.immediate_downstream,