FND-3857: pyatlan_v9 creators for the 7 Qlik Sense types

pyatlan_v9/model/assets/_overlays/qlik_app.py

@@ -0,0 +1,52 @@
+# INTERNAL_IMPORT: from pyatlan.utils import init_guid, validate_required_fields
+
+    @classmethod
+    @init_guid
+    def creator(
+        cls,
+        *,
+        name: str,
+        qlik_id: str,
+        space_qualified_name: str,
+    ) -> "QlikApp":
+        """
+        Build a Qlik app; qualifiedName is {connection}/app/{qlik_id}.
+
+        space_qualified_name is the qualifiedName of the QlikSpace
+        ({connection}/space/{id}) or QlikStream ({connection}/stream/{id})
+        the app lives in. Its segment decides which type the qlik_space
+        relationship points at.
+        """
+        validate_required_fields(
+            ["name", "qlik_id", "space_qualified_name"],
+            [name, qlik_id, space_qualified_name],
+        )
+        space_parts = space_qualified_name.rsplit("/", 2)
+        if len(space_parts) != 3 or space_parts[1] not in ("space", "stream"):
+            raise ValueError(
+                "space_qualified_name must end in /space/<id> or /stream/<id>"
+            )
+        connection_qualified_name, space_kind, space_id = space_parts
+        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}/app/{qlik_id}",
+            connection_qualified_name=connection_qualified_name,
+            connector_name=connector_name,
+            qlik_id=qlik_id,
+            qlik_space_id=space_id,
+            qlik_space_qualified_name=space_qualified_name,
+            qlik_space=RelatedQlikSpace(
+                type_name="QlikStream" if space_kind == "stream" else "QlikSpace",
+                qualified_name=space_qualified_name,
+            ),
+        )
+
+    @classmethod
+    def create(cls, **kwargs) -> "QlikApp":
+        return cls.creator(**kwargs)
+
+    @classmethod
+    def create_for_modification(cls, **kwargs) -> "QlikApp":
+        return cls.updater(**kwargs)

pyatlan_v9/model/assets/_overlays/qlik_chart.py

@@ -0,0 +1,58 @@
+# INTERNAL_IMPORT: from pyatlan.utils import init_guid, validate_required_fields
+
+    @classmethod
+    @init_guid
+    def creator(
+        cls,
+        *,
+        name: str,
+        qlik_id: str,
+        sheet_qualified_name: str,
+        space_qualified_name: Union[str, None] = None,
+    ) -> "QlikChart":
+        """
+        Build a Qlik chart; qualifiedName is {app_qualified_name}/chart/{qlik_id}.
+
+        The chart's qualifiedName hangs off the app, not the sheet: the app is
+        taken from sheet_qualified_name ({app_qualified_name}/sheet/{id}).
+        space_qualified_name is optional and only fills the denormalised
+        qlik_space_id / qlik_space_qualified_name of the app's space or stream.
+        """
+        validate_required_fields(
+            ["name", "qlik_id", "sheet_qualified_name"],
+            [name, qlik_id, sheet_qualified_name],
+        )
+        sheet_parts = sheet_qualified_name.rsplit("/", 4)
+        if len(sheet_parts) != 5 or sheet_parts[1] != "app" or sheet_parts[3] != "sheet":
+            raise ValueError("sheet_qualified_name must end in /app/<id>/sheet/<id>")
+        connection_qualified_name, _, app_id, _, _ = sheet_parts
+        app_qualified_name = f"{connection_qualified_name}/app/{app_id}"
+        fields = connection_qualified_name.split("/")
+        connector_name = fields[1] if len(fields) > 1 else None
+        asset = cls(
+            name=name,
+            qualified_name=f"{app_qualified_name}/chart/{qlik_id}",
+            connection_qualified_name=connection_qualified_name,
+            connector_name=connector_name,
+            qlik_id=qlik_id,
+            qlik_app_id=app_id,
+            qlik_app_qualified_name=app_qualified_name,
+            qlik_sheet=RelatedQlikSheet(qualified_name=sheet_qualified_name),
+        )
+        if space_qualified_name:
+            space_parts = space_qualified_name.rsplit("/", 2)
+            if len(space_parts) != 3 or space_parts[1] not in ("space", "stream"):
+                raise ValueError(
+                    "space_qualified_name must end in /space/<id> or /stream/<id>"
+                )
+            asset.qlik_space_id = space_parts[2]
+            asset.qlik_space_qualified_name = space_qualified_name
+        return asset
+
+    @classmethod
+    def create(cls, **kwargs) -> "QlikChart":
+        return cls.creator(**kwargs)
+
+    @classmethod
+    def create_for_modification(cls, **kwargs) -> "QlikChart":
+        return cls.updater(**kwargs)

pyatlan_v9/model/assets/_overlays/qlik_column.py

@@ -0,0 +1,81 @@
+# STDLIB_IMPORT: import hashlib
+# INTERNAL_IMPORT: from pyatlan.utils import init_guid, validate_required_fields
+
+    @classmethod
+    @init_guid
+    def creator(
+        cls,
+        *,
+        name: str,
+        parent_qualified_name: str,
+        space_qualified_name: Union[str, None] = None,
+    ) -> "QlikColumn":
+        """
+        Build a Qlik column under a sheet, chart or dataset.
+
+        qualifiedName is {parent_qualified_name}/column/{md5}, where md5 is taken
+        over "{parent path}/column/{name}" and the parent path is the parent's
+        qualifiedName without the connection (app/<id>/sheet/<id>,
+        app/<id>/chart/<id> or dataset/<id>), exactly as atlan-qlik-sense-app
+        builds it. The parent type is read from parent_qualified_name.
+        space_qualified_name is optional and only fills the denormalised
+        qlik_space_id / qlik_space_qualified_name of the space or stream.
+        """
+        validate_required_fields(
+            ["name", "parent_qualified_name"], [name, parent_qualified_name]
+        )
+        dataset_parts = parent_qualified_name.rsplit("/", 2)
+        app_parts = parent_qualified_name.rsplit("/", 4)
+        if len(dataset_parts) == 3 and dataset_parts[1] == "dataset":
+            connection_qualified_name, parent_kind = dataset_parts[0], "dataset"
+        elif (
+            len(app_parts) == 5
+            and app_parts[1] == "app"
+            and app_parts[3] in ("sheet", "chart")
+        ):
+            connection_qualified_name, parent_kind = app_parts[0], app_parts[3]
+        else:
+            raise ValueError(
+                "parent_qualified_name must end in /app/<id>/sheet/<id>, "
+                "/app/<id>/chart/<id> or /dataset/<id>"
+            )
+        parent_path = parent_qualified_name[len(connection_qualified_name) + 1 :]
+        # deepcode ignore InsecureHash/test: an identifier, not a security key
+        column_hash = hashlib.md5(  # noqa: S324
+            f"{parent_path}/column/{name}".encode("utf-8")
+        ).hexdigest()
+        fields = connection_qualified_name.split("/")
+        connector_name = fields[1] if len(fields) > 1 else None
+        asset = cls(
+            name=name,
+            qualified_name=f"{parent_qualified_name}/column/{column_hash}",
+            connection_qualified_name=connection_qualified_name,
+            connector_name=connector_name,
+            qlik_column_name=name,
+            qlik_parent_qualified_name=parent_qualified_name,
+        )
+        if parent_kind == "sheet":
+            asset.qlik_sheet = RelatedQlikSheet(qualified_name=parent_qualified_name)
+        elif parent_kind == "chart":
+            asset.qlik_chart = RelatedQlikChart(qualified_name=parent_qualified_name)
+        else:
+            asset.qlik_dataset = RelatedQlikDataset(
+                qualified_name=parent_qualified_name
+            )
+        if space_qualified_name:
+            space_parts = space_qualified_name.rsplit("/", 2)
+            if len(space_parts) != 3 or space_parts[1] not in ("space", "stream"):
+                raise ValueError(
+                    "space_qualified_name must end in /space/<id> or /stream/<id>"
+                )
+            asset.qlik_space_id = space_parts[2]
+            asset.qlik_space_qualified_name = space_qualified_name
+        return asset
+
+    @classmethod
+    def create(cls, **kwargs) -> "QlikColumn":
+        return cls.creator(**kwargs)
+
+    @classmethod
+    def create_for_modification(cls, **kwargs) -> "QlikColumn":
+        return cls.updater(**kwargs)

pyatlan_v9/model/assets/_overlays/qlik_dataset.py

@@ -0,0 +1,52 @@
+# INTERNAL_IMPORT: from pyatlan.utils import init_guid, validate_required_fields
+
+    @classmethod
+    @init_guid
+    def creator(
+        cls,
+        *,
+        name: str,
+        qlik_id: str,
+        space_qualified_name: str,
+    ) -> "QlikDataset":
+        """
+        Build a Qlik dataset; qualifiedName is {connection}/dataset/{qlik_id}.
+
+        space_qualified_name is the qualifiedName of the QlikSpace
+        ({connection}/space/{id}) or QlikStream ({connection}/stream/{id})
+        the dataset lives in. Its segment decides which type the qlik_space
+        relationship points at.
+        """
+        validate_required_fields(
+            ["name", "qlik_id", "space_qualified_name"],
+            [name, qlik_id, space_qualified_name],
+        )
+        space_parts = space_qualified_name.rsplit("/", 2)
+        if len(space_parts) != 3 or space_parts[1] not in ("space", "stream"):
+            raise ValueError(
+                "space_qualified_name must end in /space/<id> or /stream/<id>"
+            )
+        connection_qualified_name, space_kind, space_id = space_parts
+        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/{qlik_id}",
+            connection_qualified_name=connection_qualified_name,
+            connector_name=connector_name,
+            qlik_id=qlik_id,
+            qlik_space_id=space_id,
+            qlik_space_qualified_name=space_qualified_name,
+            qlik_space=RelatedQlikSpace(
+                type_name="QlikStream" if space_kind == "stream" else "QlikSpace",
+                qualified_name=space_qualified_name,
+            ),
+        )
+
+    @classmethod
+    def create(cls, **kwargs) -> "QlikDataset":
+        return cls.creator(**kwargs)
+
+    @classmethod
+    def create_for_modification(cls, **kwargs) -> "QlikDataset":
+        return cls.updater(**kwargs)

pyatlan_v9/model/assets/_overlays/qlik_sheet.py

@@ -0,0 +1,55 @@
+# INTERNAL_IMPORT: from pyatlan.utils import init_guid, validate_required_fields
+
+    @classmethod
+    @init_guid
+    def creator(
+        cls,
+        *,
+        name: str,
+        qlik_id: str,
+        app_qualified_name: str,
+        space_qualified_name: Union[str, None] = None,
+    ) -> "QlikSheet":
+        """
+        Build a Qlik sheet; qualifiedName is {app_qualified_name}/sheet/{qlik_id}.
+
+        space_qualified_name is optional and only fills the denormalised
+        qlik_space_id / qlik_space_qualified_name of the app's space or stream.
+        """
+        validate_required_fields(
+            ["name", "qlik_id", "app_qualified_name"],
+            [name, qlik_id, app_qualified_name],
+        )
+        app_parts = app_qualified_name.rsplit("/", 2)
+        if len(app_parts) != 3 or app_parts[1] != "app":
+            raise ValueError("app_qualified_name must end in /app/<id>")
+        connection_qualified_name, _, app_id = app_parts
+        fields = connection_qualified_name.split("/")
+        connector_name = fields[1] if len(fields) > 1 else None
+        asset = cls(
+            name=name,
+            qualified_name=f"{app_qualified_name}/sheet/{qlik_id}",
+            connection_qualified_name=connection_qualified_name,
+            connector_name=connector_name,
+            qlik_id=qlik_id,
+            qlik_app_id=app_id,
+            qlik_app_qualified_name=app_qualified_name,
+            qlik_app=RelatedQlikApp(qualified_name=app_qualified_name),
+        )
+        if space_qualified_name:
+            space_parts = space_qualified_name.rsplit("/", 2)
+            if len(space_parts) != 3 or space_parts[1] not in ("space", "stream"):
+                raise ValueError(
+                    "space_qualified_name must end in /space/<id> or /stream/<id>"
+                )
+            asset.qlik_space_id = space_parts[2]
+            asset.qlik_space_qualified_name = space_qualified_name
+        return asset
+
+    @classmethod
+    def create(cls, **kwargs) -> "QlikSheet":
+        return cls.creator(**kwargs)
+
+    @classmethod
+    def create_for_modification(cls, **kwargs) -> "QlikSheet":
+        return cls.updater(**kwargs)

pyatlan_v9/model/assets/_overlays/qlik_space.py

@@ -0,0 +1,33 @@
+# INTERNAL_IMPORT: from pyatlan.utils import init_guid, validate_required_fields
+
+    @classmethod
+    @init_guid
+    def creator(
+        cls,
+        *,
+        name: str,
+        connection_qualified_name: str,
+        qlik_id: str,
+    ) -> "QlikSpace":
+        """Build a Qlik space; qualifiedName is {connection}/space/{qlik_id}."""
+        validate_required_fields(
+            ["name", "connection_qualified_name", "qlik_id"],
+            [name, connection_qualified_name, qlik_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}/space/{qlik_id}",
+            connection_qualified_name=connection_qualified_name,
+            connector_name=connector_name,
+            qlik_id=qlik_id,
+        )
+
+    @classmethod
+    def create(cls, **kwargs) -> "QlikSpace":
+        return cls.creator(**kwargs)
+
+    @classmethod
+    def create_for_modification(cls, **kwargs) -> "QlikSpace":
+        return cls.updater(**kwargs)

pyatlan_v9/model/assets/_overlays/qlik_stream.py

@@ -0,0 +1,33 @@
+# INTERNAL_IMPORT: from pyatlan.utils import init_guid, validate_required_fields
+
+    @classmethod
+    @init_guid
+    def creator(
+        cls,
+        *,
+        name: str,
+        connection_qualified_name: str,
+        qlik_id: str,
+    ) -> "QlikStream":
+        """Build a Qlik stream; qualifiedName is {connection}/stream/{qlik_id}."""
+        validate_required_fields(
+            ["name", "connection_qualified_name", "qlik_id"],
+            [name, connection_qualified_name, qlik_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}/stream/{qlik_id}",
+            connection_qualified_name=connection_qualified_name,
+            connector_name=connector_name,
+            qlik_id=qlik_id,
+        )
+
+    @classmethod
+    def create(cls, **kwargs) -> "QlikStream":
+        return cls.creator(**kwargs)
+
+    @classmethod
+    def create_for_modification(cls, **kwargs) -> "QlikStream":
+        return cls.updater(**kwargs)