Source code for pyatlan.model.assets.core.alpha__d_q_rule

# SPDX-License-Identifier: Apache-2.0
# Copyright 2025 Atlan Pte. Ltd.


from __future__ import annotations

import json
import time
import uuid
from datetime import datetime
from typing import TYPE_CHECKING, ClassVar, List, Optional, Set

from pydantic.v1 import Field, validator

from pyatlan.errors import ErrorCode
from pyatlan.model.enums import (
    alpha_DQDimension,
    alpha_DQResult,
    alpha_DQRuleAlertPriority,
    alpha_DQRuleStatus,
    alpha_DQRuleThresholdCompareOperator,
    alpha_DQRuleThresholdUnit,
    alpha_DQSourceSyncStatus,
)
from pyatlan.model.fields.atlan_fields import (
    KeywordField,
    KeywordTextField,
    NumericField,
    RelationField,
    TextField,
)
from pyatlan.model.structs import (
    alpha_DQRuleConfigArguments,
    alpha_DQRuleThresholdObject,
)
from pyatlan.utils import init_guid, validate_required_fields

from .asset import SelfAsset
from .data_quality import DataQuality

if TYPE_CHECKING:
    from pyatlan.client.atlan import AtlanClient
    from pyatlan.model.assets import Column


[docs] class alpha_DQRule(DataQuality): """Description""" @classmethod @init_guid def custom_sql_creator( cls, *, client: AtlanClient, rule_name: str, asset: Asset, custom_sql: str, threshold_compare_operator: alpha_DQRuleThresholdCompareOperator, threshold_value: int, alert_priority: alpha_DQRuleAlertPriority, dimension: alpha_DQDimension, description: Optional[str] = None, ) -> alpha_DQRule: validate_required_fields( [ "client", "rule_name", "asset", "threshold_compare_operator", "threshold_value", "alert_priority", "dimension", "custom_sql", ], [ client, rule_name, asset, threshold_compare_operator, threshold_value, alert_priority, dimension, custom_sql, ], ) attributes = alpha_DQRule.Attributes.creator( client=client, rule_name=rule_name, rule_type="Custom SQL", asset=asset, threshold_compare_operator=threshold_compare_operator, threshold_value=threshold_value, alert_priority=alert_priority, dimension=dimension, custom_sql=custom_sql, description=description, column=None, threshold_unit=None, ) return cls(attributes=attributes) @classmethod @init_guid def table_level_rule_creator( cls, *, client: AtlanClient, rule_type: str, asset: Asset, threshold_compare_operator: alpha_DQRuleThresholdCompareOperator, threshold_value: int, alert_priority: alpha_DQRuleAlertPriority, ) -> alpha_DQRule: validate_required_fields( [ "client", "rule_type", "asset", "threshold_compare_operator", "threshold_value", "alert_priority", ], [ client, rule_type, asset, threshold_compare_operator, threshold_value, alert_priority, ], ) attributes = alpha_DQRule.Attributes.creator( client=client, rule_type=rule_type, asset=asset, threshold_compare_operator=threshold_compare_operator, threshold_value=threshold_value, alert_priority=alert_priority, rule_name=None, column=None, threshold_unit=None, dimension=None, custom_sql=None, description=None, ) return cls(attributes=attributes) @classmethod @init_guid def column_level_rule_creator( cls, *, client: AtlanClient, rule_type: str, asset: Asset, column: Asset, threshold_value: int, alert_priority: alpha_DQRuleAlertPriority, threshold_compare_operator: Optional[ alpha_DQRuleThresholdCompareOperator ] = alpha_DQRuleThresholdCompareOperator.LESS_THAN_EQUAL, threshold_unit: Optional[alpha_DQRuleThresholdUnit] = None, ) -> alpha_DQRule: validate_required_fields( [ "client", "rule_type", "asset", "column", "threshold_compare_operator", "threshold_value", "alert_priority", ], [ client, rule_type, asset, column, threshold_compare_operator, threshold_value, alert_priority, ], ) attributes = alpha_DQRule.Attributes.creator( client=client, rule_type=rule_type, asset=asset, column=column, threshold_compare_operator=threshold_compare_operator, threshold_value=threshold_value, alert_priority=alert_priority, threshold_unit=threshold_unit, rule_name=None, dimension=None, custom_sql=None, description=None, ) return cls(attributes=attributes) @classmethod @init_guid def updater( cls: type[SelfAsset], client: AtlanClient, qualified_name: str, threshold_compare_operator: Optional[ alpha_DQRuleThresholdCompareOperator ] = None, threshold_value: Optional[int] = None, alert_priority: Optional[alpha_DQRuleAlertPriority] = None, threshold_unit: Optional[alpha_DQRuleThresholdUnit] = None, dimension: Optional[alpha_DQDimension] = None, custom_sql: Optional[str] = None, rule_name: Optional[str] = None, description: Optional[str] = None, ) -> SelfAsset: from pyatlan.model.fluent_search import FluentSearch validate_required_fields( ["client", "qualified_name"], [client, qualified_name], ) request = ( FluentSearch() .where(alpha_DQRule.QUALIFIED_NAME.eq(qualified_name)) .include_on_results(alpha_DQRule.NAME) .include_on_results(alpha_DQRule.ALPHADQ_RULE_TEMPLATE_NAME) .include_on_results(alpha_DQRule.ALPHADQ_RULE_TEMPLATE) .include_on_results(alpha_DQRule.ALPHADQ_RULE_BASE_DATASET) .include_on_results(alpha_DQRule.ALPHADQ_RULE_BASE_COLUMN) .include_on_results(alpha_DQRule.ALPHADQ_RULE_ALERT_PRIORITY) .include_on_results(alpha_DQRule.DISPLAY_NAME) .include_on_results(alpha_DQRule.ALPHADQ_RULE_CUSTOM_SQL) .include_on_results(alpha_DQRule.USER_DESCRIPTION) .include_on_results(alpha_DQRule.ALPHADQ_RULE_DIMENSION) .include_on_results(alpha_DQRule.ALPHADQ_RULE_CONFIG_ARGUMENTS) .include_on_results(alpha_DQRule.ALPHADQ_RULE_SOURCE_SYNC_STATUS) .include_on_results(alpha_DQRule.ALPHADQ_RULE_STATUS) ).to_request() results = client.asset.search(request) if results.count != 1: raise ValueError( f"Expected exactly 1 asset for qualified_name: {qualified_name}, " f"but found: {results.count}" ) search_result = results.current_page()[0] retrieved_custom_sql = search_result.alpha_dq_rule_custom_s_q_l # type: ignore[attr-defined] retrieved_rule_name = search_result.display_name retrieved_dimension = search_result.alpha_dq_rule_dimension # type: ignore[attr-defined] retrieved_column = search_result.alpha_dq_rule_base_column # type: ignore[attr-defined] retrieved_alert_priority = search_result.alpha_dq_rule_alert_priority # type: ignore[attr-defined] retrieved_description = search_result.user_description retrieved_asset = search_result.alpha_dq_rule_base_dataset # type: ignore[attr-defined] retrieved_template_rule_name = search_result.alpha_dq_rule_template_name # type: ignore[attr-defined] retrieved_template = search_result.alpha_dq_rule_template # type: ignore[attr-defined] retrieved_threshold_compare_operator = ( search_result.alpha_dq_rule_config_arguments.alpha_dq_rule_threshold_object.alpha_dq_rule_threshold_compare_operator # type: ignore[attr-defined] if search_result.alpha_dq_rule_config_arguments is not None # type: ignore[attr-defined] and search_result.alpha_dq_rule_config_arguments.alpha_dq_rule_threshold_object # type: ignore[attr-defined] is not None else None ) retrieved_threshold_value = ( search_result.alpha_dq_rule_config_arguments.alpha_dq_rule_threshold_object.alpha_dq_rule_threshold_value # type: ignore[attr-defined] if search_result.alpha_dq_rule_config_arguments is not None # type: ignore[attr-defined] and search_result.alpha_dq_rule_config_arguments.alpha_dq_rule_threshold_object # type: ignore[attr-defined] is not None else None ) # type: ignore[attr-defined] retrieved_threshold_unit = ( search_result.alpha_dq_rule_config_arguments.alpha_dq_rule_threshold_object.alpha_dq_rule_threshold_unit # type: ignore[attr-defined] if search_result.alpha_dq_rule_config_arguments is not None # type: ignore[attr-defined] and search_result.alpha_dq_rule_config_arguments.alpha_dq_rule_threshold_object # type: ignore[attr-defined] is not None else None ) # type: ignore[attr-defined] config_arguments_raw = alpha_DQRule.Attributes._generate_config_arguments_raw( is_alert_enabled=True, custom_sql=custom_sql or retrieved_custom_sql, display_name=rule_name or retrieved_rule_name, dimension=dimension or retrieved_dimension, compare_operator=threshold_compare_operator or retrieved_threshold_compare_operator, threshold_value=threshold_value or retrieved_threshold_value, threshold_unit=threshold_unit or retrieved_threshold_unit, column=retrieved_column, dq_priority=alert_priority or retrieved_alert_priority, description=description or retrieved_description, ) attr_dq = cls.Attributes( name="", alpha_dq_rule_config_arguments=alpha_DQRuleConfigArguments( alpha_dq_rule_threshold_object=alpha_DQRuleThresholdObject( alpha_dq_rule_threshold_compare_operator=threshold_compare_operator or retrieved_threshold_compare_operator, alpha_dq_rule_threshold_value=threshold_value or retrieved_threshold_value, alpha_dq_rule_threshold_unit=threshold_unit or retrieved_threshold_unit, ), alpha_dq_rule_config_arguments_raw=config_arguments_raw, ), alpha_dq_rule_base_dataset_qualified_name=retrieved_asset.qualified_name, alpha_dq_rule_alert_priority=alert_priority or retrieved_alert_priority, alpha_dq_rule_base_dataset=retrieved_asset, qualified_name=qualified_name, alpha_dq_rule_dimension=dimension or retrieved_dimension, alpha_dq_rule_template_name=retrieved_template_rule_name, alpha_dq_rule_template=alpha_DQRuleTemplate.ref_by_qualified_name( qualified_name=retrieved_template.qualified_name ), ) if retrieved_column is not None: attr_dq.alpha_dq_rule_base_column_qualified_name = ( retrieved_column.qualified_name ) attr_dq.alpha_dq_rule_base_column = retrieved_column # type: ignore custom_sql = custom_sql or retrieved_custom_sql if custom_sql is not None: attr_dq.alpha_dq_rule_custom_s_q_l = custom_sql attr_dq.display_name = rule_name or retrieved_rule_name if description is not None: attr_dq.user_description = description or retrieved_description return cls(attributes=attr_dq) type_name: str = Field(default="alpha_DQRule", allow_mutation=False) @validator("type_name") def validate_type_name(cls, v): if v != "alpha_DQRule": raise ValueError("must be alpha_DQRule") return v def __setattr__(self, name, value): if name in alpha_DQRule._convenience_properties: return object.__setattr__(self, name, value) super().__setattr__(name, value) ALPHADQ_RULE_BASE_DATASET_QUALIFIED_NAME: ClassVar[KeywordTextField] = ( KeywordTextField( "alpha_dqRuleBaseDatasetQualifiedName", "alpha_dqRuleBaseDatasetQualifiedName", "alpha_dqRuleBaseDatasetQualifiedName.text", ) ) """ Base dataset qualified name that attached to this rule. """ ALPHADQ_RULE_BASE_COLUMN_QUALIFIED_NAME: ClassVar[KeywordTextField] = ( KeywordTextField( "alpha_dqRuleBaseColumnQualifiedName", "alpha_dqRuleBaseColumnQualifiedName", "alpha_dqRuleBaseColumnQualifiedName.text", ) ) """ Base column qualified name that attached to this rule. """ ALPHADQ_RULE_REFERENCE_DATASET_QUALIFIED_NAMES: ClassVar[KeywordTextField] = ( KeywordTextField( "alpha_dqRuleReferenceDatasetQualifiedNames", "alpha_dqRuleReferenceDatasetQualifiedNames", "alpha_dqRuleReferenceDatasetQualifiedNames.text", ) ) """ List of unique reference dataset's qualified names related to this rule. """ ALPHADQ_RULE_REFERENCE_COLUMN_QUALIFIED_NAMES: ClassVar[KeywordTextField] = ( KeywordTextField( "alpha_dqRuleReferenceColumnQualifiedNames", "alpha_dqRuleReferenceColumnQualifiedNames", "alpha_dqRuleReferenceColumnQualifiedNames.text", ) ) """ List of unique reference dataset's qualified names related to this rule. """ ALPHADQ_RULE_SOURCE_SYNC_STATUS: ClassVar[KeywordField] = KeywordField( "alpha_dqRuleSourceSyncStatus", "alpha_dqRuleSourceSyncStatus" ) """ Latest sync status of the rule to the source. """ ALPHADQ_RULE_SOURCE_SYNC_ERROR_CODE: ClassVar[TextField] = TextField( "alpha_dqRuleSourceSyncErrorCode", "alpha_dqRuleSourceSyncErrorCode" ) """ Error code in the case of state being "failure". """ ALPHADQ_RULE_SOURCE_SYNC_ERROR_MESSAGE: ClassVar[TextField] = TextField( "alpha_dqRuleSourceSyncErrorMessage", "alpha_dqRuleSourceSyncErrorMessage" ) """ Error message in the case of state being "error". """ ALPHADQ_RULE_SOURCE_SYNCED_AT: ClassVar[NumericField] = NumericField( "alpha_dqRuleSourceSyncedAt", "alpha_dqRuleSourceSyncedAt" ) """ Time (epoch) at which the rule synced to the source. """ ALPHADQ_RULE_LATEST_RESULT: ClassVar[KeywordField] = KeywordField( "alpha_dqRuleLatestResult", "alpha_dqRuleLatestResult" ) """ Latest result of the rule. """ ALPHADQ_RULE_LATEST_RESULT_COMPUTED_AT: ClassVar[NumericField] = NumericField( "alpha_dqRuleLatestResultComputedAt", "alpha_dqRuleLatestResultComputedAt" ) """ Time (epoch) at which the latest rule result was evaluated. """ ALPHADQ_RULE_LATEST_RESULT_FETCHED_AT: ClassVar[NumericField] = NumericField( "alpha_dqRuleLatestResultFetchedAt", "alpha_dqRuleLatestResultFetchedAt" ) """ Time (epoch) at which the latest rule result was fetched. """ ALPHADQ_RULE_LATEST_METRIC_VALUE: ClassVar[KeywordField] = KeywordField( "alpha_dqRuleLatestMetricValue", "alpha_dqRuleLatestMetricValue" ) """ Last Result metrics value of the rule. """ ALPHADQ_RULE_LATEST_METRIC_VALUE_COMPUTED_AT: ClassVar[NumericField] = NumericField( "alpha_dqRuleLatestMetricValueComputedAt", "alpha_dqRuleLatestMetricValueComputedAt", ) """ Time (epoch) at which the latest metric value was evaluated in the source. """ ALPHADQ_RULE_DIMENSION: ClassVar[KeywordField] = KeywordField( "alpha_dqRuleDimension", "alpha_dqRuleDimension" ) """ dimension of the data quality rule """ ALPHADQ_RULE_TEMPLATE_NAME: ClassVar[KeywordTextField] = KeywordTextField( "alpha_dqRuleTemplateName", "alpha_dqRuleTemplateName", "alpha_dqRuleTemplateName.text", ) """ Name of the rule template corresponding to the rule. """ ALPHADQ_RULE_STATUS: ClassVar[KeywordField] = KeywordField( "alpha_dqRuleStatus", "alpha_dqRuleStatus" ) """ Status of the rule. """ ALPHADQ_RULE_ALERT_PRIORITY: ClassVar[KeywordField] = KeywordField( "alpha_dqRuleAlertPriority", "alpha_dqRuleAlertPriority" ) """ Default priority level for alerts involving this rule. """ ALPHADQ_RULE_CONFIG_ARGUMENTS: ClassVar[KeywordField] = KeywordField( "alpha_dqRuleConfigArguments", "alpha_dqRuleConfigArguments" ) """ Json string of the rule config that contains the rule definitions. """ ALPHADQ_RULE_CUSTOM_SQL: ClassVar[TextField] = TextField( "alpha_dqRuleCustomSQL", "alpha_dqRuleCustomSQL" ) """ SQL code for custom SQL rules """ ALPHADQ_RULE_REFERENCE_COLUMNS: ClassVar[RelationField] = RelationField( "alpha_dqRuleReferenceColumns" ) """ TBC """ ALPHADQ_RULE_BASE_COLUMN: ClassVar[RelationField] = RelationField( "alpha_dqRuleBaseColumn" ) """ TBC """ ALPHADQ_RULE_BASE_DATASET: ClassVar[RelationField] = RelationField( "alpha_dqRuleBaseDataset" ) """ TBC """ ALPHADQ_RULE_TEMPLATE: ClassVar[RelationField] = RelationField( "alpha_dqRuleTemplate" ) """ TBC """ ALPHADQ_RULE_REFERENCE_DATASETS: ClassVar[RelationField] = RelationField( "alpha_dqRuleReferenceDatasets" ) """ TBC """ _convenience_properties: ClassVar[List[str]] = [ "alpha_dq_rule_base_dataset_qualified_name", "alpha_dq_rule_base_column_qualified_name", "alpha_dq_rule_reference_dataset_qualified_names", "alpha_dq_rule_reference_column_qualified_names", "alpha_dq_rule_source_sync_status", "alpha_dq_rule_source_sync_error_code", "alpha_dq_rule_source_sync_error_message", "alpha_dq_rule_source_synced_at", "alpha_dq_rule_latest_result", "alpha_dq_rule_latest_result_computed_at", "alpha_dq_rule_latest_result_fetched_at", "alpha_dq_rule_latest_metric_value", "alpha_dq_rule_latest_metric_value_computed_at", "alpha_dq_rule_dimension", "alpha_dq_rule_template_name", "alpha_dq_rule_status", "alpha_dq_rule_alert_priority", "alpha_dq_rule_config_arguments", "alpha_dq_rule_custom_s_q_l", "alpha_dq_rule_reference_columns", "alpha_dq_rule_base_column", "alpha_dq_rule_base_dataset", "alpha_dq_rule_template", "alpha_dq_rule_reference_datasets", ] @property def alpha_dq_rule_base_dataset_qualified_name(self) -> Optional[str]: return ( None if self.attributes is None else self.attributes.alpha_dq_rule_base_dataset_qualified_name ) @alpha_dq_rule_base_dataset_qualified_name.setter def alpha_dq_rule_base_dataset_qualified_name( self, alpha_dq_rule_base_dataset_qualified_name: Optional[str] ): if self.attributes is None: self.attributes = self.Attributes() self.attributes.alpha_dq_rule_base_dataset_qualified_name = ( alpha_dq_rule_base_dataset_qualified_name ) @property def alpha_dq_rule_base_column_qualified_name(self) -> Optional[str]: return ( None if self.attributes is None else self.attributes.alpha_dq_rule_base_column_qualified_name ) @alpha_dq_rule_base_column_qualified_name.setter def alpha_dq_rule_base_column_qualified_name( self, alpha_dq_rule_base_column_qualified_name: Optional[str] ): if self.attributes is None: self.attributes = self.Attributes() self.attributes.alpha_dq_rule_base_column_qualified_name = ( alpha_dq_rule_base_column_qualified_name ) @property def alpha_dq_rule_reference_dataset_qualified_names(self) -> Optional[Set[str]]: return ( None if self.attributes is None else self.attributes.alpha_dq_rule_reference_dataset_qualified_names ) @alpha_dq_rule_reference_dataset_qualified_names.setter def alpha_dq_rule_reference_dataset_qualified_names( self, alpha_dq_rule_reference_dataset_qualified_names: Optional[Set[str]] ): if self.attributes is None: self.attributes = self.Attributes() self.attributes.alpha_dq_rule_reference_dataset_qualified_names = ( alpha_dq_rule_reference_dataset_qualified_names ) @property def alpha_dq_rule_reference_column_qualified_names(self) -> Optional[Set[str]]: return ( None if self.attributes is None else self.attributes.alpha_dq_rule_reference_column_qualified_names ) @alpha_dq_rule_reference_column_qualified_names.setter def alpha_dq_rule_reference_column_qualified_names( self, alpha_dq_rule_reference_column_qualified_names: Optional[Set[str]] ): if self.attributes is None: self.attributes = self.Attributes() self.attributes.alpha_dq_rule_reference_column_qualified_names = ( alpha_dq_rule_reference_column_qualified_names ) @property def alpha_dq_rule_source_sync_status(self) -> Optional[alpha_DQSourceSyncStatus]: return ( None if self.attributes is None else self.attributes.alpha_dq_rule_source_sync_status ) @alpha_dq_rule_source_sync_status.setter def alpha_dq_rule_source_sync_status( self, alpha_dq_rule_source_sync_status: Optional[alpha_DQSourceSyncStatus] ): if self.attributes is None: self.attributes = self.Attributes() self.attributes.alpha_dq_rule_source_sync_status = ( alpha_dq_rule_source_sync_status ) @property def alpha_dq_rule_source_sync_error_code(self) -> Optional[str]: return ( None if self.attributes is None else self.attributes.alpha_dq_rule_source_sync_error_code ) @alpha_dq_rule_source_sync_error_code.setter def alpha_dq_rule_source_sync_error_code( self, alpha_dq_rule_source_sync_error_code: Optional[str] ): if self.attributes is None: self.attributes = self.Attributes() self.attributes.alpha_dq_rule_source_sync_error_code = ( alpha_dq_rule_source_sync_error_code ) @property def alpha_dq_rule_source_sync_error_message(self) -> Optional[str]: return ( None if self.attributes is None else self.attributes.alpha_dq_rule_source_sync_error_message ) @alpha_dq_rule_source_sync_error_message.setter def alpha_dq_rule_source_sync_error_message( self, alpha_dq_rule_source_sync_error_message: Optional[str] ): if self.attributes is None: self.attributes = self.Attributes() self.attributes.alpha_dq_rule_source_sync_error_message = ( alpha_dq_rule_source_sync_error_message ) @property def alpha_dq_rule_source_synced_at(self) -> Optional[datetime]: return ( None if self.attributes is None else self.attributes.alpha_dq_rule_source_synced_at ) @alpha_dq_rule_source_synced_at.setter def alpha_dq_rule_source_synced_at( self, alpha_dq_rule_source_synced_at: Optional[datetime] ): if self.attributes is None: self.attributes = self.Attributes() self.attributes.alpha_dq_rule_source_synced_at = alpha_dq_rule_source_synced_at @property def alpha_dq_rule_latest_result(self) -> Optional[alpha_DQResult]: return ( None if self.attributes is None else self.attributes.alpha_dq_rule_latest_result ) @alpha_dq_rule_latest_result.setter def alpha_dq_rule_latest_result( self, alpha_dq_rule_latest_result: Optional[alpha_DQResult] ): if self.attributes is None: self.attributes = self.Attributes() self.attributes.alpha_dq_rule_latest_result = alpha_dq_rule_latest_result @property def alpha_dq_rule_latest_result_computed_at(self) -> Optional[datetime]: return ( None if self.attributes is None else self.attributes.alpha_dq_rule_latest_result_computed_at ) @alpha_dq_rule_latest_result_computed_at.setter def alpha_dq_rule_latest_result_computed_at( self, alpha_dq_rule_latest_result_computed_at: Optional[datetime] ): if self.attributes is None: self.attributes = self.Attributes() self.attributes.alpha_dq_rule_latest_result_computed_at = ( alpha_dq_rule_latest_result_computed_at ) @property def alpha_dq_rule_latest_result_fetched_at(self) -> Optional[datetime]: return ( None if self.attributes is None else self.attributes.alpha_dq_rule_latest_result_fetched_at ) @alpha_dq_rule_latest_result_fetched_at.setter def alpha_dq_rule_latest_result_fetched_at( self, alpha_dq_rule_latest_result_fetched_at: Optional[datetime] ): if self.attributes is None: self.attributes = self.Attributes() self.attributes.alpha_dq_rule_latest_result_fetched_at = ( alpha_dq_rule_latest_result_fetched_at ) @property def alpha_dq_rule_latest_metric_value(self) -> Optional[str]: return ( None if self.attributes is None else self.attributes.alpha_dq_rule_latest_metric_value ) @alpha_dq_rule_latest_metric_value.setter def alpha_dq_rule_latest_metric_value( self, alpha_dq_rule_latest_metric_value: Optional[str] ): if self.attributes is None: self.attributes = self.Attributes() self.attributes.alpha_dq_rule_latest_metric_value = ( alpha_dq_rule_latest_metric_value ) @property def alpha_dq_rule_latest_metric_value_computed_at(self) -> Optional[datetime]: return ( None if self.attributes is None else self.attributes.alpha_dq_rule_latest_metric_value_computed_at ) @alpha_dq_rule_latest_metric_value_computed_at.setter def alpha_dq_rule_latest_metric_value_computed_at( self, alpha_dq_rule_latest_metric_value_computed_at: Optional[datetime] ): if self.attributes is None: self.attributes = self.Attributes() self.attributes.alpha_dq_rule_latest_metric_value_computed_at = ( alpha_dq_rule_latest_metric_value_computed_at ) @property def alpha_dq_rule_dimension(self) -> Optional[alpha_DQDimension]: return ( None if self.attributes is None else self.attributes.alpha_dq_rule_dimension ) @alpha_dq_rule_dimension.setter def alpha_dq_rule_dimension( self, alpha_dq_rule_dimension: Optional[alpha_DQDimension] ): if self.attributes is None: self.attributes = self.Attributes() self.attributes.alpha_dq_rule_dimension = alpha_dq_rule_dimension @property def alpha_dq_rule_template_name(self) -> Optional[str]: return ( None if self.attributes is None else self.attributes.alpha_dq_rule_template_name ) @alpha_dq_rule_template_name.setter def alpha_dq_rule_template_name(self, alpha_dq_rule_template_name: Optional[str]): if self.attributes is None: self.attributes = self.Attributes() self.attributes.alpha_dq_rule_template_name = alpha_dq_rule_template_name @property def alpha_dq_rule_status(self) -> Optional[alpha_DQRuleStatus]: return None if self.attributes is None else self.attributes.alpha_dq_rule_status @alpha_dq_rule_status.setter def alpha_dq_rule_status(self, alpha_dq_rule_status: Optional[alpha_DQRuleStatus]): if self.attributes is None: self.attributes = self.Attributes() self.attributes.alpha_dq_rule_status = alpha_dq_rule_status @property def alpha_dq_rule_alert_priority(self) -> Optional[alpha_DQRuleAlertPriority]: return ( None if self.attributes is None else self.attributes.alpha_dq_rule_alert_priority ) @alpha_dq_rule_alert_priority.setter def alpha_dq_rule_alert_priority( self, alpha_dq_rule_alert_priority: Optional[alpha_DQRuleAlertPriority] ): if self.attributes is None: self.attributes = self.Attributes() self.attributes.alpha_dq_rule_alert_priority = alpha_dq_rule_alert_priority @property def alpha_dq_rule_config_arguments(self) -> Optional[alpha_DQRuleConfigArguments]: return ( None if self.attributes is None else self.attributes.alpha_dq_rule_config_arguments ) @alpha_dq_rule_config_arguments.setter def alpha_dq_rule_config_arguments( self, alpha_dq_rule_config_arguments: Optional[alpha_DQRuleConfigArguments] ): if self.attributes is None: self.attributes = self.Attributes() self.attributes.alpha_dq_rule_config_arguments = alpha_dq_rule_config_arguments @property def alpha_dq_rule_custom_s_q_l(self) -> Optional[str]: return ( None if self.attributes is None else self.attributes.alpha_dq_rule_custom_s_q_l ) @alpha_dq_rule_custom_s_q_l.setter def alpha_dq_rule_custom_s_q_l(self, alpha_dq_rule_custom_s_q_l: Optional[str]): if self.attributes is None: self.attributes = self.Attributes() self.attributes.alpha_dq_rule_custom_s_q_l = alpha_dq_rule_custom_s_q_l @property def alpha_dq_rule_reference_columns(self) -> Optional[List[Column]]: return ( None if self.attributes is None else self.attributes.alpha_dq_rule_reference_columns ) @alpha_dq_rule_reference_columns.setter def alpha_dq_rule_reference_columns( self, alpha_dq_rule_reference_columns: Optional[List[Column]] ): if self.attributes is None: self.attributes = self.Attributes() self.attributes.alpha_dq_rule_reference_columns = ( alpha_dq_rule_reference_columns ) @property def alpha_dq_rule_base_column(self) -> Optional[Column]: return ( None if self.attributes is None else self.attributes.alpha_dq_rule_base_column ) @alpha_dq_rule_base_column.setter def alpha_dq_rule_base_column(self, alpha_dq_rule_base_column: Optional[Column]): if self.attributes is None: self.attributes = self.Attributes() self.attributes.alpha_dq_rule_base_column = alpha_dq_rule_base_column @property def alpha_dq_rule_base_dataset(self) -> Optional[Asset]: return ( None if self.attributes is None else self.attributes.alpha_dq_rule_base_dataset ) @alpha_dq_rule_base_dataset.setter def alpha_dq_rule_base_dataset(self, alpha_dq_rule_base_dataset: Optional[Asset]): if self.attributes is None: self.attributes = self.Attributes() self.attributes.alpha_dq_rule_base_dataset = alpha_dq_rule_base_dataset @property def alpha_dq_rule_template(self) -> Optional[alpha_DQRuleTemplate]: return ( None if self.attributes is None else self.attributes.alpha_dq_rule_template ) @alpha_dq_rule_template.setter def alpha_dq_rule_template( self, alpha_dq_rule_template: Optional[alpha_DQRuleTemplate] ): if self.attributes is None: self.attributes = self.Attributes() self.attributes.alpha_dq_rule_template = alpha_dq_rule_template @property def alpha_dq_rule_reference_datasets(self) -> Optional[List[Asset]]: return ( None if self.attributes is None else self.attributes.alpha_dq_rule_reference_datasets ) @alpha_dq_rule_reference_datasets.setter def alpha_dq_rule_reference_datasets( self, alpha_dq_rule_reference_datasets: Optional[List[Asset]] ): if self.attributes is None: self.attributes = self.Attributes() self.attributes.alpha_dq_rule_reference_datasets = ( alpha_dq_rule_reference_datasets ) class Attributes(DataQuality.Attributes): alpha_dq_rule_base_dataset_qualified_name: Optional[str] = Field( default=None, description="" ) alpha_dq_rule_base_column_qualified_name: Optional[str] = Field( default=None, description="" ) alpha_dq_rule_reference_dataset_qualified_names: Optional[Set[str]] = Field( default=None, description="" ) alpha_dq_rule_reference_column_qualified_names: Optional[Set[str]] = Field( default=None, description="" ) alpha_dq_rule_source_sync_status: Optional[alpha_DQSourceSyncStatus] = Field( default=None, description="" ) alpha_dq_rule_source_sync_error_code: Optional[str] = Field( default=None, description="" ) alpha_dq_rule_source_sync_error_message: Optional[str] = Field( default=None, description="" ) alpha_dq_rule_source_synced_at: Optional[datetime] = Field( default=None, description="" ) alpha_dq_rule_latest_result: Optional[alpha_DQResult] = Field( default=None, description="" ) alpha_dq_rule_latest_result_computed_at: Optional[datetime] = Field( default=None, description="" ) alpha_dq_rule_latest_result_fetched_at: Optional[datetime] = Field( default=None, description="" ) alpha_dq_rule_latest_metric_value: Optional[str] = Field( default=None, description="" ) alpha_dq_rule_latest_metric_value_computed_at: Optional[datetime] = Field( default=None, description="" ) alpha_dq_rule_dimension: Optional[alpha_DQDimension] = Field( default=None, description="" ) alpha_dq_rule_template_name: Optional[str] = Field(default=None, description="") alpha_dq_rule_status: Optional[alpha_DQRuleStatus] = Field( default=None, description="" ) alpha_dq_rule_alert_priority: Optional[alpha_DQRuleAlertPriority] = Field( default=None, description="" ) alpha_dq_rule_config_arguments: Optional[alpha_DQRuleConfigArguments] = Field( default=None, description="" ) alpha_dq_rule_custom_s_q_l: Optional[str] = Field(default=None, description="") alpha_dq_rule_reference_columns: Optional[List[Column]] = Field( default=None, description="" ) # relationship alpha_dq_rule_base_column: Optional[Column] = Field( default=None, description="" ) # relationship alpha_dq_rule_base_dataset: Optional[Asset] = Field( default=None, description="" ) # relationship alpha_dq_rule_template: Optional[alpha_DQRuleTemplate] = Field( default=None, description="" ) # relationship alpha_dq_rule_reference_datasets: Optional[List[Asset]] = Field( default=None, description="" ) # relationship @staticmethod def _generate_config_arguments_raw( *, is_alert_enabled: bool = True, custom_sql: Optional[str] = None, display_name: Optional[str] = None, dimension: Optional[alpha_DQDimension] = None, compare_operator: alpha_DQRuleThresholdCompareOperator, threshold_value: int, threshold_unit: Optional[alpha_DQRuleThresholdUnit] = None, column: Optional[Asset] = None, dq_priority: alpha_DQRuleAlertPriority, description: Optional[str] = None, ) -> str: config = { "isAlertEnabled": is_alert_enabled, "alpha_dqRuleTemplateConfigThresholdObject": { "alpha_dqRuleTemplateConfigThresholdCompareOperator": compare_operator, "alpha_dqRuleTemplateConfigThresholdValue": threshold_value, "alpha_dqRuleTemplateConfigThresholdUnit": threshold_unit, }, "alpha_dqRuleTemplateAdvancedSettings.dqPriority": dq_priority, } if column is not None: config["alpha_dqRuleTemplateConfigBaseColumnQualifiedName"] = ( column.qualified_name ) if description is not None: config["alpha_dqRuleTemplateConfigUserDescription"] = description if custom_sql is not None: config["alpha_dqRuleTemplateConfigCustomSQL"] = custom_sql if display_name is not None: config["alpha_dqRuleTemplateConfigDisplayName"] = display_name if dimension is not None: config["alpha_dqRuleTemplateConfigDimension"] = dimension return json.dumps(config) @staticmethod def _generate_uuid(): d = int(time.time() * 1000) random_bytes = uuid.uuid4().bytes rand_index = 0 def replace_char(c): nonlocal d, rand_index r = (d + random_bytes[rand_index % 16]) % 16 rand_index += 1 d = d // 16 if c == "x": return hex(r)[2:] elif c == "y": return hex((r & 0x3) | 0x8)[2:] # y -> 8 to b else: return c template = "xxxxxxxx-xxxx-4xxx-yxxx-xxxxxxxxxxxx" uuid_str = "".join(replace_char(c) if c in "xy" else c for c in template) return uuid_str @classmethod @init_guid def creator( cls, *, client: AtlanClient, rule_name: str, rule_type: str, asset: Asset, threshold_compare_operator: alpha_DQRuleThresholdCompareOperator, threshold_value: int, alert_priority: alpha_DQRuleAlertPriority, column: Optional[Asset] = None, threshold_unit: Optional[alpha_DQRuleThresholdUnit] = None, dimension: Optional[alpha_DQDimension] = None, custom_sql: Optional[str] = None, description: Optional[str] = None, ) -> alpha_DQRule.Attributes: template_config = client.dq_template_config_cache.get_template_config( rule_type ) if template_config is None: raise ErrorCode.DQ_RULE_NOT_FOUND.exception_with_parameters(rule_type) template_rule_name = template_config.get("name") template_qualified_name = template_config.get("qualified_name") if dimension is None: dimension = template_config.get("dimension") if threshold_unit is None: config = template_config.get("config") if config is not None: threashold_object = ( config.alpha_dq_rule_template_config_threshold_object ) threashold_object_json = json.loads(threashold_object) properties = threashold_object_json.get("properties", {}) threshold_unit_field = properties.get( "alpha_dqRuleTemplateConfigThresholdUnit", {} ) default_value = threshold_unit_field.get("default") threshold_unit = default_value config_arguments_raw = ( alpha_DQRule.Attributes._generate_config_arguments_raw( is_alert_enabled=True, custom_sql=custom_sql, display_name=rule_name, dimension=dimension, compare_operator=threshold_compare_operator, threshold_value=threshold_value, threshold_unit=threshold_unit, column=column, dq_priority=alert_priority, description=description, ) ) attr_dq = alpha_DQRule.Attributes( name="", alpha_dq_rule_config_arguments=alpha_DQRuleConfigArguments( alpha_dq_rule_threshold_object=alpha_DQRuleThresholdObject( alpha_dq_rule_threshold_compare_operator=threshold_compare_operator, alpha_dq_rule_threshold_value=threshold_value, alpha_dq_rule_threshold_unit=threshold_unit, ), alpha_dq_rule_config_arguments_raw=config_arguments_raw, ), alpha_dq_rule_base_dataset_qualified_name=asset.qualified_name, alpha_dq_rule_alert_priority=alert_priority, alpha_dq_rule_source_sync_status=alpha_DQSourceSyncStatus.IN_PROGRESS, alpha_dq_rule_status=alpha_DQRuleStatus.ACTIVE, alpha_dq_rule_base_dataset=asset, qualified_name=f"{asset.qualified_name}/rule/{str(cls._generate_uuid())}", alpha_dq_rule_dimension=dimension, alpha_dq_rule_template_name=template_rule_name, alpha_dq_rule_template=alpha_DQRuleTemplate.ref_by_qualified_name( qualified_name=template_qualified_name # type: ignore ), ) if column is not None: attr_dq.alpha_dq_rule_base_column_qualified_name = column.qualified_name attr_dq.alpha_dq_rule_base_column = column # type: ignore if custom_sql is not None: attr_dq.alpha_dq_rule_custom_s_q_l = custom_sql attr_dq.display_name = rule_name if description is not None: attr_dq.user_description = description return attr_dq attributes: alpha_DQRule.Attributes = Field( default_factory=lambda: alpha_DQRule.Attributes(), description=( "Map of attributes in the instance and their values. " "The specific keys of this map will vary by type, " "so are described in the sub-types of this schema." ), )
from .alpha__d_q_rule_template import alpha_DQRuleTemplate # noqa: E402, F401 from .asset import Asset # noqa: E402, F401 # from .column import co