Merge pull request #687 from atlanhq/APP-7850

APP-7850 : Add CRUD on DQ rules (v1)

Author: Aryamanz29

docs/asset/alpha_dqrule.rst

@@ -0,0 +1,10 @@
+.. _alpha_dqrule:
+
+alpha_DQRule
+============
+
+.. module:: pyatlan.model.assets
+    :no-index:
+
+.. autoclass:: alpha_DQRule
+    :members:

docs/asset/alpha_dqruletemplate.rst

@@ -0,0 +1,10 @@
+.. _alpha_dqruletemplate:
+
+alpha_DQRuleTemplate
+====================
+
+.. module:: pyatlan.model.assets
+    :no-index:
+
+.. autoclass:: alpha_DQRuleTemplate
+    :members:

docs/assets.rst

@@ -366,3 +366,5 @@ You can interact with all of the following different kinds of assets:
     asset/view
     asset/workflow
     asset/workflowrun
+    asset/alpha_dqrule
+    asset/alpha_dqruletemplate

pyatlan/cache/dq_template_config_cache.py

@@ -0,0 +1,78 @@
+# SPDX-License-Identifier: Apache-2.0
+# Copyright 2025 Atlan Pte. Ltd.
+from __future__ import annotations
+
+import threading
+from typing import TYPE_CHECKING, Dict, Optional
+
+from pyatlan.model.assets import Asset
+from pyatlan.model.fluent_search import FluentSearch
+
+if TYPE_CHECKING:
+    from pyatlan.client.atlan import AtlanClient
+
+
+class DQTemplateConfigCache:
+    """
+    Lazily-loaded cache for DQ rule template configurations to avoid multiple API calls.
+    """
+
+    def __init__(self, client: AtlanClient):
+        self.client: AtlanClient = client
+        self._cache: Dict[str, Dict] = {}
+        self._lock: threading.Lock = threading.Lock()
+        self._initialized: bool = False
+
+    def get_template_config(self, rule_type: str) -> Optional[Dict]:
+        """
+        Get template configuration for a specific rule type.
+
+        :param rule_type: The display name of the rule type
+        :returns: Template configuration dict or None if not found
+        """
+        if not self._initialized:
+            self._refresh_cache()
+
+        return self._cache.get(rule_type)
+
+    def _refresh_cache(self) -> None:
+        """Refresh the cache by fetching all template configurations."""
+        with self._lock:
+            if self._initialized:
+                return
+
+            try:
+                from pyatlan.model.assets.core.alpha__d_q_rule_template import (
+                    alpha_DQRuleTemplate,
+                )
+
+                request = (
+                    FluentSearch()
+                    .where(Asset.TYPE_NAME.eq(alpha_DQRuleTemplate.__name__))
+                    .include_on_results(alpha_DQRuleTemplate.NAME)
+                    .include_on_results(alpha_DQRuleTemplate.QUALIFIED_NAME)
+                    .include_on_results(alpha_DQRuleTemplate.DISPLAY_NAME)
+                    .include_on_results(
+                        alpha_DQRuleTemplate.ALPHADQ_RULE_TEMPLATE_DIMENSION
+                    )
+                    .include_on_results(
+                        alpha_DQRuleTemplate.ALPHADQ_RULE_TEMPLATE_CONFIG
+                    )
+                ).to_request()
+
+                results = self.client.asset.search(request)
+                for result in results:
+                    template_config = {
+                        "name": result.name,
+                        "qualified_name": result.qualified_name,
+                        "display_name": result.display_name,
+                        "dimension": result.alpha_dq_rule_template_dimension,  # type: ignore
+                        "config": result.alpha_dq_rule_template_config,  # type: ignore
+                    }
+                    self._cache[result.display_name] = template_config  # type: ignore
+
+                self._initialized = True
+            except Exception:
+                # If cache refresh fails, mark as initialized to prevent infinite retries
+                self._initialized = True
+                raise

pyatlan/client/atlan.py

@@ -33,6 +33,7 @@
 from pyatlan.cache.atlan_tag_cache import AtlanTagCache
 from pyatlan.cache.connection_cache import ConnectionCache
 from pyatlan.cache.custom_metadata_cache import CustomMetadataCache
+from pyatlan.cache.dq_template_config_cache import DQTemplateConfigCache
 from pyatlan.cache.enum_cache import EnumCache
 from pyatlan.cache.group_cache import GroupCache
 from pyatlan.cache.role_cache import RoleCache
@@ -174,6 +175,9 @@ class AtlanClient(BaseSettings):
     _custom_metadata_cache: Optional[CustomMetadataCache] = PrivateAttr(default=None)
     _connection_cache: Optional[ConnectionCache] = PrivateAttr(default=None)
     _source_tag_cache: Optional[SourceTagCache] = PrivateAttr(default=None)
+    _dq_template_config_cache: Optional[DQTemplateConfigCache] = PrivateAttr(
+        default=None
+    )
 
     class Config:
         env_prefix = "atlan_"
@@ -347,6 +351,12 @@ def source_tag_cache(self) -> SourceTagCache:
             self._source_tag_cache = SourceTagCache(client=self)
         return self._source_tag_cache
 
+    @property
+    def dq_template_config_cache(self) -> DQTemplateConfigCache:
+        if self._dq_template_config_cache is None:
+            self._dq_template_config_cache = DQTemplateConfigCache(client=self)
+        return self._dq_template_config_cache
+
     @classmethod
     def from_token_guid(cls, guid: str) -> AtlanClient:
         """

pyatlan/errors.py

@@ -932,6 +932,13 @@ class ErrorCode(Enum):
         "Verify the API token provided is a valid username for that token.",
         NotFoundError,
     )
+    DQ_RULE_NOT_FOUND = (
+        404,
+        "ATLAN-PYTHON-404-029",
+        "DQ rule with type {0} was not found.",
+        "Verify you have provided a valid DQ rule type.",
+        NotFoundError,
+    )
     CONFLICT_PASSTHROUGH = (
         409,
         "ATLAN-PYTHON-409-000",

pyatlan/generator/class_generator.py

@@ -227,6 +227,7 @@ class AssetInfo:
         "DataProduct",
         "Referenceable",
         "Purpose",
+        "alpha_DQRule",
     }
     _CORE_ASSETS = {
         "Referenceable",