diff --git a/pyatlan_v9/model/assets/_overlays/cosmos_mongo_db_account.py b/pyatlan_v9/model/assets/_overlays/cosmos_mongo_db_account.py new file mode 100644 index 000000000..a7b409896 --- /dev/null +++ b/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) diff --git a/pyatlan_v9/model/assets/_overlays/cosmos_mongo_db_collection.py b/pyatlan_v9/model/assets/_overlays/cosmos_mongo_db_collection.py new file mode 100644 index 000000000..47d4635fa --- /dev/null +++ b/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) diff --git a/pyatlan_v9/model/assets/_overlays/cosmos_mongo_db_database.py b/pyatlan_v9/model/assets/_overlays/cosmos_mongo_db_database.py new file mode 100644 index 000000000..94440abb2 --- /dev/null +++ b/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) diff --git a/pyatlan_v9/model/assets/cosmos_mongo_db_account.py b/pyatlan_v9/model/assets/cosmos_mongo_db_account.py index 0e75da308..8d5be095b 100644 --- a/pyatlan_v9/model/assets/cosmos_mongo_db_account.py +++ b/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) # ========================================================================= diff --git a/pyatlan_v9/model/assets/cosmos_mongo_db_collection.py b/pyatlan_v9/model/assets/cosmos_mongo_db_collection.py index a336cb69f..aaa4da01b 100644 --- a/pyatlan_v9/model/assets/cosmos_mongo_db_collection.py +++ b/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) # ========================================================================= diff --git a/pyatlan_v9/model/assets/cosmos_mongo_db_database.py b/pyatlan_v9/model/assets/cosmos_mongo_db_database.py index 1d24d87f9..d5ec74eba 100644 --- a/pyatlan_v9/model/assets/cosmos_mongo_db_database.py +++ b/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) # ========================================================================= diff --git a/tests_v9/unit/model/cosmos_mongo_db_test.py b/tests_v9/unit/model/cosmos_mongo_db_test.py new file mode 100644 index 000000000..d548bbb6b --- /dev/null +++ b/tests_v9/unit/model/cosmos_mongo_db_test.py @@ -0,0 +1,210 @@ +# SPDX-License-Identifier: Apache-2.0 +# Copyright 2026 Atlan Pte. Ltd. + +"""Unit tests for the Cosmos MongoDB creators in pyatlan_v9. + +The expected qualifiedNames are the exact strings atlan-cosmosdb-app and +atlan-cosmosnosql-app emit today (FND-3838 census), so a creator change that would +re-key existing Cosmos MongoDB assets fails here. +""" + +import pytest + +from pyatlan_v9.model import ( + CosmosMongoDBAccount, + CosmosMongoDBCollection, + CosmosMongoDBDatabase, +) + +CONNECTION_QN = "default/cosmos/1712345678" +ACCOUNT_NAME = "sales-account" +DATABASE_NAME = "orders" +COLLECTION_NAME = "line_items" + +# {conn}/accounts/{account} +ACCOUNT_QN = "default/cosmos/1712345678/accounts/sales-account" +# {conn}/accounts/{account}/databases/{db} +DATABASE_QN = "default/cosmos/1712345678/accounts/sales-account/databases/orders" +# {conn}/accounts/{account}/databases/{db}/collections/{collection} +COLLECTION_QN = ( + "default/cosmos/1712345678/accounts/sales-account/databases/orders" + "/collections/line_items" +) + + +def test_account_creator_matches_app_qualified_name(): + sut = CosmosMongoDBAccount.creator( + name=ACCOUNT_NAME, connection_qualified_name=CONNECTION_QN + ) + + assert sut.qualified_name == ACCOUNT_QN + assert sut.name == ACCOUNT_NAME + assert sut.connection_qualified_name == CONNECTION_QN + assert sut.connector_name == "cosmos" + assert sut._QUALIFIED_NAME_PATTERN.match(sut.qualified_name) + sut.validate(for_creation=True) + + +@pytest.mark.parametrize("connection_qualified_name", [None, CONNECTION_QN]) +def test_database_creator_matches_app_qualified_name(connection_qualified_name): + sut = CosmosMongoDBDatabase.creator( + name=DATABASE_NAME, + cosmos_mongo_db_account_qualified_name=ACCOUNT_QN, + connection_qualified_name=connection_qualified_name, + ) + + assert sut.qualified_name == DATABASE_QN + assert sut.name == DATABASE_NAME + assert sut.connection_qualified_name == CONNECTION_QN + assert sut.connector_name == "cosmos" + assert sut.cosmos_mongo_db_account_qualified_name == ACCOUNT_QN + assert sut.cosmos_mongo_db_account.unique_attributes == { + "qualifiedName": ACCOUNT_QN + } + assert sut._QUALIFIED_NAME_PATTERN.match(sut.qualified_name) + sut.validate(for_creation=True) + + +@pytest.mark.parametrize("connection_qualified_name", [None, CONNECTION_QN]) +def test_collection_creator_matches_app_qualified_name(connection_qualified_name): + sut = CosmosMongoDBCollection.creator( + name=COLLECTION_NAME, + cosmos_mongo_db_database_qualified_name=DATABASE_QN, + connection_qualified_name=connection_qualified_name, + ) + + assert sut.qualified_name == COLLECTION_QN + assert sut.name == COLLECTION_NAME + assert sut.connection_qualified_name == CONNECTION_QN + assert sut.connector_name == "cosmos" + assert sut.cosmos_mongo_db_database_qualified_name == DATABASE_QN + assert sut.database_name == DATABASE_NAME + assert sut.database_qualified_name == DATABASE_QN + assert sut.cosmos_mongo_db_database.unique_attributes == { + "qualifiedName": DATABASE_QN + } + assert sut._QUALIFIED_NAME_PATTERN.match(sut.qualified_name) + sut.validate(for_creation=True) + + +def test_creators_chain_from_account_to_collection(): + account = CosmosMongoDBAccount.creator( + name=ACCOUNT_NAME, connection_qualified_name=CONNECTION_QN + ) + database = CosmosMongoDBDatabase.creator( + name=DATABASE_NAME, + cosmos_mongo_db_account_qualified_name=account.qualified_name, + ) + collection = CosmosMongoDBCollection.creator( + name=COLLECTION_NAME, + cosmos_mongo_db_database_qualified_name=database.qualified_name, + ) + + assert collection.qualified_name == COLLECTION_QN + + +@pytest.mark.parametrize( + "name, connection_qualified_name, message", + [ + (None, CONNECTION_QN, "name is required"), + (ACCOUNT_NAME, None, "connection_qualified_name is required"), + (" ", CONNECTION_QN, "name"), + (ACCOUNT_NAME, "", "connection_qualified_name"), + ], +) +def test_account_creator_rejects_missing_or_blank_inputs( + name, connection_qualified_name, message +): + with pytest.raises(ValueError, match=message): + CosmosMongoDBAccount.creator( + name=name, connection_qualified_name=connection_qualified_name + ) + + +@pytest.mark.parametrize( + "name, account_qualified_name, message", + [ + (None, ACCOUNT_QN, "name is required"), + (DATABASE_NAME, None, "cosmos_mongo_db_account_qualified_name is required"), + (" ", ACCOUNT_QN, "name"), + (DATABASE_NAME, "", "cosmos_mongo_db_account_qualified_name"), + (DATABASE_NAME, "default/cosmos", "cosmos_mongo_db_account_qualified_name"), + ], +) +def test_database_creator_rejects_missing_or_invalid_inputs( + name, account_qualified_name, message +): + with pytest.raises(ValueError, match=message): + CosmosMongoDBDatabase.creator( + name=name, cosmos_mongo_db_account_qualified_name=account_qualified_name + ) + + +@pytest.mark.parametrize( + "name, database_qualified_name, message", + [ + (None, DATABASE_QN, "name is required"), + ( + COLLECTION_NAME, + None, + "cosmos_mongo_db_database_qualified_name is required", + ), + (" ", DATABASE_QN, "name"), + (COLLECTION_NAME, "", "cosmos_mongo_db_database_qualified_name"), + (COLLECTION_NAME, ACCOUNT_QN, "cosmos_mongo_db_database_qualified_name"), + ], +) +def test_collection_creator_rejects_missing_or_invalid_inputs( + name, database_qualified_name, message +): + with pytest.raises(ValueError, match=message): + CosmosMongoDBCollection.creator( + name=name, + cosmos_mongo_db_database_qualified_name=database_qualified_name, + ) + + +@pytest.mark.parametrize( + "asset_type, kwargs, expected_qn", + [ + ( + CosmosMongoDBAccount, + {"name": ACCOUNT_NAME, "connection_qualified_name": CONNECTION_QN}, + ACCOUNT_QN, + ), + ( + CosmosMongoDBDatabase, + { + "name": DATABASE_NAME, + "cosmos_mongo_db_account_qualified_name": ACCOUNT_QN, + }, + DATABASE_QN, + ), + ( + CosmosMongoDBCollection, + { + "name": COLLECTION_NAME, + "cosmos_mongo_db_database_qualified_name": DATABASE_QN, + }, + COLLECTION_QN, + ), + ], +) +def test_create_alias(asset_type, kwargs, expected_qn): + assert asset_type.create(**kwargs).qualified_name == expected_qn + + +@pytest.mark.parametrize( + "asset_type, qualified_name, name", + [ + (CosmosMongoDBAccount, ACCOUNT_QN, ACCOUNT_NAME), + (CosmosMongoDBDatabase, DATABASE_QN, DATABASE_NAME), + (CosmosMongoDBCollection, COLLECTION_QN, COLLECTION_NAME), + ], +) +def test_create_for_modification_alias(asset_type, qualified_name, name): + sut = asset_type.create_for_modification(qualified_name=qualified_name, name=name) + + assert isinstance(sut, asset_type) + assert sut.qualified_name == qualified_name + assert sut.name == name