diff --git a/pyatlan/model/enums.py b/pyatlan/model/enums.py index 28b30f197..43589a6c0 100644 --- a/pyatlan/model/enums.py +++ b/pyatlan/model/enums.py @@ -247,6 +247,7 @@ def get_connector_name( BIGQUERY = ("bigquery", AtlanConnectionCategory.WAREHOUSE) LOOKER = ("looker", AtlanConnectionCategory.BI) METABASE = ("metabase", AtlanConnectionCategory.BI) + COGNOS = ("cognos", AtlanConnectionCategory.BI) SALESFORCE = ("salesforce", AtlanConnectionCategory.SAAS) DATAVERSE = ("dataverse", AtlanConnectionCategory.SAAS) MYSQL = ("mysql", AtlanConnectionCategory.WAREHOUSE) diff --git a/pyatlan_v9/model/assets/_overlays/cognos_column.py b/pyatlan_v9/model/assets/_overlays/cognos_column.py new file mode 100644 index 000000000..6859e4996 --- /dev/null +++ b/pyatlan_v9/model/assets/_overlays/cognos_column.py @@ -0,0 +1,32 @@ +# INTERNAL_IMPORT: from pyatlan.utils import init_guid, validate_required_fields + + @classmethod + @init_guid + def creator( + cls, + *, + name: str, + connection_qualified_name: str, + cognos_id: str, + ) -> "CognosColumn": + validate_required_fields( + ["name", "connection_qualified_name", "cognos_id"], + [name, connection_qualified_name, cognos_id], + ) + 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}/column/{cognos_id}", + connection_qualified_name=connection_qualified_name, + connector_name=connector_name, + cognos_id=cognos_id, + ) + + @classmethod + def create(cls, **kwargs) -> "CognosColumn": + return cls.creator(**kwargs) + + @classmethod + def create_for_modification(cls, **kwargs) -> "CognosColumn": + return cls.updater(**kwargs) diff --git a/pyatlan_v9/model/assets/_overlays/cognos_dashboard.py b/pyatlan_v9/model/assets/_overlays/cognos_dashboard.py new file mode 100644 index 000000000..47c53eec9 --- /dev/null +++ b/pyatlan_v9/model/assets/_overlays/cognos_dashboard.py @@ -0,0 +1,32 @@ +# INTERNAL_IMPORT: from pyatlan.utils import init_guid, validate_required_fields + + @classmethod + @init_guid + def creator( + cls, + *, + name: str, + connection_qualified_name: str, + cognos_id: str, + ) -> "CognosDashboard": + validate_required_fields( + ["name", "connection_qualified_name", "cognos_id"], + [name, connection_qualified_name, cognos_id], + ) + 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}/dashboard/{cognos_id}", + connection_qualified_name=connection_qualified_name, + connector_name=connector_name, + cognos_id=cognos_id, + ) + + @classmethod + def create(cls, **kwargs) -> "CognosDashboard": + return cls.creator(**kwargs) + + @classmethod + def create_for_modification(cls, **kwargs) -> "CognosDashboard": + return cls.updater(**kwargs) diff --git a/pyatlan_v9/model/assets/_overlays/cognos_dataset.py b/pyatlan_v9/model/assets/_overlays/cognos_dataset.py new file mode 100644 index 000000000..1752168a5 --- /dev/null +++ b/pyatlan_v9/model/assets/_overlays/cognos_dataset.py @@ -0,0 +1,32 @@ +# INTERNAL_IMPORT: from pyatlan.utils import init_guid, validate_required_fields + + @classmethod + @init_guid + def creator( + cls, + *, + name: str, + connection_qualified_name: str, + cognos_id: str, + ) -> "CognosDataset": + validate_required_fields( + ["name", "connection_qualified_name", "cognos_id"], + [name, connection_qualified_name, cognos_id], + ) + 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}/dataSet2/{cognos_id}", + connection_qualified_name=connection_qualified_name, + connector_name=connector_name, + cognos_id=cognos_id, + ) + + @classmethod + def create(cls, **kwargs) -> "CognosDataset": + return cls.creator(**kwargs) + + @classmethod + def create_for_modification(cls, **kwargs) -> "CognosDataset": + return cls.updater(**kwargs) diff --git a/pyatlan_v9/model/assets/_overlays/cognos_datasource.py b/pyatlan_v9/model/assets/_overlays/cognos_datasource.py new file mode 100644 index 000000000..a26cedb86 --- /dev/null +++ b/pyatlan_v9/model/assets/_overlays/cognos_datasource.py @@ -0,0 +1,32 @@ +# INTERNAL_IMPORT: from pyatlan.utils import init_guid, validate_required_fields + + @classmethod + @init_guid + def creator( + cls, + *, + name: str, + connection_qualified_name: str, + cognos_id: str, + ) -> "CognosDatasource": + validate_required_fields( + ["name", "connection_qualified_name", "cognos_id"], + [name, connection_qualified_name, cognos_id], + ) + 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}/datasource/{cognos_id}", + connection_qualified_name=connection_qualified_name, + connector_name=connector_name, + cognos_id=cognos_id, + ) + + @classmethod + def create(cls, **kwargs) -> "CognosDatasource": + return cls.creator(**kwargs) + + @classmethod + def create_for_modification(cls, **kwargs) -> "CognosDatasource": + return cls.updater(**kwargs) diff --git a/pyatlan_v9/model/assets/_overlays/cognos_exploration.py b/pyatlan_v9/model/assets/_overlays/cognos_exploration.py new file mode 100644 index 000000000..aff3f6752 --- /dev/null +++ b/pyatlan_v9/model/assets/_overlays/cognos_exploration.py @@ -0,0 +1,32 @@ +# INTERNAL_IMPORT: from pyatlan.utils import init_guid, validate_required_fields + + @classmethod + @init_guid + def creator( + cls, + *, + name: str, + connection_qualified_name: str, + cognos_id: str, + ) -> "CognosExploration": + validate_required_fields( + ["name", "connection_qualified_name", "cognos_id"], + [name, connection_qualified_name, cognos_id], + ) + 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}/exploration/{cognos_id}", + connection_qualified_name=connection_qualified_name, + connector_name=connector_name, + cognos_id=cognos_id, + ) + + @classmethod + def create(cls, **kwargs) -> "CognosExploration": + return cls.creator(**kwargs) + + @classmethod + def create_for_modification(cls, **kwargs) -> "CognosExploration": + return cls.updater(**kwargs) diff --git a/pyatlan_v9/model/assets/_overlays/cognos_file.py b/pyatlan_v9/model/assets/_overlays/cognos_file.py new file mode 100644 index 000000000..f82159554 --- /dev/null +++ b/pyatlan_v9/model/assets/_overlays/cognos_file.py @@ -0,0 +1,32 @@ +# INTERNAL_IMPORT: from pyatlan.utils import init_guid, validate_required_fields + + @classmethod + @init_guid + def creator( + cls, + *, + name: str, + connection_qualified_name: str, + cognos_id: str, + ) -> "CognosFile": + validate_required_fields( + ["name", "connection_qualified_name", "cognos_id"], + [name, connection_qualified_name, cognos_id], + ) + 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}/file/{cognos_id}", + connection_qualified_name=connection_qualified_name, + connector_name=connector_name, + cognos_id=cognos_id, + ) + + @classmethod + def create(cls, **kwargs) -> "CognosFile": + return cls.creator(**kwargs) + + @classmethod + def create_for_modification(cls, **kwargs) -> "CognosFile": + return cls.updater(**kwargs) diff --git a/pyatlan_v9/model/assets/_overlays/cognos_folder.py b/pyatlan_v9/model/assets/_overlays/cognos_folder.py new file mode 100644 index 000000000..545d36604 --- /dev/null +++ b/pyatlan_v9/model/assets/_overlays/cognos_folder.py @@ -0,0 +1,32 @@ +# INTERNAL_IMPORT: from pyatlan.utils import init_guid, validate_required_fields + + @classmethod + @init_guid + def creator( + cls, + *, + name: str, + connection_qualified_name: str, + cognos_id: str, + ) -> "CognosFolder": + validate_required_fields( + ["name", "connection_qualified_name", "cognos_id"], + [name, connection_qualified_name, cognos_id], + ) + 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}/folder/{cognos_id}", + connection_qualified_name=connection_qualified_name, + connector_name=connector_name, + cognos_id=cognos_id, + ) + + @classmethod + def create(cls, **kwargs) -> "CognosFolder": + return cls.creator(**kwargs) + + @classmethod + def create_for_modification(cls, **kwargs) -> "CognosFolder": + return cls.updater(**kwargs) diff --git a/pyatlan_v9/model/assets/_overlays/cognos_module.py b/pyatlan_v9/model/assets/_overlays/cognos_module.py new file mode 100644 index 000000000..35df75d30 --- /dev/null +++ b/pyatlan_v9/model/assets/_overlays/cognos_module.py @@ -0,0 +1,32 @@ +# INTERNAL_IMPORT: from pyatlan.utils import init_guid, validate_required_fields + + @classmethod + @init_guid + def creator( + cls, + *, + name: str, + connection_qualified_name: str, + cognos_id: str, + ) -> "CognosModule": + validate_required_fields( + ["name", "connection_qualified_name", "cognos_id"], + [name, connection_qualified_name, cognos_id], + ) + 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}/module/{cognos_id}", + connection_qualified_name=connection_qualified_name, + connector_name=connector_name, + cognos_id=cognos_id, + ) + + @classmethod + def create(cls, **kwargs) -> "CognosModule": + return cls.creator(**kwargs) + + @classmethod + def create_for_modification(cls, **kwargs) -> "CognosModule": + return cls.updater(**kwargs) diff --git a/pyatlan_v9/model/assets/_overlays/cognos_package.py b/pyatlan_v9/model/assets/_overlays/cognos_package.py new file mode 100644 index 000000000..bffa45a86 --- /dev/null +++ b/pyatlan_v9/model/assets/_overlays/cognos_package.py @@ -0,0 +1,32 @@ +# INTERNAL_IMPORT: from pyatlan.utils import init_guid, validate_required_fields + + @classmethod + @init_guid + def creator( + cls, + *, + name: str, + connection_qualified_name: str, + cognos_id: str, + ) -> "CognosPackage": + validate_required_fields( + ["name", "connection_qualified_name", "cognos_id"], + [name, connection_qualified_name, cognos_id], + ) + 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}/package/{cognos_id}", + connection_qualified_name=connection_qualified_name, + connector_name=connector_name, + cognos_id=cognos_id, + ) + + @classmethod + def create(cls, **kwargs) -> "CognosPackage": + return cls.creator(**kwargs) + + @classmethod + def create_for_modification(cls, **kwargs) -> "CognosPackage": + return cls.updater(**kwargs) diff --git a/pyatlan_v9/model/assets/_overlays/cognos_report.py b/pyatlan_v9/model/assets/_overlays/cognos_report.py new file mode 100644 index 000000000..e8685840a --- /dev/null +++ b/pyatlan_v9/model/assets/_overlays/cognos_report.py @@ -0,0 +1,32 @@ +# INTERNAL_IMPORT: from pyatlan.utils import init_guid, validate_required_fields + + @classmethod + @init_guid + def creator( + cls, + *, + name: str, + connection_qualified_name: str, + cognos_id: str, + ) -> "CognosReport": + validate_required_fields( + ["name", "connection_qualified_name", "cognos_id"], + [name, connection_qualified_name, cognos_id], + ) + 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}/report/{cognos_id}", + connection_qualified_name=connection_qualified_name, + connector_name=connector_name, + cognos_id=cognos_id, + ) + + @classmethod + def create(cls, **kwargs) -> "CognosReport": + return cls.creator(**kwargs) + + @classmethod + def create_for_modification(cls, **kwargs) -> "CognosReport": + return cls.updater(**kwargs) diff --git a/pyatlan_v9/model/assets/cognos_column.py b/pyatlan_v9/model/assets/cognos_column.py index 3f5a65b10..628a4006a 100644 --- a/pyatlan_v9/model/assets/cognos_column.py +++ b/pyatlan_v9/model/assets/cognos_column.py @@ -25,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 .airflow_related import RelatedAirflowTask from .anomalo_related import RelatedAnomaloCheck @@ -301,9 +302,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"^.+/column/[^/]+$") def validate(self, for_creation: bool = False) -> None: """ @@ -371,6 +370,37 @@ def relate(self) -> "RelatedCognosColumn": return RelatedCognosColumn(guid=self.guid) return RelatedCognosColumn(qualified_name=self.qualified_name) + @classmethod + @init_guid + def creator( + cls, + *, + name: str, + connection_qualified_name: str, + cognos_id: str, + ) -> "CognosColumn": + validate_required_fields( + ["name", "connection_qualified_name", "cognos_id"], + [name, connection_qualified_name, cognos_id], + ) + 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}/column/{cognos_id}", + connection_qualified_name=connection_qualified_name, + connector_name=connector_name, + cognos_id=cognos_id, + ) + + @classmethod + def create(cls, **kwargs) -> "CognosColumn": + return cls.creator(**kwargs) + + @classmethod + def create_for_modification(cls, **kwargs) -> "CognosColumn": + return cls.updater(**kwargs) + # ========================================================================= # Optimized Serialization Methods (override Asset base class) # ========================================================================= diff --git a/pyatlan_v9/model/assets/cognos_dashboard.py b/pyatlan_v9/model/assets/cognos_dashboard.py index 986425d7b..e13efb384 100644 --- a/pyatlan_v9/model/assets/cognos_dashboard.py +++ b/pyatlan_v9/model/assets/cognos_dashboard.py @@ -25,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 .airflow_related import RelatedAirflowTask from .anomalo_related import RelatedAnomaloCheck @@ -269,9 +270,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"^.+/dashboard/[^/]+$") def validate(self, for_creation: bool = False) -> None: """ @@ -339,6 +338,37 @@ def relate(self) -> "RelatedCognosDashboard": return RelatedCognosDashboard(guid=self.guid) return RelatedCognosDashboard(qualified_name=self.qualified_name) + @classmethod + @init_guid + def creator( + cls, + *, + name: str, + connection_qualified_name: str, + cognos_id: str, + ) -> "CognosDashboard": + validate_required_fields( + ["name", "connection_qualified_name", "cognos_id"], + [name, connection_qualified_name, cognos_id], + ) + 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}/dashboard/{cognos_id}", + connection_qualified_name=connection_qualified_name, + connector_name=connector_name, + cognos_id=cognos_id, + ) + + @classmethod + def create(cls, **kwargs) -> "CognosDashboard": + return cls.creator(**kwargs) + + @classmethod + def create_for_modification(cls, **kwargs) -> "CognosDashboard": + return cls.updater(**kwargs) + # ========================================================================= # Optimized Serialization Methods (override Asset base class) # ========================================================================= diff --git a/pyatlan_v9/model/assets/cognos_dataset.py b/pyatlan_v9/model/assets/cognos_dataset.py index 893de18e0..5dedade9e 100644 --- a/pyatlan_v9/model/assets/cognos_dataset.py +++ b/pyatlan_v9/model/assets/cognos_dataset.py @@ -25,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 .airflow_related import RelatedAirflowTask from .anomalo_related import RelatedAnomaloCheck @@ -269,9 +270,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"^.+/dataSet2/[^/]+$") def validate(self, for_creation: bool = False) -> None: """ @@ -339,6 +338,37 @@ def relate(self) -> "RelatedCognosDataset": return RelatedCognosDataset(guid=self.guid) return RelatedCognosDataset(qualified_name=self.qualified_name) + @classmethod + @init_guid + def creator( + cls, + *, + name: str, + connection_qualified_name: str, + cognos_id: str, + ) -> "CognosDataset": + validate_required_fields( + ["name", "connection_qualified_name", "cognos_id"], + [name, connection_qualified_name, cognos_id], + ) + 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}/dataSet2/{cognos_id}", + connection_qualified_name=connection_qualified_name, + connector_name=connector_name, + cognos_id=cognos_id, + ) + + @classmethod + def create(cls, **kwargs) -> "CognosDataset": + return cls.creator(**kwargs) + + @classmethod + def create_for_modification(cls, **kwargs) -> "CognosDataset": + return cls.updater(**kwargs) + # ========================================================================= # Optimized Serialization Methods (override Asset base class) # ========================================================================= diff --git a/pyatlan_v9/model/assets/cognos_datasource.py b/pyatlan_v9/model/assets/cognos_datasource.py index 32040741c..fbc29878a 100644 --- a/pyatlan_v9/model/assets/cognos_datasource.py +++ b/pyatlan_v9/model/assets/cognos_datasource.py @@ -14,6 +14,7 @@ from __future__ import annotations +import re from typing import Any, ClassVar, 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 .airflow_related import RelatedAirflowTask from .anomalo_related import RelatedAnomaloCheck @@ -260,6 +262,8 @@ def __post_init__(self) -> None: # SDK Methods # ========================================================================= + _QUALIFIED_NAME_PATTERN: ClassVar[re.Pattern] = re.compile(r"^.+/datasource/[^/]+$") + def validate(self, for_creation: bool = False) -> None: """ Dry-run validation of this CognosDatasource instance. @@ -284,6 +288,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"CognosDatasource validation failed: {errors}") @@ -316,6 +325,37 @@ def relate(self) -> "RelatedCognosDatasource": return RelatedCognosDatasource(guid=self.guid) return RelatedCognosDatasource(qualified_name=self.qualified_name) + @classmethod + @init_guid + def creator( + cls, + *, + name: str, + connection_qualified_name: str, + cognos_id: str, + ) -> "CognosDatasource": + validate_required_fields( + ["name", "connection_qualified_name", "cognos_id"], + [name, connection_qualified_name, cognos_id], + ) + 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}/datasource/{cognos_id}", + connection_qualified_name=connection_qualified_name, + connector_name=connector_name, + cognos_id=cognos_id, + ) + + @classmethod + def create(cls, **kwargs) -> "CognosDatasource": + return cls.creator(**kwargs) + + @classmethod + def create_for_modification(cls, **kwargs) -> "CognosDatasource": + return cls.updater(**kwargs) + # ========================================================================= # Optimized Serialization Methods (override Asset base class) # ========================================================================= diff --git a/pyatlan_v9/model/assets/cognos_exploration.py b/pyatlan_v9/model/assets/cognos_exploration.py index d67356ef1..e9ae8f3a9 100644 --- a/pyatlan_v9/model/assets/cognos_exploration.py +++ b/pyatlan_v9/model/assets/cognos_exploration.py @@ -25,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 .airflow_related import RelatedAirflowTask from .anomalo_related import RelatedAnomaloCheck @@ -270,7 +271,7 @@ def __post_init__(self) -> None: # ========================================================================= _QUALIFIED_NAME_PATTERN: ClassVar[re.Pattern] = re.compile( - r"^.+/[^/]+/[^/]+/[^/]+$" + r"^.+/exploration/[^/]+$" ) def validate(self, for_creation: bool = False) -> None: @@ -339,6 +340,37 @@ def relate(self) -> "RelatedCognosExploration": return RelatedCognosExploration(guid=self.guid) return RelatedCognosExploration(qualified_name=self.qualified_name) + @classmethod + @init_guid + def creator( + cls, + *, + name: str, + connection_qualified_name: str, + cognos_id: str, + ) -> "CognosExploration": + validate_required_fields( + ["name", "connection_qualified_name", "cognos_id"], + [name, connection_qualified_name, cognos_id], + ) + 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}/exploration/{cognos_id}", + connection_qualified_name=connection_qualified_name, + connector_name=connector_name, + cognos_id=cognos_id, + ) + + @classmethod + def create(cls, **kwargs) -> "CognosExploration": + return cls.creator(**kwargs) + + @classmethod + def create_for_modification(cls, **kwargs) -> "CognosExploration": + return cls.updater(**kwargs) + # ========================================================================= # Optimized Serialization Methods (override Asset base class) # ========================================================================= diff --git a/pyatlan_v9/model/assets/cognos_file.py b/pyatlan_v9/model/assets/cognos_file.py index b2371e1e7..ad9e9b9b1 100644 --- a/pyatlan_v9/model/assets/cognos_file.py +++ b/pyatlan_v9/model/assets/cognos_file.py @@ -25,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 .airflow_related import RelatedAirflowTask from .anomalo_related import RelatedAnomaloCheck @@ -265,9 +266,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"^.+/file/[^/]+$") def validate(self, for_creation: bool = False) -> None: """ @@ -335,6 +334,37 @@ def relate(self) -> "RelatedCognosFile": return RelatedCognosFile(guid=self.guid) return RelatedCognosFile(qualified_name=self.qualified_name) + @classmethod + @init_guid + def creator( + cls, + *, + name: str, + connection_qualified_name: str, + cognos_id: str, + ) -> "CognosFile": + validate_required_fields( + ["name", "connection_qualified_name", "cognos_id"], + [name, connection_qualified_name, cognos_id], + ) + 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}/file/{cognos_id}", + connection_qualified_name=connection_qualified_name, + connector_name=connector_name, + cognos_id=cognos_id, + ) + + @classmethod + def create(cls, **kwargs) -> "CognosFile": + return cls.creator(**kwargs) + + @classmethod + def create_for_modification(cls, **kwargs) -> "CognosFile": + return cls.updater(**kwargs) + # ========================================================================= # Optimized Serialization Methods (override Asset base class) # ========================================================================= diff --git a/pyatlan_v9/model/assets/cognos_folder.py b/pyatlan_v9/model/assets/cognos_folder.py index 6d693fef3..15f1fb96a 100644 --- a/pyatlan_v9/model/assets/cognos_folder.py +++ b/pyatlan_v9/model/assets/cognos_folder.py @@ -25,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 .airflow_related import RelatedAirflowTask from .anomalo_related import RelatedAnomaloCheck @@ -310,7 +311,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"^.+/folder/[^/]+$") def validate(self, for_creation: bool = False) -> None: """ @@ -376,6 +377,37 @@ def relate(self) -> "RelatedCognosFolder": return RelatedCognosFolder(guid=self.guid) return RelatedCognosFolder(qualified_name=self.qualified_name) + @classmethod + @init_guid + def creator( + cls, + *, + name: str, + connection_qualified_name: str, + cognos_id: str, + ) -> "CognosFolder": + validate_required_fields( + ["name", "connection_qualified_name", "cognos_id"], + [name, connection_qualified_name, cognos_id], + ) + 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}/folder/{cognos_id}", + connection_qualified_name=connection_qualified_name, + connector_name=connector_name, + cognos_id=cognos_id, + ) + + @classmethod + def create(cls, **kwargs) -> "CognosFolder": + return cls.creator(**kwargs) + + @classmethod + def create_for_modification(cls, **kwargs) -> "CognosFolder": + return cls.updater(**kwargs) + # ========================================================================= # Optimized Serialization Methods (override Asset base class) # ========================================================================= diff --git a/pyatlan_v9/model/assets/cognos_module.py b/pyatlan_v9/model/assets/cognos_module.py index 47a2b6e1a..1262f8acb 100644 --- a/pyatlan_v9/model/assets/cognos_module.py +++ b/pyatlan_v9/model/assets/cognos_module.py @@ -25,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 .airflow_related import RelatedAirflowTask from .anomalo_related import RelatedAnomaloCheck @@ -269,9 +270,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"^.+/module/[^/]+$") def validate(self, for_creation: bool = False) -> None: """ @@ -339,6 +338,37 @@ def relate(self) -> "RelatedCognosModule": return RelatedCognosModule(guid=self.guid) return RelatedCognosModule(qualified_name=self.qualified_name) + @classmethod + @init_guid + def creator( + cls, + *, + name: str, + connection_qualified_name: str, + cognos_id: str, + ) -> "CognosModule": + validate_required_fields( + ["name", "connection_qualified_name", "cognos_id"], + [name, connection_qualified_name, cognos_id], + ) + 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}/module/{cognos_id}", + connection_qualified_name=connection_qualified_name, + connector_name=connector_name, + cognos_id=cognos_id, + ) + + @classmethod + def create(cls, **kwargs) -> "CognosModule": + return cls.creator(**kwargs) + + @classmethod + def create_for_modification(cls, **kwargs) -> "CognosModule": + return cls.updater(**kwargs) + # ========================================================================= # Optimized Serialization Methods (override Asset base class) # ========================================================================= diff --git a/pyatlan_v9/model/assets/cognos_package.py b/pyatlan_v9/model/assets/cognos_package.py index e7f2f39e6..1c2190d2c 100644 --- a/pyatlan_v9/model/assets/cognos_package.py +++ b/pyatlan_v9/model/assets/cognos_package.py @@ -25,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 .airflow_related import RelatedAirflowTask from .anomalo_related import RelatedAnomaloCheck @@ -269,9 +270,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"^.+/package/[^/]+$") def validate(self, for_creation: bool = False) -> None: """ @@ -339,6 +338,37 @@ def relate(self) -> "RelatedCognosPackage": return RelatedCognosPackage(guid=self.guid) return RelatedCognosPackage(qualified_name=self.qualified_name) + @classmethod + @init_guid + def creator( + cls, + *, + name: str, + connection_qualified_name: str, + cognos_id: str, + ) -> "CognosPackage": + validate_required_fields( + ["name", "connection_qualified_name", "cognos_id"], + [name, connection_qualified_name, cognos_id], + ) + 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}/package/{cognos_id}", + connection_qualified_name=connection_qualified_name, + connector_name=connector_name, + cognos_id=cognos_id, + ) + + @classmethod + def create(cls, **kwargs) -> "CognosPackage": + return cls.creator(**kwargs) + + @classmethod + def create_for_modification(cls, **kwargs) -> "CognosPackage": + return cls.updater(**kwargs) + # ========================================================================= # Optimized Serialization Methods (override Asset base class) # ========================================================================= diff --git a/pyatlan_v9/model/assets/cognos_report.py b/pyatlan_v9/model/assets/cognos_report.py index 141efd27a..77c8c401b 100644 --- a/pyatlan_v9/model/assets/cognos_report.py +++ b/pyatlan_v9/model/assets/cognos_report.py @@ -25,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 .airflow_related import RelatedAirflowTask from .anomalo_related import RelatedAnomaloCheck @@ -261,9 +262,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"^.+/report/[^/]+$") def validate(self, for_creation: bool = False) -> None: """ @@ -331,6 +330,37 @@ def relate(self) -> "RelatedCognosReport": return RelatedCognosReport(guid=self.guid) return RelatedCognosReport(qualified_name=self.qualified_name) + @classmethod + @init_guid + def creator( + cls, + *, + name: str, + connection_qualified_name: str, + cognos_id: str, + ) -> "CognosReport": + validate_required_fields( + ["name", "connection_qualified_name", "cognos_id"], + [name, connection_qualified_name, cognos_id], + ) + 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}/report/{cognos_id}", + connection_qualified_name=connection_qualified_name, + connector_name=connector_name, + cognos_id=cognos_id, + ) + + @classmethod + def create(cls, **kwargs) -> "CognosReport": + return cls.creator(**kwargs) + + @classmethod + def create_for_modification(cls, **kwargs) -> "CognosReport": + return cls.updater(**kwargs) + # ========================================================================= # Optimized Serialization Methods (override Asset base class) # ========================================================================= diff --git a/tests_v9/integration/cognos_asset_test.py b/tests_v9/integration/cognos_asset_test.py new file mode 100644 index 000000000..e82335894 --- /dev/null +++ b/tests_v9/integration/cognos_asset_test.py @@ -0,0 +1,159 @@ +# SPDX-License-Identifier: Apache-2.0 +# Copyright 2026 Atlan Pte. Ltd. + +from typing import Generator + +import pytest + +from pyatlan_v9.client.atlan import AtlanClient +from pyatlan_v9.model.assets import ( + CognosColumn, + CognosDashboard, + CognosDataset, + CognosDatasource, + CognosExploration, + CognosFile, + CognosFolder, + CognosModule, + CognosPackage, + CognosReport, + Connection, +) +from pyatlan_v9.model.enums import AtlanConnectorType +from tests_v9.integration.client import TestId, delete_asset +from tests_v9.integration.connection_test import create_connection + +MODULE_NAME = TestId.make_unique("COGNOS") + +FOLDER_CHILDREN = [ + (CognosReport, "report"), + (CognosDashboard, "dashboard"), + (CognosExploration, "exploration"), + (CognosFile, "file"), + (CognosModule, "module"), + (CognosPackage, "package"), + (CognosDataset, "dataSet2"), +] + + +def _cognos_id(slug: str) -> str: + return f"i{slug}{MODULE_NAME}" + + +@pytest.fixture(scope="module") +def connection(client: AtlanClient) -> Generator[Connection, None, None]: + result = create_connection( + client=client, name=MODULE_NAME, connector_type=AtlanConnectorType.COGNOS + ) + yield result + delete_asset(client, guid=result.guid, asset_type=Connection) + + +@pytest.fixture(scope="module") +def folder( + client: AtlanClient, connection: Connection +) -> Generator[CognosFolder, None, None]: + to_create = CognosFolder.creator( + name=f"folder_{MODULE_NAME}", + connection_qualified_name=connection.qualified_name, + cognos_id=_cognos_id("folder"), + ) + result = client.asset.save(to_create).assets_created(asset_type=CognosFolder)[0] + yield result + delete_asset(client, guid=result.guid, asset_type=CognosFolder) + + +@pytest.fixture(scope="module") +def folder_children( + client: AtlanClient, connection: Connection, folder: CognosFolder +) -> Generator[dict, None, None]: + created = {} + for asset_type, slug in FOLDER_CHILDREN: + to_create = asset_type.creator( + name=f"{slug}_{MODULE_NAME}", + connection_qualified_name=connection.qualified_name, + cognos_id=_cognos_id(slug), + ) + to_create.cognos_folder = folder.relate() + created[asset_type] = client.asset.save(to_create).assets_created( + asset_type=asset_type + )[0] + yield created + for asset_type, asset in created.items(): + delete_asset(client, guid=asset.guid, asset_type=asset_type) + + +@pytest.fixture(scope="module") +def datasource( + client: AtlanClient, connection: Connection +) -> Generator[CognosDatasource, None, None]: + to_create = CognosDatasource.creator( + name=f"datasource_{MODULE_NAME}", + connection_qualified_name=connection.qualified_name, + cognos_id=_cognos_id("datasource"), + ) + result = client.asset.save(to_create).assets_created(asset_type=CognosDatasource)[0] + yield result + delete_asset(client, guid=result.guid, asset_type=CognosDatasource) + + +@pytest.fixture(scope="module") +def column( + client: AtlanClient, connection: Connection, folder_children: dict +) -> Generator[CognosColumn, None, None]: + to_create = CognosColumn.creator( + name=f"column_{MODULE_NAME}", + connection_qualified_name=connection.qualified_name, + cognos_id=_cognos_id("column"), + ) + to_create.cognos_dataset = folder_children[CognosDataset].relate() + result = client.asset.save(to_create).assets_created(asset_type=CognosColumn)[0] + yield result + delete_asset(client, guid=result.guid, asset_type=CognosColumn) + + +def _assert_retrieved(client, asset, asset_type, connection, slug): + retrieved = client.asset.get_by_guid( + asset.guid, asset_type=asset_type, ignore_relationships=False + ) + assert retrieved.qualified_name == ( + f"{connection.qualified_name}/{slug}/{_cognos_id(slug)}" + ) + assert retrieved.name == asset.name + assert retrieved.cognos_id == _cognos_id(slug) + assert retrieved.connector_name == AtlanConnectorType.COGNOS + assert retrieved.connection_qualified_name == connection.qualified_name + return retrieved + + +def test_folder(client: AtlanClient, connection: Connection, folder: CognosFolder): + _assert_retrieved(client, folder, CognosFolder, connection, "folder") + + +def test_folder_children( + client: AtlanClient, + connection: Connection, + folder: CognosFolder, + folder_children: dict, +): + for asset_type, slug in FOLDER_CHILDREN: + retrieved = _assert_retrieved( + client, folder_children[asset_type], asset_type, connection, slug + ) + assert retrieved.cognos_folder.guid == folder.guid + + +def test_datasource( + client: AtlanClient, connection: Connection, datasource: CognosDatasource +): + _assert_retrieved(client, datasource, CognosDatasource, connection, "datasource") + + +def test_column( + client: AtlanClient, + connection: Connection, + folder_children: dict, + column: CognosColumn, +): + retrieved = _assert_retrieved(client, column, CognosColumn, connection, "column") + assert retrieved.cognos_dataset.guid == folder_children[CognosDataset].guid diff --git a/tests_v9/unit/model/cognos_test.py b/tests_v9/unit/model/cognos_test.py new file mode 100644 index 000000000..ee5dee0d7 --- /dev/null +++ b/tests_v9/unit/model/cognos_test.py @@ -0,0 +1,86 @@ +# SPDX-License-Identifier: Apache-2.0 +# Copyright 2026 Atlan Pte. Ltd. + +"""Unit tests for the Cognos creators in pyatlan_v9. + +The expected qualifiedNames are the exact strings atlan-cognos-app emits today +(FND-3838 census), so a creator change that would re-key existing Cognos assets fails here. +""" + +import pytest + +from pyatlan_v9.model import ( + CognosColumn, + CognosDashboard, + CognosDataset, + CognosDatasource, + CognosExploration, + CognosFile, + CognosFolder, + CognosModule, + CognosPackage, + CognosReport, +) + +CONNECTION_QN = "default/cognos/1712345678" +COGNOS_ID = "i1A2B3C4D5E6F" +NAME = "Quarterly Sales" + +ALL_TYPES = [ + (CognosReport, "default/cognos/1712345678/report/i1A2B3C4D5E6F"), + (CognosDashboard, "default/cognos/1712345678/dashboard/i1A2B3C4D5E6F"), + (CognosExploration, "default/cognos/1712345678/exploration/i1A2B3C4D5E6F"), + (CognosFile, "default/cognos/1712345678/file/i1A2B3C4D5E6F"), + (CognosModule, "default/cognos/1712345678/module/i1A2B3C4D5E6F"), + (CognosPackage, "default/cognos/1712345678/package/i1A2B3C4D5E6F"), + (CognosDataset, "default/cognos/1712345678/dataSet2/i1A2B3C4D5E6F"), + (CognosFolder, "default/cognos/1712345678/folder/i1A2B3C4D5E6F"), + (CognosDatasource, "default/cognos/1712345678/datasource/i1A2B3C4D5E6F"), + (CognosColumn, "default/cognos/1712345678/column/i1A2B3C4D5E6F"), +] + + +@pytest.mark.parametrize("asset_type, expected_qn", ALL_TYPES) +def test_creator_matches_app_qualified_name(asset_type, expected_qn): + sut = asset_type.creator( + name=NAME, connection_qualified_name=CONNECTION_QN, cognos_id=COGNOS_ID + ) + + assert sut.qualified_name == expected_qn + assert sut.name == NAME + assert sut.cognos_id == COGNOS_ID + assert sut.connection_qualified_name == CONNECTION_QN + assert sut.connector_name == "cognos" + assert sut._QUALIFIED_NAME_PATTERN.match(sut.qualified_name) + sut.validate() + + +@pytest.mark.parametrize("asset_type, _", ALL_TYPES) +@pytest.mark.parametrize( + "name, connection_qualified_name, cognos_id, message", + [ + (None, CONNECTION_QN, COGNOS_ID, "name is required"), + (NAME, None, COGNOS_ID, "connection_qualified_name is required"), + (NAME, CONNECTION_QN, None, "cognos_id is required"), + (NAME, CONNECTION_QN, " ", "cognos_id"), + (NAME, "", COGNOS_ID, "connection_qualified_name"), + ], +) +def test_creator_rejects_missing_or_blank_inputs( + asset_type, _, name, connection_qualified_name, cognos_id, message +): + with pytest.raises(ValueError, match=message): + asset_type.creator( + name=name, + connection_qualified_name=connection_qualified_name, + cognos_id=cognos_id, + ) + + +@pytest.mark.parametrize("asset_type, expected_qn", ALL_TYPES) +def test_create_alias(asset_type, expected_qn): + sut = asset_type.create( + name=NAME, connection_qualified_name=CONNECTION_QN, cognos_id=COGNOS_ID + ) + + assert sut.qualified_name == expected_qn