From 6544f642ffe9cd0dc52b11b6df6c4963391e7808 Mon Sep 17 00:00:00 2001 From: Christopher Grote Date: Wed, 14 Oct 2026 16:44:15 +0100 Subject: [PATCH] feat(v9): add creators for the 7 Sigma asset types (FND-3856) Overlays for SigmaWorkbook, SigmaPage, SigmaDataElement, SigmaDataElementField, SigmaDataset, SigmaDataModel and SigmaDataModelColumn. Each creator emits the qualifiedName atlan-sigma-app emits today, sets the parent relationship and the denormalised parent attributes, and raises on blank inputs. Classes regenerated from atlanhq/models#2082. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_0166vmu2bYFX2P4LFGH3GR4F --- .../assets/_overlays/sigma_data_element.py | 48 ++++ .../_overlays/sigma_data_element_field.py | 62 +++++ .../assets/_overlays/sigma_data_model.py | 32 +++ .../_overlays/sigma_data_model_column.py | 49 ++++ .../model/assets/_overlays/sigma_dataset.py | 31 +++ .../model/assets/_overlays/sigma_page.py | 39 +++ .../model/assets/_overlays/sigma_workbook.py | 31 +++ pyatlan_v9/model/assets/sigma_data_element.py | 48 ++++ .../model/assets/sigma_data_element_field.py | 62 +++++ pyatlan_v9/model/assets/sigma_data_model.py | 40 +++ .../model/assets/sigma_data_model_column.py | 53 +++- pyatlan_v9/model/assets/sigma_dataset.py | 39 +++ pyatlan_v9/model/assets/sigma_page.py | 39 +++ pyatlan_v9/model/assets/sigma_workbook.py | 39 +++ tests_v9/integration/sigma_asset_test.py | 240 ++++++++++++++++++ tests_v9/unit/model/sigma_test.py | 230 +++++++++++++++++ 16 files changed, 1081 insertions(+), 1 deletion(-) create mode 100644 pyatlan_v9/model/assets/_overlays/sigma_data_element.py create mode 100644 pyatlan_v9/model/assets/_overlays/sigma_data_element_field.py create mode 100644 pyatlan_v9/model/assets/_overlays/sigma_data_model.py create mode 100644 pyatlan_v9/model/assets/_overlays/sigma_data_model_column.py create mode 100644 pyatlan_v9/model/assets/_overlays/sigma_dataset.py create mode 100644 pyatlan_v9/model/assets/_overlays/sigma_page.py create mode 100644 pyatlan_v9/model/assets/_overlays/sigma_workbook.py create mode 100644 tests_v9/integration/sigma_asset_test.py create mode 100644 tests_v9/unit/model/sigma_test.py diff --git a/pyatlan_v9/model/assets/_overlays/sigma_data_element.py b/pyatlan_v9/model/assets/_overlays/sigma_data_element.py new file mode 100644 index 000000000..30537a3a7 --- /dev/null +++ b/pyatlan_v9/model/assets/_overlays/sigma_data_element.py @@ -0,0 +1,48 @@ +# INTERNAL_IMPORT: from pyatlan.utils import init_guid, validate_required_fields + + @classmethod + @init_guid + def creator( + cls, + *, + name: str, + page_qualified_name: str, + page_name: str, + workbook_name: str, + data_element_id: str, + ) -> "SigmaDataElement": + validate_required_fields( + [ + "name", + "page_qualified_name", + "page_name", + "workbook_name", + "data_element_id", + ], + [name, page_qualified_name, page_name, workbook_name, data_element_id], + ) + fields = page_qualified_name.split("/") + if len(fields) != 5: + raise ValueError( + f"Invalid page_qualified_name: {page_qualified_name}. " + "Expected format: default/sigma///" + ) + return cls( + name=name, + qualified_name=f"{page_qualified_name}/{data_element_id}", + connection_qualified_name="/".join(fields[:3]), + connector_name=fields[1], + sigma_workbook_qualified_name="/".join(fields[:4]), + sigma_workbook_name=workbook_name, + sigma_page_qualified_name=page_qualified_name, + sigma_page_name=page_name, + sigma_page=RelatedSigmaPage(qualified_name=page_qualified_name), + ) + + @classmethod + def create(cls, **kwargs) -> "SigmaDataElement": + return cls.creator(**kwargs) + + @classmethod + def create_for_modification(cls, **kwargs) -> "SigmaDataElement": + return cls.updater(**kwargs) diff --git a/pyatlan_v9/model/assets/_overlays/sigma_data_element_field.py b/pyatlan_v9/model/assets/_overlays/sigma_data_element_field.py new file mode 100644 index 000000000..d2fa9af2a --- /dev/null +++ b/pyatlan_v9/model/assets/_overlays/sigma_data_element_field.py @@ -0,0 +1,62 @@ +# INTERNAL_IMPORT: from pyatlan.utils import init_guid, validate_required_fields + + @classmethod + @init_guid + def creator( + cls, + *, + name: str, + data_element_qualified_name: str, + data_element_name: str, + page_name: str, + workbook_name: str, + field_id: str, + ) -> "SigmaDataElementField": + validate_required_fields( + [ + "name", + "data_element_qualified_name", + "data_element_name", + "page_name", + "workbook_name", + "field_id", + ], + [ + name, + data_element_qualified_name, + data_element_name, + page_name, + workbook_name, + field_id, + ], + ) + fields = data_element_qualified_name.split("/") + if len(fields) != 6: + raise ValueError( + f"Invalid data_element_qualified_name: {data_element_qualified_name}. " + "Expected format: " + "default/sigma////" + ) + return cls( + name=name, + qualified_name=f"{data_element_qualified_name}/{field_id}", + connection_qualified_name="/".join(fields[:3]), + connector_name=fields[1], + sigma_workbook_qualified_name="/".join(fields[:4]), + sigma_workbook_name=workbook_name, + sigma_page_qualified_name="/".join(fields[:5]), + sigma_page_name=page_name, + sigma_data_element_qualified_name=data_element_qualified_name, + sigma_data_element_name=data_element_name, + sigma_data_element=RelatedSigmaDataElement( + qualified_name=data_element_qualified_name + ), + ) + + @classmethod + def create(cls, **kwargs) -> "SigmaDataElementField": + return cls.creator(**kwargs) + + @classmethod + def create_for_modification(cls, **kwargs) -> "SigmaDataElementField": + return cls.updater(**kwargs) diff --git a/pyatlan_v9/model/assets/_overlays/sigma_data_model.py b/pyatlan_v9/model/assets/_overlays/sigma_data_model.py new file mode 100644 index 000000000..556e27102 --- /dev/null +++ b/pyatlan_v9/model/assets/_overlays/sigma_data_model.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, + data_model_id: str, + ) -> "SigmaDataModel": + validate_required_fields( + ["name", "connection_qualified_name", "data_model_id"], + [name, connection_qualified_name, data_model_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}/dataModels/{data_model_id}", + connection_qualified_name=connection_qualified_name, + connector_name=connector_name, + asset_source_id=data_model_id, + ) + + @classmethod + def create(cls, **kwargs) -> "SigmaDataModel": + return cls.creator(**kwargs) + + @classmethod + def create_for_modification(cls, **kwargs) -> "SigmaDataModel": + return cls.updater(**kwargs) diff --git a/pyatlan_v9/model/assets/_overlays/sigma_data_model_column.py b/pyatlan_v9/model/assets/_overlays/sigma_data_model_column.py new file mode 100644 index 000000000..c77aa6782 --- /dev/null +++ b/pyatlan_v9/model/assets/_overlays/sigma_data_model_column.py @@ -0,0 +1,49 @@ +# INTERNAL_IMPORT: from pyatlan.utils import init_guid, validate_required_fields + + @classmethod + @init_guid + def creator( + cls, + *, + name: str, + data_model_qualified_name: str, + data_model_name: str, + element_id: str, + column_id: str, + ) -> "SigmaDataModelColumn": + validate_required_fields( + [ + "name", + "data_model_qualified_name", + "data_model_name", + "element_id", + "column_id", + ], + [name, data_model_qualified_name, data_model_name, element_id, column_id], + ) + fields = data_model_qualified_name.split("/") + if len(fields) != 5 or fields[3] != "dataModels": + raise ValueError( + f"Invalid data_model_qualified_name: {data_model_qualified_name}. " + "Expected format: default/sigma//dataModels/" + ) + return cls( + name=name, + qualified_name=f"{data_model_qualified_name}/{element_id}/{column_id}", + connection_qualified_name="/".join(fields[:3]), + connector_name=fields[1], + asset_source_id=column_id, + sigma_data_model_qualified_name=data_model_qualified_name, + sigma_data_model_name=data_model_name, + sigma_data_model=RelatedSigmaDataModel( + qualified_name=data_model_qualified_name + ), + ) + + @classmethod + def create(cls, **kwargs) -> "SigmaDataModelColumn": + return cls.creator(**kwargs) + + @classmethod + def create_for_modification(cls, **kwargs) -> "SigmaDataModelColumn": + return cls.updater(**kwargs) diff --git a/pyatlan_v9/model/assets/_overlays/sigma_dataset.py b/pyatlan_v9/model/assets/_overlays/sigma_dataset.py new file mode 100644 index 000000000..7f24f6969 --- /dev/null +++ b/pyatlan_v9/model/assets/_overlays/sigma_dataset.py @@ -0,0 +1,31 @@ +# INTERNAL_IMPORT: from pyatlan.utils import init_guid, validate_required_fields + + @classmethod + @init_guid + def creator( + cls, + *, + name: str, + connection_qualified_name: str, + dataset_id: str, + ) -> "SigmaDataset": + validate_required_fields( + ["name", "connection_qualified_name", "dataset_id"], + [name, connection_qualified_name, dataset_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}/{dataset_id}", + connection_qualified_name=connection_qualified_name, + connector_name=connector_name, + ) + + @classmethod + def create(cls, **kwargs) -> "SigmaDataset": + return cls.creator(**kwargs) + + @classmethod + def create_for_modification(cls, **kwargs) -> "SigmaDataset": + return cls.updater(**kwargs) diff --git a/pyatlan_v9/model/assets/_overlays/sigma_page.py b/pyatlan_v9/model/assets/_overlays/sigma_page.py new file mode 100644 index 000000000..3cbfef750 --- /dev/null +++ b/pyatlan_v9/model/assets/_overlays/sigma_page.py @@ -0,0 +1,39 @@ +# INTERNAL_IMPORT: from pyatlan.utils import init_guid, validate_required_fields + + @classmethod + @init_guid + def creator( + cls, + *, + name: str, + workbook_qualified_name: str, + workbook_name: str, + page_id: str, + ) -> "SigmaPage": + validate_required_fields( + ["name", "workbook_qualified_name", "workbook_name", "page_id"], + [name, workbook_qualified_name, workbook_name, page_id], + ) + fields = workbook_qualified_name.split("/") + if len(fields) != 4: + raise ValueError( + f"Invalid workbook_qualified_name: {workbook_qualified_name}. " + "Expected format: default/sigma//" + ) + return cls( + name=name, + qualified_name=f"{workbook_qualified_name}/{page_id}", + connection_qualified_name="/".join(fields[:3]), + connector_name=fields[1], + sigma_workbook_qualified_name=workbook_qualified_name, + sigma_workbook_name=workbook_name, + sigma_workbook=RelatedSigmaWorkbook(qualified_name=workbook_qualified_name), + ) + + @classmethod + def create(cls, **kwargs) -> "SigmaPage": + return cls.creator(**kwargs) + + @classmethod + def create_for_modification(cls, **kwargs) -> "SigmaPage": + return cls.updater(**kwargs) diff --git a/pyatlan_v9/model/assets/_overlays/sigma_workbook.py b/pyatlan_v9/model/assets/_overlays/sigma_workbook.py new file mode 100644 index 000000000..95cb9f9d4 --- /dev/null +++ b/pyatlan_v9/model/assets/_overlays/sigma_workbook.py @@ -0,0 +1,31 @@ +# INTERNAL_IMPORT: from pyatlan.utils import init_guid, validate_required_fields + + @classmethod + @init_guid + def creator( + cls, + *, + name: str, + connection_qualified_name: str, + workbook_id: str, + ) -> "SigmaWorkbook": + validate_required_fields( + ["name", "connection_qualified_name", "workbook_id"], + [name, connection_qualified_name, workbook_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}/{workbook_id}", + connection_qualified_name=connection_qualified_name, + connector_name=connector_name, + ) + + @classmethod + def create(cls, **kwargs) -> "SigmaWorkbook": + return cls.creator(**kwargs) + + @classmethod + def create_for_modification(cls, **kwargs) -> "SigmaWorkbook": + return cls.updater(**kwargs) diff --git a/pyatlan_v9/model/assets/sigma_data_element.py b/pyatlan_v9/model/assets/sigma_data_element.py index 4027390d5..96571c530 100644 --- a/pyatlan_v9/model/assets/sigma_data_element.py +++ b/pyatlan_v9/model/assets/sigma_data_element.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 @@ -349,6 +350,53 @@ def relate(self) -> "RelatedSigmaDataElement": return RelatedSigmaDataElement(guid=self.guid) return RelatedSigmaDataElement(qualified_name=self.qualified_name) + @classmethod + @init_guid + def creator( + cls, + *, + name: str, + page_qualified_name: str, + page_name: str, + workbook_name: str, + data_element_id: str, + ) -> "SigmaDataElement": + validate_required_fields( + [ + "name", + "page_qualified_name", + "page_name", + "workbook_name", + "data_element_id", + ], + [name, page_qualified_name, page_name, workbook_name, data_element_id], + ) + fields = page_qualified_name.split("/") + if len(fields) != 5: + raise ValueError( + f"Invalid page_qualified_name: {page_qualified_name}. " + "Expected format: default/sigma///" + ) + return cls( + name=name, + qualified_name=f"{page_qualified_name}/{data_element_id}", + connection_qualified_name="/".join(fields[:3]), + connector_name=fields[1], + sigma_workbook_qualified_name="/".join(fields[:4]), + sigma_workbook_name=workbook_name, + sigma_page_qualified_name=page_qualified_name, + sigma_page_name=page_name, + sigma_page=RelatedSigmaPage(qualified_name=page_qualified_name), + ) + + @classmethod + def create(cls, **kwargs) -> "SigmaDataElement": + return cls.creator(**kwargs) + + @classmethod + def create_for_modification(cls, **kwargs) -> "SigmaDataElement": + return cls.updater(**kwargs) + # ========================================================================= # Optimized Serialization Methods (override Asset base class) # ========================================================================= diff --git a/pyatlan_v9/model/assets/sigma_data_element_field.py b/pyatlan_v9/model/assets/sigma_data_element_field.py index 57ee9beeb..01fc813d2 100644 --- a/pyatlan_v9/model/assets/sigma_data_element_field.py +++ b/pyatlan_v9/model/assets/sigma_data_element_field.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 @@ -341,6 +342,67 @@ def relate(self) -> "RelatedSigmaDataElementField": return RelatedSigmaDataElementField(guid=self.guid) return RelatedSigmaDataElementField(qualified_name=self.qualified_name) + @classmethod + @init_guid + def creator( + cls, + *, + name: str, + data_element_qualified_name: str, + data_element_name: str, + page_name: str, + workbook_name: str, + field_id: str, + ) -> "SigmaDataElementField": + validate_required_fields( + [ + "name", + "data_element_qualified_name", + "data_element_name", + "page_name", + "workbook_name", + "field_id", + ], + [ + name, + data_element_qualified_name, + data_element_name, + page_name, + workbook_name, + field_id, + ], + ) + fields = data_element_qualified_name.split("/") + if len(fields) != 6: + raise ValueError( + f"Invalid data_element_qualified_name: {data_element_qualified_name}. " + "Expected format: " + "default/sigma////" + ) + return cls( + name=name, + qualified_name=f"{data_element_qualified_name}/{field_id}", + connection_qualified_name="/".join(fields[:3]), + connector_name=fields[1], + sigma_workbook_qualified_name="/".join(fields[:4]), + sigma_workbook_name=workbook_name, + sigma_page_qualified_name="/".join(fields[:5]), + sigma_page_name=page_name, + sigma_data_element_qualified_name=data_element_qualified_name, + sigma_data_element_name=data_element_name, + sigma_data_element=RelatedSigmaDataElement( + qualified_name=data_element_qualified_name + ), + ) + + @classmethod + def create(cls, **kwargs) -> "SigmaDataElementField": + return cls.creator(**kwargs) + + @classmethod + def create_for_modification(cls, **kwargs) -> "SigmaDataElementField": + return cls.updater(**kwargs) + # ========================================================================= # Optimized Serialization Methods (override Asset base class) # ========================================================================= diff --git a/pyatlan_v9/model/assets/sigma_data_model.py b/pyatlan_v9/model/assets/sigma_data_model.py index ac2a2cda0..18e43e1ae 100644 --- a/pyatlan_v9/model/assets/sigma_data_model.py +++ b/pyatlan_v9/model/assets/sigma_data_model.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 @@ -278,6 +280,8 @@ def __post_init__(self) -> None: # SDK Methods # ========================================================================= + _QUALIFIED_NAME_PATTERN: ClassVar[re.Pattern] = re.compile(r"^.+/dataModels/[^/]+$") + def validate(self, for_creation: bool = False) -> None: """ Dry-run validation of this SigmaDataModel instance. @@ -302,6 +306,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"SigmaDataModel validation failed: {errors}") @@ -334,6 +343,37 @@ def relate(self) -> "RelatedSigmaDataModel": return RelatedSigmaDataModel(guid=self.guid) return RelatedSigmaDataModel(qualified_name=self.qualified_name) + @classmethod + @init_guid + def creator( + cls, + *, + name: str, + connection_qualified_name: str, + data_model_id: str, + ) -> "SigmaDataModel": + validate_required_fields( + ["name", "connection_qualified_name", "data_model_id"], + [name, connection_qualified_name, data_model_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}/dataModels/{data_model_id}", + connection_qualified_name=connection_qualified_name, + connector_name=connector_name, + asset_source_id=data_model_id, + ) + + @classmethod + def create(cls, **kwargs) -> "SigmaDataModel": + return cls.creator(**kwargs) + + @classmethod + def create_for_modification(cls, **kwargs) -> "SigmaDataModel": + return cls.updater(**kwargs) + # ========================================================================= # Optimized Serialization Methods (override Asset base class) # ========================================================================= diff --git a/pyatlan_v9/model/assets/sigma_data_model_column.py b/pyatlan_v9/model/assets/sigma_data_model_column.py index 0917f1ff4..52f274f9a 100644 --- a/pyatlan_v9/model/assets/sigma_data_model_column.py +++ b/pyatlan_v9/model/assets/sigma_data_model_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 @@ -265,7 +266,9 @@ 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"^.+/dataModels/[^/]+/[^/]+/[^/]+$" + ) def validate(self, for_creation: bool = False) -> None: """ @@ -339,6 +342,54 @@ def relate(self) -> "RelatedSigmaDataModelColumn": return RelatedSigmaDataModelColumn(guid=self.guid) return RelatedSigmaDataModelColumn(qualified_name=self.qualified_name) + @classmethod + @init_guid + def creator( + cls, + *, + name: str, + data_model_qualified_name: str, + data_model_name: str, + element_id: str, + column_id: str, + ) -> "SigmaDataModelColumn": + validate_required_fields( + [ + "name", + "data_model_qualified_name", + "data_model_name", + "element_id", + "column_id", + ], + [name, data_model_qualified_name, data_model_name, element_id, column_id], + ) + fields = data_model_qualified_name.split("/") + if len(fields) != 5 or fields[3] != "dataModels": + raise ValueError( + f"Invalid data_model_qualified_name: {data_model_qualified_name}. " + "Expected format: default/sigma//dataModels/" + ) + return cls( + name=name, + qualified_name=f"{data_model_qualified_name}/{element_id}/{column_id}", + connection_qualified_name="/".join(fields[:3]), + connector_name=fields[1], + asset_source_id=column_id, + sigma_data_model_qualified_name=data_model_qualified_name, + sigma_data_model_name=data_model_name, + sigma_data_model=RelatedSigmaDataModel( + qualified_name=data_model_qualified_name + ), + ) + + @classmethod + def create(cls, **kwargs) -> "SigmaDataModelColumn": + return cls.creator(**kwargs) + + @classmethod + def create_for_modification(cls, **kwargs) -> "SigmaDataModelColumn": + return cls.updater(**kwargs) + # ========================================================================= # Optimized Serialization Methods (override Asset base class) # ========================================================================= diff --git a/pyatlan_v9/model/assets/sigma_dataset.py b/pyatlan_v9/model/assets/sigma_dataset.py index 867902120..8978a6cdc 100644 --- a/pyatlan_v9/model/assets/sigma_dataset.py +++ b/pyatlan_v9/model/assets/sigma_dataset.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 @@ -254,6 +256,8 @@ def __post_init__(self) -> None: # SDK Methods # ========================================================================= + _QUALIFIED_NAME_PATTERN: ClassVar[re.Pattern] = re.compile(r"^.+/[^/]+$") + def validate(self, for_creation: bool = False) -> None: """ Dry-run validation of this SigmaDataset instance. @@ -278,6 +282,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"SigmaDataset validation failed: {errors}") @@ -310,6 +319,36 @@ def relate(self) -> "RelatedSigmaDataset": return RelatedSigmaDataset(guid=self.guid) return RelatedSigmaDataset(qualified_name=self.qualified_name) + @classmethod + @init_guid + def creator( + cls, + *, + name: str, + connection_qualified_name: str, + dataset_id: str, + ) -> "SigmaDataset": + validate_required_fields( + ["name", "connection_qualified_name", "dataset_id"], + [name, connection_qualified_name, dataset_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}/{dataset_id}", + connection_qualified_name=connection_qualified_name, + connector_name=connector_name, + ) + + @classmethod + def create(cls, **kwargs) -> "SigmaDataset": + return cls.creator(**kwargs) + + @classmethod + def create_for_modification(cls, **kwargs) -> "SigmaDataset": + return cls.updater(**kwargs) + # ========================================================================= # Optimized Serialization Methods (override Asset base class) # ========================================================================= diff --git a/pyatlan_v9/model/assets/sigma_page.py b/pyatlan_v9/model/assets/sigma_page.py index cb7ee07c4..d6d6f2dcb 100644 --- a/pyatlan_v9/model/assets/sigma_page.py +++ b/pyatlan_v9/model/assets/sigma_page.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 @@ -333,6 +334,44 @@ def relate(self) -> "RelatedSigmaPage": return RelatedSigmaPage(guid=self.guid) return RelatedSigmaPage(qualified_name=self.qualified_name) + @classmethod + @init_guid + def creator( + cls, + *, + name: str, + workbook_qualified_name: str, + workbook_name: str, + page_id: str, + ) -> "SigmaPage": + validate_required_fields( + ["name", "workbook_qualified_name", "workbook_name", "page_id"], + [name, workbook_qualified_name, workbook_name, page_id], + ) + fields = workbook_qualified_name.split("/") + if len(fields) != 4: + raise ValueError( + f"Invalid workbook_qualified_name: {workbook_qualified_name}. " + "Expected format: default/sigma//" + ) + return cls( + name=name, + qualified_name=f"{workbook_qualified_name}/{page_id}", + connection_qualified_name="/".join(fields[:3]), + connector_name=fields[1], + sigma_workbook_qualified_name=workbook_qualified_name, + sigma_workbook_name=workbook_name, + sigma_workbook=RelatedSigmaWorkbook(qualified_name=workbook_qualified_name), + ) + + @classmethod + def create(cls, **kwargs) -> "SigmaPage": + return cls.creator(**kwargs) + + @classmethod + def create_for_modification(cls, **kwargs) -> "SigmaPage": + return cls.updater(**kwargs) + # ========================================================================= # Optimized Serialization Methods (override Asset base class) # ========================================================================= diff --git a/pyatlan_v9/model/assets/sigma_workbook.py b/pyatlan_v9/model/assets/sigma_workbook.py index a3cff7bc3..598ba7ff0 100644 --- a/pyatlan_v9/model/assets/sigma_workbook.py +++ b/pyatlan_v9/model/assets/sigma_workbook.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 @@ -252,6 +254,8 @@ def __post_init__(self) -> None: # SDK Methods # ========================================================================= + _QUALIFIED_NAME_PATTERN: ClassVar[re.Pattern] = re.compile(r"^.+/[^/]+$") + def validate(self, for_creation: bool = False) -> None: """ Dry-run validation of this SigmaWorkbook instance. @@ -276,6 +280,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"SigmaWorkbook validation failed: {errors}") @@ -308,6 +317,36 @@ def relate(self) -> "RelatedSigmaWorkbook": return RelatedSigmaWorkbook(guid=self.guid) return RelatedSigmaWorkbook(qualified_name=self.qualified_name) + @classmethod + @init_guid + def creator( + cls, + *, + name: str, + connection_qualified_name: str, + workbook_id: str, + ) -> "SigmaWorkbook": + validate_required_fields( + ["name", "connection_qualified_name", "workbook_id"], + [name, connection_qualified_name, workbook_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}/{workbook_id}", + connection_qualified_name=connection_qualified_name, + connector_name=connector_name, + ) + + @classmethod + def create(cls, **kwargs) -> "SigmaWorkbook": + return cls.creator(**kwargs) + + @classmethod + def create_for_modification(cls, **kwargs) -> "SigmaWorkbook": + return cls.updater(**kwargs) + # ========================================================================= # Optimized Serialization Methods (override Asset base class) # ========================================================================= diff --git a/tests_v9/integration/sigma_asset_test.py b/tests_v9/integration/sigma_asset_test.py new file mode 100644 index 000000000..7414de3ec --- /dev/null +++ b/tests_v9/integration/sigma_asset_test.py @@ -0,0 +1,240 @@ +# 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 ( + Connection, + SigmaDataElement, + SigmaDataElementField, + SigmaDataModel, + SigmaDataModelColumn, + SigmaDataset, + SigmaPage, + SigmaWorkbook, +) +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("SIGMA") + +WORKBOOK_ID = f"wb-{MODULE_NAME}" +PAGE_ID = f"pg-{MODULE_NAME}" +DATA_ELEMENT_ID = f"el-{MODULE_NAME}" +FIELD_ID = f"fld-{MODULE_NAME}" +DATASET_ID = f"ds-{MODULE_NAME}" +DATA_MODEL_ID = f"dm-{MODULE_NAME}" +ELEMENT_ID = f"dmel-{MODULE_NAME}" +COLUMN_ID = f"col-{MODULE_NAME}" + + +def _save(client: AtlanClient, to_create, asset_type): + return client.asset.save(to_create).assets_created(asset_type=asset_type)[0] + + +@pytest.fixture(scope="module") +def connection(client: AtlanClient) -> Generator[Connection, None, None]: + result = create_connection( + client=client, name=MODULE_NAME, connector_type=AtlanConnectorType.SIGMA + ) + yield result + delete_asset(client, guid=result.guid, asset_type=Connection) + + +@pytest.fixture(scope="module") +def workbook( + client: AtlanClient, connection: Connection +) -> Generator[SigmaWorkbook, None, None]: + to_create = SigmaWorkbook.creator( + name=f"workbook_{MODULE_NAME}", + connection_qualified_name=connection.qualified_name, + workbook_id=WORKBOOK_ID, + ) + result = _save(client, to_create, SigmaWorkbook) + yield result + delete_asset(client, guid=result.guid, asset_type=SigmaWorkbook) + + +@pytest.fixture(scope="module") +def page( + client: AtlanClient, workbook: SigmaWorkbook +) -> Generator[SigmaPage, None, None]: + to_create = SigmaPage.creator( + name=f"page_{MODULE_NAME}", + workbook_qualified_name=workbook.qualified_name, + workbook_name=workbook.name, + page_id=PAGE_ID, + ) + result = _save(client, to_create, SigmaPage) + yield result + delete_asset(client, guid=result.guid, asset_type=SigmaPage) + + +@pytest.fixture(scope="module") +def data_element( + client: AtlanClient, workbook: SigmaWorkbook, page: SigmaPage +) -> Generator[SigmaDataElement, None, None]: + to_create = SigmaDataElement.creator( + name=f"data_element_{MODULE_NAME}", + page_qualified_name=page.qualified_name, + page_name=page.name, + workbook_name=workbook.name, + data_element_id=DATA_ELEMENT_ID, + ) + result = _save(client, to_create, SigmaDataElement) + yield result + delete_asset(client, guid=result.guid, asset_type=SigmaDataElement) + + +@pytest.fixture(scope="module") +def data_element_field( + client: AtlanClient, + workbook: SigmaWorkbook, + page: SigmaPage, + data_element: SigmaDataElement, +) -> Generator[SigmaDataElementField, None, None]: + to_create = SigmaDataElementField.creator( + name=f"field_{MODULE_NAME}", + data_element_qualified_name=data_element.qualified_name, + data_element_name=data_element.name, + page_name=page.name, + workbook_name=workbook.name, + field_id=FIELD_ID, + ) + result = _save(client, to_create, SigmaDataElementField) + yield result + delete_asset(client, guid=result.guid, asset_type=SigmaDataElementField) + + +@pytest.fixture(scope="module") +def dataset( + client: AtlanClient, connection: Connection +) -> Generator[SigmaDataset, None, None]: + to_create = SigmaDataset.creator( + name=f"dataset_{MODULE_NAME}", + connection_qualified_name=connection.qualified_name, + dataset_id=DATASET_ID, + ) + result = _save(client, to_create, SigmaDataset) + yield result + delete_asset(client, guid=result.guid, asset_type=SigmaDataset) + + +@pytest.fixture(scope="module") +def data_model( + client: AtlanClient, connection: Connection +) -> Generator[SigmaDataModel, None, None]: + to_create = SigmaDataModel.creator( + name=f"data_model_{MODULE_NAME}", + connection_qualified_name=connection.qualified_name, + data_model_id=DATA_MODEL_ID, + ) + result = _save(client, to_create, SigmaDataModel) + yield result + delete_asset(client, guid=result.guid, asset_type=SigmaDataModel) + + +@pytest.fixture(scope="module") +def data_model_column( + client: AtlanClient, data_model: SigmaDataModel +) -> Generator[SigmaDataModelColumn, None, None]: + to_create = SigmaDataModelColumn.creator( + name=f"column_{MODULE_NAME}", + data_model_qualified_name=data_model.qualified_name, + data_model_name=data_model.name, + element_id=ELEMENT_ID, + column_id=COLUMN_ID, + ) + result = _save(client, to_create, SigmaDataModelColumn) + yield result + delete_asset(client, guid=result.guid, asset_type=SigmaDataModelColumn) + + +def _assert_retrieved(client, asset, asset_type, connection, expected_suffix): + retrieved = client.asset.get_by_guid( + asset.guid, asset_type=asset_type, ignore_relationships=False + ) + assert retrieved.qualified_name == f"{connection.qualified_name}/{expected_suffix}" + assert retrieved.name == asset.name + assert retrieved.connector_name == AtlanConnectorType.SIGMA + assert retrieved.connection_qualified_name == connection.qualified_name + return retrieved + + +def test_workbook(client: AtlanClient, connection: Connection, workbook): + _assert_retrieved(client, workbook, SigmaWorkbook, connection, WORKBOOK_ID) + + +def test_page(client: AtlanClient, connection: Connection, workbook, page): + retrieved = _assert_retrieved( + client, page, SigmaPage, connection, f"{WORKBOOK_ID}/{PAGE_ID}" + ) + assert retrieved.sigma_workbook.guid == workbook.guid + assert retrieved.sigma_workbook_qualified_name == workbook.qualified_name + assert retrieved.sigma_workbook_name == workbook.name + + +def test_data_element( + client: AtlanClient, connection: Connection, workbook, page, data_element +): + retrieved = _assert_retrieved( + client, + data_element, + SigmaDataElement, + connection, + f"{WORKBOOK_ID}/{PAGE_ID}/{DATA_ELEMENT_ID}", + ) + assert retrieved.sigma_page.guid == page.guid + assert retrieved.sigma_page_qualified_name == page.qualified_name + assert retrieved.sigma_workbook_qualified_name == workbook.qualified_name + + +def test_data_element_field( + client: AtlanClient, + connection: Connection, + workbook, + page, + data_element, + data_element_field, +): + retrieved = _assert_retrieved( + client, + data_element_field, + SigmaDataElementField, + connection, + f"{WORKBOOK_ID}/{PAGE_ID}/{DATA_ELEMENT_ID}/{FIELD_ID}", + ) + assert retrieved.sigma_data_element.guid == data_element.guid + assert retrieved.sigma_data_element_qualified_name == data_element.qualified_name + assert retrieved.sigma_page_qualified_name == page.qualified_name + assert retrieved.sigma_workbook_qualified_name == workbook.qualified_name + + +def test_dataset(client: AtlanClient, connection: Connection, dataset): + _assert_retrieved(client, dataset, SigmaDataset, connection, DATASET_ID) + + +def test_data_model(client: AtlanClient, connection: Connection, data_model): + retrieved = _assert_retrieved( + client, data_model, SigmaDataModel, connection, f"dataModels/{DATA_MODEL_ID}" + ) + assert retrieved.asset_source_id == DATA_MODEL_ID + + +def test_data_model_column( + client: AtlanClient, connection: Connection, data_model, data_model_column +): + retrieved = _assert_retrieved( + client, + data_model_column, + SigmaDataModelColumn, + connection, + f"dataModels/{DATA_MODEL_ID}/{ELEMENT_ID}/{COLUMN_ID}", + ) + assert retrieved.sigma_data_model.guid == data_model.guid + assert retrieved.sigma_data_model_qualified_name == data_model.qualified_name + assert retrieved.asset_source_id == COLUMN_ID diff --git a/tests_v9/unit/model/sigma_test.py b/tests_v9/unit/model/sigma_test.py new file mode 100644 index 000000000..fd17e4dca --- /dev/null +++ b/tests_v9/unit/model/sigma_test.py @@ -0,0 +1,230 @@ +# SPDX-License-Identifier: Apache-2.0 +# Copyright 2026 Atlan Pte. Ltd. + +"""Unit tests for the Sigma creators in pyatlan_v9. + +The expected qualifiedNames are the exact strings atlan-sigma-app emits today +(FND-3838 census), so a creator change that would re-key existing Sigma assets fails here. +""" + +import pytest + +from pyatlan_v9.model import ( + SigmaDataElement, + SigmaDataElementField, + SigmaDataModel, + SigmaDataModelColumn, + SigmaDataset, + SigmaPage, + SigmaWorkbook, +) + +CONNECTION_QN = "default/sigma/1712345678" +WORKBOOK_ID = "3f2b9c1e-7a4d-4e8b-9c0d-1a2b3c4d5e6f" +PAGE_ID = "pg_Ab12Cd34" +DATA_ELEMENT_ID = "el-Xy9Zw8Vu" +FIELD_ID = "fld_Qr7St6" +DATASET_ID = "9e8d7c6b-5a4f-4e3d-8c2b-1a0f9e8d7c6b" +DATA_MODEL_ID = "dm-1a2b3c4d" +ELEMENT_ID = "dmel_Mn5Op4" +COLUMN_ID = "col_Gh3Ij2" + +WORKBOOK_QN = f"{CONNECTION_QN}/{WORKBOOK_ID}" +PAGE_QN = f"{WORKBOOK_QN}/{PAGE_ID}" +DATA_ELEMENT_QN = f"{PAGE_QN}/{DATA_ELEMENT_ID}" +DATA_MODEL_QN = f"{CONNECTION_QN}/dataModels/{DATA_MODEL_ID}" + +NAME = "Revenue" +WORKBOOK_NAME = "Sales Overview" +PAGE_NAME = "Summary" +DATA_ELEMENT_NAME = "Revenue by Region" +DATA_MODEL_NAME = "Orders Model" + +# (type, creator kwargs, the literal qualifiedName atlan-sigma-app emits) +ALL_TYPES = [ + ( + SigmaWorkbook, + dict(connection_qualified_name=CONNECTION_QN, workbook_id=WORKBOOK_ID), + "default/sigma/1712345678/3f2b9c1e-7a4d-4e8b-9c0d-1a2b3c4d5e6f", + ), + ( + SigmaDataset, + dict(connection_qualified_name=CONNECTION_QN, dataset_id=DATASET_ID), + "default/sigma/1712345678/9e8d7c6b-5a4f-4e3d-8c2b-1a0f9e8d7c6b", + ), + ( + SigmaDataModel, + dict(connection_qualified_name=CONNECTION_QN, data_model_id=DATA_MODEL_ID), + "default/sigma/1712345678/dataModels/dm-1a2b3c4d", + ), + ( + SigmaPage, + dict( + workbook_qualified_name=WORKBOOK_QN, + workbook_name=WORKBOOK_NAME, + page_id=PAGE_ID, + ), + "default/sigma/1712345678/3f2b9c1e-7a4d-4e8b-9c0d-1a2b3c4d5e6f/pg_Ab12Cd34", + ), + ( + SigmaDataElement, + dict( + page_qualified_name=PAGE_QN, + page_name=PAGE_NAME, + workbook_name=WORKBOOK_NAME, + data_element_id=DATA_ELEMENT_ID, + ), + "default/sigma/1712345678/3f2b9c1e-7a4d-4e8b-9c0d-1a2b3c4d5e6f/pg_Ab12Cd34" + "/el-Xy9Zw8Vu", + ), + ( + SigmaDataElementField, + dict( + data_element_qualified_name=DATA_ELEMENT_QN, + data_element_name=DATA_ELEMENT_NAME, + page_name=PAGE_NAME, + workbook_name=WORKBOOK_NAME, + field_id=FIELD_ID, + ), + "default/sigma/1712345678/3f2b9c1e-7a4d-4e8b-9c0d-1a2b3c4d5e6f/pg_Ab12Cd34" + "/el-Xy9Zw8Vu/fld_Qr7St6", + ), + ( + SigmaDataModelColumn, + dict( + data_model_qualified_name=DATA_MODEL_QN, + data_model_name=DATA_MODEL_NAME, + element_id=ELEMENT_ID, + column_id=COLUMN_ID, + ), + "default/sigma/1712345678/dataModels/dm-1a2b3c4d/dmel_Mn5Op4/col_Gh3Ij2", + ), +] +IDS = [t[0].__name__ for t in ALL_TYPES] + + +@pytest.mark.parametrize("asset_type, kwargs, expected_qn", ALL_TYPES, ids=IDS) +def test_creator_matches_app_qualified_name(asset_type, kwargs, expected_qn): + sut = asset_type.creator(name=NAME, **kwargs) + + assert sut.qualified_name == expected_qn + assert sut.name == NAME + assert sut.connection_qualified_name == CONNECTION_QN + assert sut.connector_name == "sigma" + assert sut._QUALIFIED_NAME_PATTERN.match(sut.qualified_name) + sut.validate(for_creation=True) + + +@pytest.mark.parametrize("asset_type, kwargs, expected_qn", ALL_TYPES, ids=IDS) +def test_create_alias(asset_type, kwargs, expected_qn): + assert asset_type.create(name=NAME, **kwargs).qualified_name == expected_qn + + +@pytest.mark.parametrize("asset_type, kwargs, _", ALL_TYPES, ids=IDS) +def test_creator_rejects_missing_or_blank_inputs(asset_type, kwargs, _): + for param in ["name", *kwargs]: + args = {"name": NAME, **kwargs} + args[param] = None + with pytest.raises(ValueError, match=f"{param} is required"): + asset_type.creator(**args) + args[param] = " " + with pytest.raises(ValueError, match=f"{param} cannot be blank"): + asset_type.creator(**args) + + +def test_page_sets_workbook(): + sut = SigmaPage.creator( + name=NAME, + workbook_qualified_name=WORKBOOK_QN, + workbook_name=WORKBOOK_NAME, + page_id=PAGE_ID, + ) + + assert sut.sigma_workbook.qualified_name == WORKBOOK_QN + assert sut.sigma_workbook_qualified_name == WORKBOOK_QN + assert sut.sigma_workbook_name == WORKBOOK_NAME + + +def test_data_element_sets_page_and_workbook(): + sut = SigmaDataElement.creator( + name=NAME, + page_qualified_name=PAGE_QN, + page_name=PAGE_NAME, + workbook_name=WORKBOOK_NAME, + data_element_id=DATA_ELEMENT_ID, + ) + + assert sut.sigma_page.qualified_name == PAGE_QN + assert sut.sigma_page_qualified_name == PAGE_QN + assert sut.sigma_page_name == PAGE_NAME + assert sut.sigma_workbook_qualified_name == WORKBOOK_QN + assert sut.sigma_workbook_name == WORKBOOK_NAME + + +def test_data_element_field_sets_all_ancestors(): + sut = SigmaDataElementField.creator( + name=NAME, + data_element_qualified_name=DATA_ELEMENT_QN, + data_element_name=DATA_ELEMENT_NAME, + page_name=PAGE_NAME, + workbook_name=WORKBOOK_NAME, + field_id=FIELD_ID, + ) + + assert sut.sigma_data_element.qualified_name == DATA_ELEMENT_QN + assert sut.sigma_data_element_qualified_name == DATA_ELEMENT_QN + assert sut.sigma_data_element_name == DATA_ELEMENT_NAME + assert sut.sigma_page_qualified_name == PAGE_QN + assert sut.sigma_page_name == PAGE_NAME + assert sut.sigma_workbook_qualified_name == WORKBOOK_QN + assert sut.sigma_workbook_name == WORKBOOK_NAME + + +def test_data_model_sets_asset_source_id(): + sut = SigmaDataModel.creator( + name=NAME, connection_qualified_name=CONNECTION_QN, data_model_id=DATA_MODEL_ID + ) + + assert sut.asset_source_id == DATA_MODEL_ID + + +def test_data_model_column_sets_data_model(): + sut = SigmaDataModelColumn.creator( + name=NAME, + data_model_qualified_name=DATA_MODEL_QN, + data_model_name=DATA_MODEL_NAME, + element_id=ELEMENT_ID, + column_id=COLUMN_ID, + ) + + assert sut.sigma_data_model.qualified_name == DATA_MODEL_QN + assert sut.sigma_data_model_qualified_name == DATA_MODEL_QN + assert sut.sigma_data_model_name == DATA_MODEL_NAME + assert sut.asset_source_id == COLUMN_ID + + +@pytest.mark.parametrize( + "asset_type, parent_param, bad_parent_qn", + [ + (SigmaPage, "workbook_qualified_name", CONNECTION_QN), + (SigmaPage, "workbook_qualified_name", PAGE_QN), + (SigmaDataElement, "page_qualified_name", WORKBOOK_QN), + (SigmaDataElement, "page_qualified_name", DATA_ELEMENT_QN), + (SigmaDataElementField, "data_element_qualified_name", PAGE_QN), + (SigmaDataElementField, "data_element_qualified_name", f"{DATA_ELEMENT_QN}/x"), + (SigmaDataModelColumn, "data_model_qualified_name", CONNECTION_QN), + # a page qn has the data model's segment count but not the `dataModels` literal + (SigmaDataModelColumn, "data_model_qualified_name", PAGE_QN), + ], +) +def test_creator_rejects_malformed_parent_qualified_name( + asset_type, parent_param, bad_parent_qn +): + kwargs = next(k for t, k, _ in ALL_TYPES if t is asset_type) + with pytest.raises(ValueError, match=f"Invalid {parent_param}"): + asset_type.creator(name=NAME, **{**kwargs, parent_param: bad_parent_qn}) + + +def test_data_model_patterns_require_data_models_segment(): + assert not SigmaDataModel._QUALIFIED_NAME_PATTERN.match(PAGE_QN) + assert not SigmaDataModelColumn._QUALIFIED_NAME_PATTERN.match(DATA_ELEMENT_QN)