FND-3849: add pyatlan_v9 creators for the 10 Tableau types

pyatlan_v9/model/assets/_overlays/tableau_calculated_field.py

@@ -0,0 +1,69 @@
+# STDLIB_IMPORT: from typing import Optional
+# INTERNAL_IMPORT: from pyatlan.utils import init_guid, validate_required_fields
+
+    @classmethod
+    @init_guid
+    def creator(
+        cls,
+        *,
+        name: str,
+        datasource_qualified_name: str,
+        field_id: str,
+        top_level_project_qualified_name: Optional[str] = None,
+    ) -> "TableauCalculatedField":
+        """
+        Builds a field in a published datasource
+        (default/tableau/<epoch>/<site_id>/<project_id>/<datasource_id>) or in an
+        embedded one (.../<project_id>/<workbook_id>/<datasource_id>).
+        """
+        validate_required_fields(
+            ["name", "datasource_qualified_name", "field_id"],
+            [name, datasource_qualified_name, field_id],
+        )
+        fields = datasource_qualified_name.split("/")
+        if len(fields) not in (6, 7):
+            raise ValueError(
+                f"Invalid datasource_qualified_name: {datasource_qualified_name}. "
+                "Expected format: "
+                "default/tableau/<epoch>/<site_id>/<project_id>/<datasource_id> "
+                "(published) or "
+                "default/tableau/<epoch>/<site_id>/<project_id>/<workbook_id>/<datasource_id> "
+                "(embedded)"
+            )
+        site_qn = "/".join(fields[:4])
+        project_qn = "/".join(fields[:5])
+        if top_level_project_qualified_name is not None and (
+            len(top_level_project_qualified_name.split("/")) != 5
+            or not top_level_project_qualified_name.startswith(f"{site_qn}/")
+        ):
+            raise ValueError(
+                f"Invalid top_level_project_qualified_name: "
+                f"{top_level_project_qualified_name}. "
+                f"Expected format: {site_qn}/<project_id>"
+            )
+        field = cls(
+            name=name,
+            qualified_name=f"{datasource_qualified_name}/{field_id}",
+            connection_qualified_name="/".join(fields[:3]),
+            connector_name=fields[1],
+            site_qualified_name=site_qn,
+            project_qualified_name=project_qn,
+            top_level_project_qualified_name=(
+                top_level_project_qualified_name or project_qn
+            ),
+            datasource_qualified_name=datasource_qualified_name,
+            datasource=RelatedTableauDatasource(
+                qualified_name=datasource_qualified_name
+            ),
+        )
+        if len(fields) == 7:
+            field.workbook_qualified_name = "/".join(fields[:6])
+        return field
+
+    @classmethod
+    def create(cls, **kwargs) -> "TableauCalculatedField":
+        return cls.creator(**kwargs)
+
+    @classmethod
+    def create_for_modification(cls, **kwargs) -> "TableauCalculatedField":
+        return cls.updater(**kwargs)

pyatlan_v9/model/assets/_overlays/tableau_dashboard.py

@@ -0,0 +1,56 @@
+# STDLIB_IMPORT: from typing import Optional
+# INTERNAL_IMPORT: from pyatlan.utils import init_guid, validate_required_fields
+
+    @classmethod
+    @init_guid
+    def creator(
+        cls,
+        *,
+        name: str,
+        workbook_qualified_name: str,
+        dashboard_id: str,
+        top_level_project_qualified_name: Optional[str] = None,
+    ) -> "TableauDashboard":
+        validate_required_fields(
+            ["name", "workbook_qualified_name", "dashboard_id"],
+            [name, workbook_qualified_name, dashboard_id],
+        )
+        fields = workbook_qualified_name.split("/")
+        if len(fields) != 6:
+            raise ValueError(
+                f"Invalid workbook_qualified_name: {workbook_qualified_name}. "
+                "Expected format: "
+                "default/tableau/<epoch>/<site_id>/<project_id>/<workbook_id>"
+            )
+        site_qn = "/".join(fields[:4])
+        project_qn = "/".join(fields[:5])
+        if top_level_project_qualified_name is not None and (
+            len(top_level_project_qualified_name.split("/")) != 5
+            or not top_level_project_qualified_name.startswith(f"{site_qn}/")
+        ):
+            raise ValueError(
+                f"Invalid top_level_project_qualified_name: "
+                f"{top_level_project_qualified_name}. "
+                f"Expected format: {site_qn}/<project_id>"
+            )
+        return cls(
+            name=name,
+            qualified_name=f"{workbook_qualified_name}/{dashboard_id}",
+            connection_qualified_name="/".join(fields[:3]),
+            connector_name=fields[1],
+            site_qualified_name=site_qn,
+            project_qualified_name=project_qn,
+            top_level_project_qualified_name=(
+                top_level_project_qualified_name or project_qn
+            ),
+            workbook_qualified_name=workbook_qualified_name,
+            workbook=RelatedTableauWorkbook(qualified_name=workbook_qualified_name),
+        )
+
+    @classmethod
+    def create(cls, **kwargs) -> "TableauDashboard":
+        return cls.creator(**kwargs)
+
+    @classmethod
+    def create_for_modification(cls, **kwargs) -> "TableauDashboard":
+        return cls.updater(**kwargs)

pyatlan_v9/model/assets/_overlays/tableau_dashboard_field.py

@@ -0,0 +1,61 @@
+# STDLIB_IMPORT: from typing import Optional
+# INTERNAL_IMPORT: from pyatlan.utils import init_guid, validate_required_fields
+
+    @classmethod
+    @init_guid
+    def creator(
+        cls,
+        *,
+        name: str,
+        dashboard_qualified_name: str,
+        field_id: str,
+        top_level_project_qualified_name: Optional[str] = None,
+    ) -> "TableauDashboardField":
+        """
+        field_id is the id of the datasource field this dashboard field uses.
+        """
+        validate_required_fields(
+            ["name", "dashboard_qualified_name", "field_id"],
+            [name, dashboard_qualified_name, field_id],
+        )
+        fields = dashboard_qualified_name.split("/")
+        if len(fields) != 7:
+            raise ValueError(
+                f"Invalid dashboard_qualified_name: {dashboard_qualified_name}. "
+                "Expected format: default/tableau/<epoch>/<site_id>/<project_id>/"
+                "<workbook_id>/<dashboard_id>"
+            )
+        site_qn = "/".join(fields[:4])
+        project_qn = "/".join(fields[:5])
+        if top_level_project_qualified_name is not None and (
+            len(top_level_project_qualified_name.split("/")) != 5
+            or not top_level_project_qualified_name.startswith(f"{site_qn}/")
+        ):
+            raise ValueError(
+                f"Invalid top_level_project_qualified_name: "
+                f"{top_level_project_qualified_name}. "
+                f"Expected format: {site_qn}/<project_id>"
+            )
+        return cls(
+            name=name,
+            qualified_name=f"{dashboard_qualified_name}/{field_id}",
+            connection_qualified_name="/".join(fields[:3]),
+            connector_name=fields[1],
+            tableau_site_qualified_name=site_qn,
+            tableau_project_qualified_name=project_qn,
+            tableau_top_level_project_qualified_name=(
+                top_level_project_qualified_name or project_qn
+            ),
+            tableau_dashboard_qualified_name=dashboard_qualified_name,
+            tableau_dashboard=RelatedTableauDashboard(
+                qualified_name=dashboard_qualified_name
+            ),
+        )
+
+    @classmethod
+    def create(cls, **kwargs) -> "TableauDashboardField":
+        return cls.creator(**kwargs)
+
+    @classmethod
+    def create_for_modification(cls, **kwargs) -> "TableauDashboardField":
+        return cls.updater(**kwargs)

pyatlan_v9/model/assets/_overlays/tableau_datasource.py

@@ -0,0 +1,73 @@
+# STDLIB_IMPORT: from typing import Optional
+# INTERNAL_IMPORT: from pyatlan.utils import init_guid, validate_required_fields
+
+    @classmethod
+    @init_guid
+    def creator(
+        cls,
+        *,
+        name: str,
+        project_qualified_name: str,
+        datasource_id: str,
+        workbook_qualified_name: Optional[str] = None,
+        top_level_project_qualified_name: Optional[str] = None,
+    ) -> "TableauDatasource":
+        """
+        Builds a published datasource, or an embedded one when
+        workbook_qualified_name is given (its qualifiedName then sits under the workbook).
+        """
+        validate_required_fields(
+            ["name", "project_qualified_name", "datasource_id"],
+            [name, project_qualified_name, datasource_id],
+        )
+        fields = project_qualified_name.split("/")
+        if len(fields) != 5:
+            raise ValueError(
+                f"Invalid project_qualified_name: {project_qualified_name}. "
+                "Expected format: default/tableau/<epoch>/<site_id>/<project_id>"
+            )
+        if workbook_qualified_name is not None and (
+            len(workbook_qualified_name.split("/")) != 6
+            or not workbook_qualified_name.startswith(f"{project_qualified_name}/")
+        ):
+            raise ValueError(
+                f"Invalid workbook_qualified_name: {workbook_qualified_name}. "
+                f"Expected format: {project_qualified_name}/<workbook_id>"
+            )
+        site_qn = "/".join(fields[:4])
+        if top_level_project_qualified_name is not None and (
+            len(top_level_project_qualified_name.split("/")) != 5
+            or not top_level_project_qualified_name.startswith(f"{site_qn}/")
+        ):
+            raise ValueError(
+                f"Invalid top_level_project_qualified_name: "
+                f"{top_level_project_qualified_name}. "
+                f"Expected format: {site_qn}/<project_id>"
+            )
+        datasource = cls(
+            name=name,
+            qualified_name=f"{workbook_qualified_name or project_qualified_name}/{datasource_id}",
+            connection_qualified_name="/".join(fields[:3]),
+            connector_name=fields[1],
+            site_qualified_name=site_qn,
+            project_qualified_name=project_qualified_name,
+            top_level_project_qualified_name=(
+                top_level_project_qualified_name or project_qualified_name
+            ),
+            is_published=workbook_qualified_name is None,
+            project=RelatedTableauProject(qualified_name=project_qualified_name),
+        )
+        if workbook_qualified_name:
+            datasource.workbook_qualified_name = workbook_qualified_name
+            datasource.workbook = RelatedTableauWorkbook(
+                qualified_name=workbook_qualified_name
+            )
+        return datasource
+
+    @classmethod
+    def create(cls, **kwargs) -> "TableauDatasource":
+        return cls.creator(**kwargs)
+
+    @classmethod
+    def create_for_modification(cls, **kwargs) -> "TableauDatasource":
+        return cls.updater(**kwargs)

pyatlan_v9/model/assets/_overlays/tableau_datasource_field.py

@@ -0,0 +1,69 @@
+# STDLIB_IMPORT: from typing import Optional
+# INTERNAL_IMPORT: from pyatlan.utils import init_guid, validate_required_fields
+
+    @classmethod
+    @init_guid
+    def creator(
+        cls,
+        *,
+        name: str,
+        datasource_qualified_name: str,
+        field_id: str,
+        top_level_project_qualified_name: Optional[str] = None,
+    ) -> "TableauDatasourceField":
+        """
+        Builds a field in a published datasource
+        (default/tableau/<epoch>/<site_id>/<project_id>/<datasource_id>) or in an
+        embedded one (.../<project_id>/<workbook_id>/<datasource_id>).
+        """
+        validate_required_fields(
+            ["name", "datasource_qualified_name", "field_id"],
+            [name, datasource_qualified_name, field_id],
+        )
+        fields = datasource_qualified_name.split("/")
+        if len(fields) not in (6, 7):
+            raise ValueError(
+                f"Invalid datasource_qualified_name: {datasource_qualified_name}. "
+                "Expected format: "
+                "default/tableau/<epoch>/<site_id>/<project_id>/<datasource_id> "
+                "(published) or "
+                "default/tableau/<epoch>/<site_id>/<project_id>/<workbook_id>/<datasource_id> "
+                "(embedded)"
+            )
+        site_qn = "/".join(fields[:4])
+        project_qn = "/".join(fields[:5])
+        if top_level_project_qualified_name is not None and (
+            len(top_level_project_qualified_name.split("/")) != 5
+            or not top_level_project_qualified_name.startswith(f"{site_qn}/")
+        ):
+            raise ValueError(
+                f"Invalid top_level_project_qualified_name: "
+                f"{top_level_project_qualified_name}. "
+                f"Expected format: {site_qn}/<project_id>"
+            )
+        field = cls(
+            name=name,
+            qualified_name=f"{datasource_qualified_name}/{field_id}",
+            connection_qualified_name="/".join(fields[:3]),
+            connector_name=fields[1],
+            site_qualified_name=site_qn,
+            project_qualified_name=project_qn,
+            top_level_project_qualified_name=(
+                top_level_project_qualified_name or project_qn
+            ),
+            datasource_qualified_name=datasource_qualified_name,
+            datasource=RelatedTableauDatasource(
+                qualified_name=datasource_qualified_name
+            ),
+        )
+        if len(fields) == 7:
+            field.workbook_qualified_name = "/".join(fields[:6])
+        return field
+
+    @classmethod
+    def create(cls, **kwargs) -> "TableauDatasourceField":
+        return cls.creator(**kwargs)
+
+    @classmethod
+    def create_for_modification(cls, **kwargs) -> "TableauDatasourceField":
+        return cls.updater(**kwargs)