FND-3847: add pyatlan_v9 creators for the 3 Cosmos MongoDB types

pyatlan_v9/model/assets/_overlays/cosmos_mongo_db_account.py

@@ -0,0 +1,30 @@
+# INTERNAL_IMPORT: from pyatlan.utils import init_guid, validate_required_fields
+
+    @classmethod
+    @init_guid
+    def creator(
+        cls,
+        *,
+        name: str,
+        connection_qualified_name: str,
+    ) -> "CosmosMongoDBAccount":
+        validate_required_fields(
+            ["name", "connection_qualified_name"],
+            [name, connection_qualified_name],
+        )
+        fields = connection_qualified_name.split("/")
+        connector_name = fields[1] if len(fields) > 1 else None
+        return cls(
+            name=name,
+            qualified_name=f"{connection_qualified_name}/accounts/{name}",
+            connection_qualified_name=connection_qualified_name,
+            connector_name=connector_name,
+        )
+
+    @classmethod
+    def create(cls, **kwargs) -> "CosmosMongoDBAccount":
+        return cls.creator(**kwargs)
+
+    @classmethod
+    def create_for_modification(cls, **kwargs) -> "CosmosMongoDBAccount":
+        return cls.updater(**kwargs)

pyatlan_v9/model/assets/_overlays/cosmos_mongo_db_collection.py

@@ -0,0 +1,44 @@
+# INTERNAL_IMPORT: from pyatlan.utils import init_guid, validate_required_fields
+
+    @classmethod
+    @init_guid
+    def creator(
+        cls,
+        *,
+        name: str,
+        cosmos_mongo_db_database_qualified_name: str,
+        connection_qualified_name: str | None = None,
+    ) -> "CosmosMongoDBCollection":
+        validate_required_fields(
+            ["name", "cosmos_mongo_db_database_qualified_name"],
+            [name, cosmos_mongo_db_database_qualified_name],
+        )
+        fields = cosmos_mongo_db_database_qualified_name.split("/")
+        if not connection_qualified_name:
+            if len(fields) < 7:
+                raise ValueError("cosmos_mongo_db_database_qualified_name is invalid")
+            connection_qualified_name = "/".join(fields[:3])
+        connection_fields = connection_qualified_name.split("/")
+        connector_name = connection_fields[1] if len(connection_fields) > 1 else None
+        return cls(
+            name=name,
+            qualified_name=f"{cosmos_mongo_db_database_qualified_name}/collections/{name}",
+            connection_qualified_name=connection_qualified_name,
+            connector_name=connector_name,
+            cosmos_mongo_db_database_qualified_name=cosmos_mongo_db_database_qualified_name,
+            database_name=fields[-1],
+            database_qualified_name=cosmos_mongo_db_database_qualified_name,
+            cosmos_mongo_db_database=RelatedCosmosMongoDBDatabase(
+                unique_attributes={
+                    "qualifiedName": cosmos_mongo_db_database_qualified_name
+                }
+            ),
+        )
+
+    @classmethod
+    def create(cls, **kwargs) -> "CosmosMongoDBCollection":
+        return cls.creator(**kwargs)
+
+    @classmethod
+    def create_for_modification(cls, **kwargs) -> "CosmosMongoDBCollection":
+        return cls.updater(**kwargs)

pyatlan_v9/model/assets/_overlays/cosmos_mongo_db_database.py

@@ -0,0 +1,42 @@
+# INTERNAL_IMPORT: from pyatlan.utils import init_guid, validate_required_fields
+
+    @classmethod
+    @init_guid
+    def creator(
+        cls,
+        *,
+        name: str,
+        cosmos_mongo_db_account_qualified_name: str,
+        connection_qualified_name: str | None = None,
+    ) -> "CosmosMongoDBDatabase":
+        validate_required_fields(
+            ["name", "cosmos_mongo_db_account_qualified_name"],
+            [name, cosmos_mongo_db_account_qualified_name],
+        )
+        if not connection_qualified_name:
+            fields = cosmos_mongo_db_account_qualified_name.split("/")
+            if len(fields) < 5:
+                raise ValueError("cosmos_mongo_db_account_qualified_name is invalid")
+            connection_qualified_name = "/".join(fields[:3])
+        fields = connection_qualified_name.split("/")
+        connector_name = fields[1] if len(fields) > 1 else None
+        return cls(
+            name=name,
+            qualified_name=f"{cosmos_mongo_db_account_qualified_name}/databases/{name}",
+            connection_qualified_name=connection_qualified_name,
+            connector_name=connector_name,
+            cosmos_mongo_db_account_qualified_name=cosmos_mongo_db_account_qualified_name,
+            cosmos_mongo_db_account=RelatedCosmosMongoDBAccount(
+                unique_attributes={
+                    "qualifiedName": cosmos_mongo_db_account_qualified_name
+                }
+            ),
+        )
+
+    @classmethod
+    def create(cls, **kwargs) -> "CosmosMongoDBDatabase":
+        return cls.creator(**kwargs)
+
+    @classmethod
+    def create_for_modification(cls, **kwargs) -> "CosmosMongoDBDatabase":
+        return cls.updater(**kwargs)

pyatlan_v9/model/assets/cosmos_mongo_db_account.py

@@ -14,6 +14,7 @@
 
 from __future__ import annotations
 
+import re
 from typing import Any, ClassVar, List, Union
 
 import msgspec
@@ -25,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
@@ -338,6 +340,8 @@ def __post_init__(self) -> None:
     # SDK Methods
     # =========================================================================
 
+    _QUALIFIED_NAME_PATTERN: ClassVar[re.Pattern] = re.compile(r"^.+/accounts/[^/]+$")
+
     def validate(self, for_creation: bool = False) -> None:
         """
         Dry-run validation of this CosmosMongoDBAccount instance.
@@ -362,6 +366,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"CosmosMongoDBAccount validation failed: {errors}")
 
@@ -394,6 +403,35 @@ def relate(self) -> "RelatedCosmosMongoDBAccount":
             return RelatedCosmosMongoDBAccount(guid=self.guid)
         return RelatedCosmosMongoDBAccount(qualified_name=self.qualified_name)
 
+    @classmethod
+    @init_guid
+    def creator(
+        cls,
+        *,
+        name: str,
+        connection_qualified_name: str,
+    ) -> "CosmosMongoDBAccount":
+        validate_required_fields(
+            ["name", "connection_qualified_name"],
+            [name, connection_qualified_name],
+        )
+        fields = connection_qualified_name.split("/")
+        connector_name = fields[1] if len(fields) > 1 else None
+        return cls(
+            name=name,
+            qualified_name=f"{connection_qualified_name}/accounts/{name}",
+            connection_qualified_name=connection_qualified_name,
+            connector_name=connector_name,
+        )
+
+    @classmethod
+    def create(cls, **kwargs) -> "CosmosMongoDBAccount":
+        return cls.creator(**kwargs)
+
+    @classmethod
+    def create_for_modification(cls, **kwargs) -> "CosmosMongoDBAccount":
+        return cls.updater(**kwargs)
+
     # =========================================================================
     # Optimized Serialization Methods (override Asset base class)
     # =========================================================================

pyatlan_v9/model/assets/cosmos_mongo_db_collection.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
@@ -677,7 +678,7 @@ def __post_init__(self) -> None:
     # =========================================================================
 
     _QUALIFIED_NAME_PATTERN: ClassVar[re.Pattern] = re.compile(
-        r"^.+/[^/]+/[^/]+/[^/]+$"
+        r"^.+/collections/[^/]+$"
     )
 
     def validate(self, for_creation: bool = False) -> None:
@@ -756,6 +757,49 @@ def relate(self) -> "RelatedCosmosMongoDBCollection":
             return RelatedCosmosMongoDBCollection(guid=self.guid)
         return RelatedCosmosMongoDBCollection(qualified_name=self.qualified_name)
 
+    @classmethod
+    @init_guid
+    def creator(
+        cls,
+        *,
+        name: str,
+        cosmos_mongo_db_database_qualified_name: str,
+        connection_qualified_name: str | None = None,
+    ) -> "CosmosMongoDBCollection":
+        validate_required_fields(
+            ["name", "cosmos_mongo_db_database_qualified_name"],
+            [name, cosmos_mongo_db_database_qualified_name],
+        )
+        fields = cosmos_mongo_db_database_qualified_name.split("/")
+        if not connection_qualified_name:
+            if len(fields) < 7:
+                raise ValueError("cosmos_mongo_db_database_qualified_name is invalid")
+            connection_qualified_name = "/".join(fields[:3])
+        connection_fields = connection_qualified_name.split("/")
+        connector_name = connection_fields[1] if len(connection_fields) > 1 else None
+        return cls(
+            name=name,
+            qualified_name=f"{cosmos_mongo_db_database_qualified_name}/collections/{name}",
+            connection_qualified_name=connection_qualified_name,
+            connector_name=connector_name,
+            cosmos_mongo_db_database_qualified_name=cosmos_mongo_db_database_qualified_name,
+            database_name=fields[-1],
+            database_qualified_name=cosmos_mongo_db_database_qualified_name,
+            cosmos_mongo_db_database=RelatedCosmosMongoDBDatabase(
+                unique_attributes={
+                    "qualifiedName": cosmos_mongo_db_database_qualified_name
+                }
+            ),
+        )
+
+    @classmethod
+    def create(cls, **kwargs) -> "CosmosMongoDBCollection":
+        return cls.creator(**kwargs)
+
+    @classmethod
+    def create_for_modification(cls, **kwargs) -> "CosmosMongoDBCollection":
+        return cls.updater(**kwargs)
+
     # =========================================================================
     # Optimized Serialization Methods (override Asset base class)
     # =========================================================================

pyatlan_v9/model/assets/cosmos_mongo_db_database.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
@@ -484,7 +485,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"^.+/databases/[^/]+$")
 
     def validate(self, for_creation: bool = False) -> None:
         """
@@ -556,6 +557,47 @@ def relate(self) -> "RelatedCosmosMongoDBDatabase":
             return RelatedCosmosMongoDBDatabase(guid=self.guid)
         return RelatedCosmosMongoDBDatabase(qualified_name=self.qualified_name)
 
+    @classmethod
+    @init_guid
+    def creator(
+        cls,
+        *,
+        name: str,
+        cosmos_mongo_db_account_qualified_name: str,
+        connection_qualified_name: str | None = None,
+    ) -> "CosmosMongoDBDatabase":
+        validate_required_fields(
+            ["name", "cosmos_mongo_db_account_qualified_name"],
+            [name, cosmos_mongo_db_account_qualified_name],
+        )
+        if not connection_qualified_name:
+            fields = cosmos_mongo_db_account_qualified_name.split("/")
+            if len(fields) < 5:
+                raise ValueError("cosmos_mongo_db_account_qualified_name is invalid")
+            connection_qualified_name = "/".join(fields[:3])
+        fields = connection_qualified_name.split("/")
+        connector_name = fields[1] if len(fields) > 1 else None
+        return cls(
+            name=name,
+            qualified_name=f"{cosmos_mongo_db_account_qualified_name}/databases/{name}",
+            connection_qualified_name=connection_qualified_name,
+            connector_name=connector_name,
+            cosmos_mongo_db_account_qualified_name=cosmos_mongo_db_account_qualified_name,
+            cosmos_mongo_db_account=RelatedCosmosMongoDBAccount(
+                unique_attributes={
+                    "qualifiedName": cosmos_mongo_db_account_qualified_name
+                }
+            ),
+        )
+
+    @classmethod
+    def create(cls, **kwargs) -> "CosmosMongoDBDatabase":
+        return cls.creator(**kwargs)
+
+    @classmethod
+    def create_for_modification(cls, **kwargs) -> "CosmosMongoDBDatabase":
+        return cls.updater(**kwargs)
+
     # =========================================================================
     # Optimized Serialization Methods (override Asset base class)
     # =========================================================================