FND-3856: add pyatlan_v9 creators for the 7 Sigma asset types

pyatlan_v9/model/assets/_overlays/sigma_data_element.py

@@ -0,0 +1,58 @@
+# 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/<epoch>/<workbook_id>/<page_id>"
+            )
+        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(
+                unique_attributes={"qualifiedName": page_qualified_name}
+            ),
+        )
+
+    @classmethod
+    def create(cls, **kwargs) -> "SigmaDataElement":
+        return cls.creator(**kwargs)
+
+    @classmethod
+    def updater(cls, *, qualified_name: str, name: str) -> "SigmaDataElement":
+        validate_required_fields(["qualified_name", "name"], [qualified_name, name])
+        return cls(qualified_name=qualified_name, name=name)
+
+    @classmethod
+    def create_for_modification(cls, **kwargs) -> "SigmaDataElement":
+        return cls.updater(**kwargs)
+
+    def trim_to_required(self) -> "SigmaDataElement":
+        return SigmaDataElement.updater(qualified_name=self.qualified_name, name=self.name)

pyatlan_v9/model/assets/_overlays/sigma_data_element_field.py

@@ -0,0 +1,70 @@
+# 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/<epoch>/<workbook_id>/<page_id>/<data_element_id>"
+            )
+        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(
+                unique_attributes={"qualifiedName": data_element_qualified_name}
+            ),
+        )
+
+    @classmethod
+    def create(cls, **kwargs) -> "SigmaDataElementField":
+        return cls.creator(**kwargs)
+
+    @classmethod
+    def updater(cls, *, qualified_name: str, name: str) -> "SigmaDataElementField":
+        validate_required_fields(["qualified_name", "name"], [qualified_name, name])
+        return cls(qualified_name=qualified_name, name=name)
+
+    @classmethod
+    def create_for_modification(cls, **kwargs) -> "SigmaDataElementField":
+        return cls.updater(**kwargs)
+
+    def trim_to_required(self) -> "SigmaDataElementField":
+        return SigmaDataElementField.updater(qualified_name=self.qualified_name, name=self.name)

pyatlan_v9/model/assets/_overlays/sigma_data_model.py

@@ -0,0 +1,40 @@
+# 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 updater(cls, *, qualified_name: str, name: str) -> "SigmaDataModel":
+        validate_required_fields(["qualified_name", "name"], [qualified_name, name])
+        return cls(qualified_name=qualified_name, name=name)
+
+    @classmethod
+    def create_for_modification(cls, **kwargs) -> "SigmaDataModel":
+        return cls.updater(**kwargs)
+
+    def trim_to_required(self) -> "SigmaDataModel":
+        return SigmaDataModel.updater(qualified_name=self.qualified_name, name=self.name)

pyatlan_v9/model/assets/_overlays/sigma_data_model_column.py

@@ -0,0 +1,57 @@
+# 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/<epoch>/dataModels/<data_model_id>"
+            )
+        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(
+                unique_attributes={"qualifiedName": data_model_qualified_name}
+            ),
+        )
+
+    @classmethod
+    def create(cls, **kwargs) -> "SigmaDataModelColumn":
+        return cls.creator(**kwargs)
+
+    @classmethod
+    def updater(cls, *, qualified_name: str, name: str) -> "SigmaDataModelColumn":
+        validate_required_fields(["qualified_name", "name"], [qualified_name, name])
+        return cls(qualified_name=qualified_name, name=name)
+
+    @classmethod
+    def create_for_modification(cls, **kwargs) -> "SigmaDataModelColumn":
+        return cls.updater(**kwargs)
+
+    def trim_to_required(self) -> "SigmaDataModelColumn":
+        return SigmaDataModelColumn.updater(qualified_name=self.qualified_name, name=self.name)

pyatlan_v9/model/assets/_overlays/sigma_dataset.py

@@ -0,0 +1,39 @@
+# 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 updater(cls, *, qualified_name: str, name: str) -> "SigmaDataset":
+        validate_required_fields(["qualified_name", "name"], [qualified_name, name])
+        return cls(qualified_name=qualified_name, name=name)
+
+    @classmethod
+    def create_for_modification(cls, **kwargs) -> "SigmaDataset":
+        return cls.updater(**kwargs)
+
+    def trim_to_required(self) -> "SigmaDataset":
+        return SigmaDataset.updater(qualified_name=self.qualified_name, name=self.name)

pyatlan_v9/model/assets/_overlays/sigma_page.py

@@ -0,0 +1,49 @@
+# 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/<epoch>/<workbook_id>"
+            )
+        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(
+                unique_attributes={"qualifiedName": workbook_qualified_name}
+            ),
+        )
+
+    @classmethod
+    def create(cls, **kwargs) -> "SigmaPage":
+        return cls.creator(**kwargs)
+
+    @classmethod
+    def updater(cls, *, qualified_name: str, name: str) -> "SigmaPage":
+        validate_required_fields(["qualified_name", "name"], [qualified_name, name])
+        return cls(qualified_name=qualified_name, name=name)
+
+    @classmethod
+    def create_for_modification(cls, **kwargs) -> "SigmaPage":
+        return cls.updater(**kwargs)
+
+    def trim_to_required(self) -> "SigmaPage":
+        return SigmaPage.updater(qualified_name=self.qualified_name, name=self.name)

pyatlan_v9/model/assets/_overlays/sigma_workbook.py

@@ -0,0 +1,39 @@
+# 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 updater(cls, *, qualified_name: str, name: str) -> "SigmaWorkbook":
+        validate_required_fields(["qualified_name", "name"], [qualified_name, name])
+        return cls(qualified_name=qualified_name, name=name)
+
+    @classmethod
+    def create_for_modification(cls, **kwargs) -> "SigmaWorkbook":
+        return cls.updater(**kwargs)
+
+    def trim_to_required(self) -> "SigmaWorkbook":
+        return SigmaWorkbook.updater(qualified_name=self.qualified_name, name=self.name)