FND-3884: add pyatlan_v9 creator for QuickSightAnalysisField

pyatlan_v9/model/assets/_overlays/quick_sight_analysis_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,
+        quick_sight_id: str,
+        quick_sight_analysis_visual_qualified_name: str,
+        quick_sight_analysis_qualified_name: Union[str, None] = None,
+        quick_sight_analysis_field_expression: Union[str, None] = None,
+        quick_sight_analysis_field_data_set_identifier: Union[str, None] = None,
+        connection_qualified_name: Union[str, None] = None,
+    ) -> "QuickSightAnalysisField":
+        """Create a new QuickSightAnalysisField asset."""
+        validate_required_fields(
+            ["name", "quick_sight_id", "quick_sight_analysis_visual_qualified_name"],
+            [name, quick_sight_id, quick_sight_analysis_visual_qualified_name],
+        )
+        # A visual qualifiedName is {connection_qn}/{analysis_id}/{sheet_id}/{visual_id}.
+        parts = quick_sight_analysis_visual_qualified_name.split("/")
+        if len(parts) < 6:
+            raise ValueError("quick_sight_analysis_visual_qualified_name is invalid")
+        if not quick_sight_analysis_qualified_name:
+            quick_sight_analysis_qualified_name = "/".join(parts[:-2])
+        if not connection_qualified_name:
+            connection_qualified_name = "/".join(parts[:3])
+        fields = connection_qualified_name.split("/")
+        connector_name = fields[1] if len(fields) > 1 else None
+        return cls(
+            name=name,
+            quick_sight_id=quick_sight_id,
+            qualified_name=f"{quick_sight_analysis_visual_qualified_name}/{quick_sight_id}",
+            quick_sight_analysis_visual_qualified_name=quick_sight_analysis_visual_qualified_name,
+            quick_sight_analysis_qualified_name=quick_sight_analysis_qualified_name,
+            quick_sight_analysis_field_expression=quick_sight_analysis_field_expression
+            if quick_sight_analysis_field_expression is not None
+            else UNSET,
+            quick_sight_analysis_field_data_set_identifier=quick_sight_analysis_field_data_set_identifier
+            if quick_sight_analysis_field_data_set_identifier is not None
+            else UNSET,
+            connection_qualified_name=connection_qualified_name,
+            connector_name=connector_name,
+            quick_sight_analysis_visual=RelatedQuickSightAnalysisVisual(
+                unique_attributes={
+                    "qualifiedName": quick_sight_analysis_visual_qualified_name
+                }
+            ),
+        )
+
+    @classmethod
+    def updater(cls, *, qualified_name: str, name: str) -> "QuickSightAnalysisField":
+        """Create a QuickSightAnalysisField instance for update operations."""
+        validate_required_fields(["qualified_name", "name"], [qualified_name, name])
+        return cls(qualified_name=qualified_name, name=name)
+
+    def trim_to_required(self) -> "QuickSightAnalysisField":
+        """Return only fields required for update operations."""
+        return QuickSightAnalysisField.updater(
+            qualified_name=self.qualified_name, name=self.name
+        )

pyatlan_v9/model/assets/quick_sight_analysis_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
@@ -338,6 +339,67 @@ def relate(self) -> "RelatedQuickSightAnalysisField":
             return RelatedQuickSightAnalysisField(guid=self.guid)
         return RelatedQuickSightAnalysisField(qualified_name=self.qualified_name)
 
+    @classmethod
+    @init_guid
+    def creator(
+        cls,
+        *,
+        name: str,
+        quick_sight_id: str,
+        quick_sight_analysis_visual_qualified_name: str,
+        quick_sight_analysis_qualified_name: Union[str, None] = None,
+        quick_sight_analysis_field_expression: Union[str, None] = None,
+        quick_sight_analysis_field_data_set_identifier: Union[str, None] = None,
+        connection_qualified_name: Union[str, None] = None,
+    ) -> "QuickSightAnalysisField":
+        """Create a new QuickSightAnalysisField asset."""
+        validate_required_fields(
+            ["name", "quick_sight_id", "quick_sight_analysis_visual_qualified_name"],
+            [name, quick_sight_id, quick_sight_analysis_visual_qualified_name],
+        )
+        # A visual qualifiedName is {connection_qn}/{analysis_id}/{sheet_id}/{visual_id}.
+        parts = quick_sight_analysis_visual_qualified_name.split("/")
+        if len(parts) < 6:
+            raise ValueError("quick_sight_analysis_visual_qualified_name is invalid")
+        if not quick_sight_analysis_qualified_name:
+            quick_sight_analysis_qualified_name = "/".join(parts[:-2])
+        if not connection_qualified_name:
+            connection_qualified_name = "/".join(parts[:3])
+        fields = connection_qualified_name.split("/")
+        connector_name = fields[1] if len(fields) > 1 else None
+        return cls(
+            name=name,
+            quick_sight_id=quick_sight_id,
+            qualified_name=f"{quick_sight_analysis_visual_qualified_name}/{quick_sight_id}",
+            quick_sight_analysis_visual_qualified_name=quick_sight_analysis_visual_qualified_name,
+            quick_sight_analysis_qualified_name=quick_sight_analysis_qualified_name,
+            quick_sight_analysis_field_expression=quick_sight_analysis_field_expression
+            if quick_sight_analysis_field_expression is not None
+            else UNSET,
+            quick_sight_analysis_field_data_set_identifier=quick_sight_analysis_field_data_set_identifier
+            if quick_sight_analysis_field_data_set_identifier is not None
+            else UNSET,
+            connection_qualified_name=connection_qualified_name,
+            connector_name=connector_name,
+            quick_sight_analysis_visual=RelatedQuickSightAnalysisVisual(
+                unique_attributes={
+                    "qualifiedName": quick_sight_analysis_visual_qualified_name
+                }
+            ),
+        )
+
+    @classmethod
+    def updater(cls, *, qualified_name: str, name: str) -> "QuickSightAnalysisField":
+        """Create a QuickSightAnalysisField instance for update operations."""
+        validate_required_fields(["qualified_name", "name"], [qualified_name, name])
+        return cls(qualified_name=qualified_name, name=name)
+
+    def trim_to_required(self) -> "QuickSightAnalysisField":
+        """Return only fields required for update operations."""
+        return QuickSightAnalysisField.updater(
+            qualified_name=self.qualified_name, name=self.name
+        )
+
     # =========================================================================
     # Optimized Serialization Methods (override Asset base class)
     # =========================================================================

tests_v9/unit/model/quick_sight_analysis_field_test.py

@@ -0,0 +1,163 @@
+# SPDX-License-Identifier: Apache-2.0
+# Copyright 2026 Atlan Pte. Ltd.
+
+"""Unit tests for QuickSightAnalysisField model in pyatlan_v9.
+
+The parity test uses the exact qualifiedName atlan-quicksight-app emits today
+(``{visualQN}/{fieldId}``, FND-3838 census), so a creator change that would
+re-key existing analysis fields fails here.
+"""
+
+import pytest
+
+from pyatlan_v9.model import QuickSightAnalysisField
+from tests_v9.unit.model.constants import (
+    QUICK_SIGHT_ANALYSIS_VISUAL_QUALIFIED_NAME,
+    QUICK_SIGHT_CONNECTION_QUALIFIED_NAME,
+    QUICK_SIGHT_CONNECTOR_TYPE,
+    QUICK_SIGHT_NAME,
+    QUICK_SIGHT_QUALIFIED_NAME,
+)
+
+FIELD_ID = "dset1-total"
+FIELD_QUALIFIED_NAME = f"{QUICK_SIGHT_ANALYSIS_VISUAL_QUALIFIED_NAME}/{FIELD_ID}"
+
+
+def test_creator_matches_app_qualified_name():
+    """Mirror atlan-quicksight-app's analysis_field mapper inputs and output."""
+    sut = QuickSightAnalysisField.creator(
+        name="total",
+        quick_sight_id="dset1-total",
+        quick_sight_analysis_visual_qualified_name="default/quicksight/123/an1/sh1/vis1",
+        quick_sight_analysis_qualified_name="default/quicksight/123/an1",
+        quick_sight_analysis_field_expression="abs({amount}+5000)",
+        quick_sight_analysis_field_data_set_identifier="ds",
+        connection_qualified_name="default/quicksight/123",
+    )
+
+    assert sut.qualified_name == "default/quicksight/123/an1/sh1/vis1/dset1-total"
+    assert sut.quick_sight_analysis_field_expression == "abs({amount}+5000)"
+    assert sut.quick_sight_analysis_field_data_set_identifier == "ds"
+    sut.validate(for_creation=True)
+
+
+@pytest.mark.parametrize(
+    "name, quick_sight_id, visual_qualified_name, message",
+    [
+        (
+            None,
+            FIELD_ID,
+            QUICK_SIGHT_ANALYSIS_VISUAL_QUALIFIED_NAME,
+            "name is required",
+        ),
+        (
+            " ",
+            FIELD_ID,
+            QUICK_SIGHT_ANALYSIS_VISUAL_QUALIFIED_NAME,
+            "name cannot be blank",
+        ),
+        (
+            QUICK_SIGHT_NAME,
+            None,
+            QUICK_SIGHT_ANALYSIS_VISUAL_QUALIFIED_NAME,
+            "quick_sight_id is required",
+        ),
+        (
+            QUICK_SIGHT_NAME,
+            FIELD_ID,
+            None,
+            "quick_sight_analysis_visual_qualified_name is required",
+        ),
+        (
+            QUICK_SIGHT_NAME,
+            FIELD_ID,
+            QUICK_SIGHT_QUALIFIED_NAME,
+            "quick_sight_analysis_visual_qualified_name is invalid",
+        ),
+    ],
+)
+def test_creator_with_missing_parameters_raise_value_error(
+    name: str, quick_sight_id: str, visual_qualified_name: str, message: str
+):
+    """Test creator validates required parameters."""
+    with pytest.raises(ValueError, match=message):
+        QuickSightAnalysisField.creator(
+            name=name,
+            quick_sight_id=quick_sight_id,
+            quick_sight_analysis_visual_qualified_name=visual_qualified_name,
+        )
+
+
+def test_creator():
+    """Test creator derives the analysis and connection from the visual."""
+    sut = QuickSightAnalysisField.creator(
+        name=QUICK_SIGHT_NAME,
+        quick_sight_id=FIELD_ID,
+        quick_sight_analysis_visual_qualified_name=QUICK_SIGHT_ANALYSIS_VISUAL_QUALIFIED_NAME,
+    )
+
+    assert sut.name == QUICK_SIGHT_NAME
+    assert sut.quick_sight_id == FIELD_ID
+    assert sut.qualified_name == FIELD_QUALIFIED_NAME
+    assert (
+        sut.quick_sight_analysis_visual_qualified_name
+        == QUICK_SIGHT_ANALYSIS_VISUAL_QUALIFIED_NAME
+    )
+    assert sut.quick_sight_analysis_qualified_name == QUICK_SIGHT_QUALIFIED_NAME
+    assert sut.connection_qualified_name == QUICK_SIGHT_CONNECTION_QUALIFIED_NAME
+    assert sut.connector_name == QUICK_SIGHT_CONNECTOR_TYPE
+    assert sut.quick_sight_analysis_visual.unique_attributes == {
+        "qualifiedName": QUICK_SIGHT_ANALYSIS_VISUAL_QUALIFIED_NAME
+    }
+    sut.validate(for_creation=True)
+
+
+def test_overload_creator():
+    """Test creator accepts explicit analysis and connection qualified names."""
+    sut = QuickSightAnalysisField.creator(
+        name=QUICK_SIGHT_NAME,
+        quick_sight_id=FIELD_ID,
+        quick_sight_analysis_visual_qualified_name=QUICK_SIGHT_ANALYSIS_VISUAL_QUALIFIED_NAME,
+        quick_sight_analysis_qualified_name=QUICK_SIGHT_QUALIFIED_NAME,
+        connection_qualified_name=QUICK_SIGHT_CONNECTION_QUALIFIED_NAME,
+    )
+
+    assert sut.qualified_name == FIELD_QUALIFIED_NAME
+    assert sut.quick_sight_analysis_qualified_name == QUICK_SIGHT_QUALIFIED_NAME
+    assert sut.connection_qualified_name == QUICK_SIGHT_CONNECTION_QUALIFIED_NAME
+    assert sut.connector_name == QUICK_SIGHT_CONNECTOR_TYPE
+
+
+@pytest.mark.parametrize(
+    "qualified_name, name, message",
+    [
+        (None, QUICK_SIGHT_NAME, "qualified_name is required"),
+        (FIELD_QUALIFIED_NAME, None, "name is required"),
+    ],
+)
+def test_updater_with_invalid_parameter_raises_value_error(
+    qualified_name: str, name: str, message: str
+):
+    """Test updater validates required parameters."""
+    with pytest.raises(ValueError, match=message):
+        QuickSightAnalysisField.updater(qualified_name=qualified_name, name=name)
+
+
+def test_updater():
+    """Test updater creates a QuickSightAnalysisField for modification."""
+    sut = QuickSightAnalysisField.updater(
+        qualified_name=FIELD_QUALIFIED_NAME, name=QUICK_SIGHT_NAME
+    )
+
+    assert sut.qualified_name == FIELD_QUALIFIED_NAME
+    assert sut.name == QUICK_SIGHT_NAME
+
+
+def test_trim_to_required():
+    """Test trim_to_required keeps only updater-required fields."""
+    sut = QuickSightAnalysisField.updater(
+        qualified_name=FIELD_QUALIFIED_NAME, name=QUICK_SIGHT_NAME
+    ).trim_to_required()
+
+    assert sut.qualified_name == FIELD_QUALIFIED_NAME
+    assert sut.name == QUICK_SIGHT_NAME