diff --git a/pyatlan_v9/model/assets/_overlays/quick_sight_analysis_field.py b/pyatlan_v9/model/assets/_overlays/quick_sight_analysis_field.py new file mode 100644 index 000000000..b1c2e189e --- /dev/null +++ b/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 + ) diff --git a/pyatlan_v9/model/assets/quick_sight_analysis_field.py b/pyatlan_v9/model/assets/quick_sight_analysis_field.py index 76a25d02d..7c851f823 100644 --- a/pyatlan_v9/model/assets/quick_sight_analysis_field.py +++ b/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) # ========================================================================= diff --git a/tests_v9/unit/model/quick_sight_analysis_field_test.py b/tests_v9/unit/model/quick_sight_analysis_field_test.py new file mode 100644 index 000000000..002495797 --- /dev/null +++ b/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