feat: add row scope filter column check for incremental rules

Author: UdaySagar-Git

pyatlan/errors.py

@@ -661,6 +661,13 @@ class ErrorCode(Enum):
         "Ensure your rule conditions are valid and match the expected format.",
         InvalidRequestError,
     )
+    DQ_ROW_SCOPE_FILTER_COLUMN_MISSING = (
+        400,
+        "ATLAN-PYTHON-400-075",
+        "Row scope filter column not configured for asset '{0}'.",
+        "please configure the row scope filter column first.",
+        InvalidRequestError,
+    )
     AUTHENTICATION_PASSTHROUGH = (
         401,
         "ATLAN-PYTHON-401-000",

pyatlan/model/assets/core/alpha__d_q_rule.py

@@ -41,7 +41,7 @@
 
 if TYPE_CHECKING:
     from pyatlan.client.atlan import AtlanClient
-    from pyatlan.model.assets import Column
+    from pyatlan.model.assets import Asset, Column
 
 
 class alpha_DQRule(DataQuality):
@@ -185,13 +185,30 @@ def column_level_rule_creator(
             ],
         )
         template_config = client.dq_template_config_cache.get_template_config(rule_type)
+
+        asset_for_validation = asset
+        if row_scope_filtering_enabled and asset.qualified_name:
+            from pyatlan.model.fluent_search import FluentSearch
+
+            search_request = (
+                FluentSearch()
+                .where(Asset.QUALIFIED_NAME.eq(asset.qualified_name))
+                .include_on_results(
+                    Asset.ALPHAASSET_DQ_ROW_SCOPE_FILTER_COLUMN_QUALIFIED_NAME
+                )
+            ).to_request()
+            results = client.asset.search(search_request)
+            if results.count == 1:
+                asset_for_validation = results.current_page()[0]
+
         threshold_compare_operator = (
             alpha_DQRule.Attributes._validate_template_features(
                 rule_type,
                 rule_conditions,
                 row_scope_filtering_enabled,
                 template_config,
                 threshold_compare_operator,
+                asset_for_validation,
             )
         )
 
@@ -311,6 +328,7 @@ def updater(
                 row_scope_filtering_enabled,
                 template_config,
                 threshold_compare_operator or retrieved_threshold_compare_operator,
+                retrieved_asset,
             )
         )
 
@@ -1093,6 +1111,7 @@ def _validate_template_features(
             threshold_compare_operator: Optional[
                 alpha_DQRuleThresholdCompareOperator
             ] = None,
+            asset: Optional[Asset] = None,
         ) -> alpha_DQRuleThresholdCompareOperator:
             if not template_config or not template_config.get("config"):
                 return
@@ -1116,6 +1135,15 @@ def _validate_template_features(
                         rule_type, "row scope filtering"
                     )
 
+                if asset and not getattr(
+                    asset,
+                    "alpha_asset_d_q_row_scope_filter_column_qualified_name",
+                    None,
+                ):
+                    raise ErrorCode.DQ_ROW_SCOPE_FILTER_COLUMN_MISSING.exception_with_parameters(
+                        getattr(asset, "qualified_name", "unknown")
+                    )
+
             if rule_conditions:
                 allowed_rule_conditions = (
                     alpha_DQRule.Attributes._get_template_config_value(

pyatlan/model/assets/core/asset.py

@@ -1143,6 +1143,15 @@ def __setattr__(self, name, value):
     """
     Internal Popularity score for this asset.
     """
+    ALPHAASSET_DQ_ROW_SCOPE_FILTER_COLUMN_QUALIFIED_NAME: ClassVar[KeywordField] = (
+        KeywordField(
+            "alpha_assetDQRowScopeFilterColumnQualifiedName",
+            "alpha_assetDQRowScopeFilterColumnQualifiedName",
+        )
+    )
+    """
+    Qualified name of the column to be used for row scope filter
+    """
     ALPHAASSET_DQ_SCHEDULE_TYPE: ClassVar[KeywordField] = KeywordField(
         "alpha_assetDQScheduleType", "alpha_assetDQScheduleType"
     )
@@ -1510,6 +1519,7 @@ def __setattr__(self, name, value):
         "application_field_qualified_name",
         "asset_user_defined_type",
         "asset_internal_popularity_score",
+        "alpha_asset_d_q_row_scope_filter_column_qualified_name",
         "alpha_asset_d_q_schedule_type",
         "alpha_asset_d_q_schedule_crontab",
         "alpha_asset_d_q_schedule_time_zone",
@@ -3561,6 +3571,24 @@ def asset_internal_popularity_score(
             asset_internal_popularity_score
         )
 
+    @property
+    def alpha_asset_d_q_row_scope_filter_column_qualified_name(self) -> Optional[str]:
+        return (
+            None
+            if self.attributes is None
+            else self.attributes.alpha_asset_d_q_row_scope_filter_column_qualified_name
+        )
+
+    @alpha_asset_d_q_row_scope_filter_column_qualified_name.setter
+    def alpha_asset_d_q_row_scope_filter_column_qualified_name(
+        self, alpha_asset_d_q_row_scope_filter_column_qualified_name: Optional[str]
+    ):
+        if self.attributes is None:
+            self.attributes = self.Attributes()
+        self.attributes.alpha_asset_d_q_row_scope_filter_column_qualified_name = (
+            alpha_asset_d_q_row_scope_filter_column_qualified_name
+        )
+
     @property
     def alpha_asset_d_q_schedule_type(self) -> Optional[alpha_DQScheduleType]:
         return (
@@ -4451,6 +4479,9 @@ class Attributes(Referenceable.Attributes):
         asset_internal_popularity_score: Optional[float] = Field(
             default=None, description=""
         )
+        alpha_asset_d_q_row_scope_filter_column_qualified_name: Optional[str] = Field(
+            default=None, description=""
+        )
         alpha_asset_d_q_schedule_type: Optional[alpha_DQScheduleType] = Field(
             default=None, description=""
         )