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

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


from __future__ import annotations

from typing import ClassVar, List, Optional

from pydantic.v1 import Field, validator

from pyatlan.model.fields.atlan_fields import KeywordField, RelationField, TextField

from .dbt import Dbt


[docs] class DbtTest(Dbt): """Description""" type_name: str = Field(default="DbtTest", allow_mutation=False) @validator("type_name") def validate_type_name(cls, v): if v != "DbtTest": raise ValueError("must be DbtTest") return v def __setattr__(self, name, value): if name in DbtTest._convenience_properties: return object.__setattr__(self, name, value) super().__setattr__(name, value) DBT_TEST_STATUS: ClassVar[KeywordField] = KeywordField( "dbtTestStatus", "dbtTestStatus" ) """ Details of the results of the test. For errors, it reads "ERROR". """ DBT_TEST_STATE: ClassVar[KeywordField] = KeywordField( "dbtTestState", "dbtTestState" ) """ Test results. Can be one of, in order of severity, "error", "fail", "warn", "pass". """ DBT_TEST_ERROR: ClassVar[TextField] = TextField("dbtTestError", "dbtTestError") """ Error message in the case of state being "error". """ DBT_TEST_RAW_SQL: ClassVar[TextField] = TextField( "dbtTestRawSQL", "dbtTestRawSQL.text" ) """ Raw SQL of the test. """ DBT_TEST_COMPILED_SQL: ClassVar[TextField] = TextField( "dbtTestCompiledSQL", "dbtTestCompiledSQL" ) """ Compiled SQL of the test. """ DBT_TEST_RAW_CODE: ClassVar[TextField] = TextField( "dbtTestRawCode", "dbtTestRawCode.text" ) """ Raw code of the test (when the test is defined using Python). """ DBT_TEST_COMPILED_CODE: ClassVar[TextField] = TextField( "dbtTestCompiledCode", "dbtTestCompiledCode" ) """ Compiled code of the test (when the test is defined using Python). """ DBT_TEST_LANGUAGE: ClassVar[TextField] = TextField( "dbtTestLanguage", "dbtTestLanguage" ) """ Language in which the test is written, for example: SQL or Python. """ DBT_SOURCES: ClassVar[RelationField] = RelationField("dbtSources") """ TBC """ SQL_ASSETS: ClassVar[RelationField] = RelationField("sqlAssets") """ TBC """ DBT_MODEL_COLUMNS: ClassVar[RelationField] = RelationField("dbtModelColumns") """ TBC """ DBT_MODELS: ClassVar[RelationField] = RelationField("dbtModels") """ TBC """ _convenience_properties: ClassVar[List[str]] = [ "dbt_test_status", "dbt_test_state", "dbt_test_error", "dbt_test_raw_s_q_l", "dbt_test_compiled_s_q_l", "dbt_test_raw_code", "dbt_test_compiled_code", "dbt_test_language", "dbt_sources", "sql_assets", "dbt_model_columns", "dbt_models", ] @property def dbt_test_status(self) -> Optional[str]: return None if self.attributes is None else self.attributes.dbt_test_status @dbt_test_status.setter def dbt_test_status(self, dbt_test_status: Optional[str]): if self.attributes is None: self.attributes = self.Attributes() self.attributes.dbt_test_status = dbt_test_status @property def dbt_test_state(self) -> Optional[str]: return None if self.attributes is None else self.attributes.dbt_test_state @dbt_test_state.setter def dbt_test_state(self, dbt_test_state: Optional[str]): if self.attributes is None: self.attributes = self.Attributes() self.attributes.dbt_test_state = dbt_test_state @property def dbt_test_error(self) -> Optional[str]: return None if self.attributes is None else self.attributes.dbt_test_error @dbt_test_error.setter def dbt_test_error(self, dbt_test_error: Optional[str]): if self.attributes is None: self.attributes = self.Attributes() self.attributes.dbt_test_error = dbt_test_error @property def dbt_test_raw_s_q_l(self) -> Optional[str]: return None if self.attributes is None else self.attributes.dbt_test_raw_s_q_l @dbt_test_raw_s_q_l.setter def dbt_test_raw_s_q_l(self, dbt_test_raw_s_q_l: Optional[str]): if self.attributes is None: self.attributes = self.Attributes() self.attributes.dbt_test_raw_s_q_l = dbt_test_raw_s_q_l @property def dbt_test_compiled_s_q_l(self) -> Optional[str]: return ( None if self.attributes is None else self.attributes.dbt_test_compiled_s_q_l ) @dbt_test_compiled_s_q_l.setter def dbt_test_compiled_s_q_l(self, dbt_test_compiled_s_q_l: Optional[str]): if self.attributes is None: self.attributes = self.Attributes() self.attributes.dbt_test_compiled_s_q_l = dbt_test_compiled_s_q_l @property def dbt_test_raw_code(self) -> Optional[str]: return None if self.attributes is None else self.attributes.dbt_test_raw_code @dbt_test_raw_code.setter def dbt_test_raw_code(self, dbt_test_raw_code: Optional[str]): if self.attributes is None: self.attributes = self.Attributes() self.attributes.dbt_test_raw_code = dbt_test_raw_code @property def dbt_test_compiled_code(self) -> Optional[str]: return ( None if self.attributes is None else self.attributes.dbt_test_compiled_code ) @dbt_test_compiled_code.setter def dbt_test_compiled_code(self, dbt_test_compiled_code: Optional[str]): if self.attributes is None: self.attributes = self.Attributes() self.attributes.dbt_test_compiled_code = dbt_test_compiled_code @property def dbt_test_language(self) -> Optional[str]: return None if self.attributes is None else self.attributes.dbt_test_language @dbt_test_language.setter def dbt_test_language(self, dbt_test_language: Optional[str]): if self.attributes is None: self.attributes = self.Attributes() self.attributes.dbt_test_language = dbt_test_language @property def dbt_sources(self) -> Optional[List[DbtSource]]: return None if self.attributes is None else self.attributes.dbt_sources @dbt_sources.setter def dbt_sources(self, dbt_sources: Optional[List[DbtSource]]): if self.attributes is None: self.attributes = self.Attributes() self.attributes.dbt_sources = dbt_sources @property def sql_assets(self) -> Optional[List[SQL]]: return None if self.attributes is None else self.attributes.sql_assets @sql_assets.setter def sql_assets(self, sql_assets: Optional[List[SQL]]): if self.attributes is None: self.attributes = self.Attributes() self.attributes.sql_assets = sql_assets @property def dbt_model_columns(self) -> Optional[List[DbtModelColumn]]: return None if self.attributes is None else self.attributes.dbt_model_columns @dbt_model_columns.setter def dbt_model_columns(self, dbt_model_columns: Optional[List[DbtModelColumn]]): if self.attributes is None: self.attributes = self.Attributes() self.attributes.dbt_model_columns = dbt_model_columns @property def dbt_models(self) -> Optional[List[DbtModel]]: return None if self.attributes is None else self.attributes.dbt_models @dbt_models.setter def dbt_models(self, dbt_models: Optional[List[DbtModel]]): if self.attributes is None: self.attributes = self.Attributes() self.attributes.dbt_models = dbt_models class Attributes(Dbt.Attributes): dbt_test_status: Optional[str] = Field(default=None, description="") dbt_test_state: Optional[str] = Field(default=None, description="") dbt_test_error: Optional[str] = Field(default=None, description="") dbt_test_raw_s_q_l: Optional[str] = Field(default=None, description="") dbt_test_compiled_s_q_l: Optional[str] = Field(default=None, description="") dbt_test_raw_code: Optional[str] = Field(default=None, description="") dbt_test_compiled_code: Optional[str] = Field(default=None, description="") dbt_test_language: Optional[str] = Field(default=None, description="") dbt_sources: Optional[List[DbtSource]] = Field( default=None, description="" ) # relationship sql_assets: Optional[List[SQL]] = Field( default=None, description="" ) # relationship dbt_model_columns: Optional[List[DbtModelColumn]] = Field( default=None, description="" ) # relationship dbt_models: Optional[List[DbtModel]] = Field( default=None, description="" ) # relationship attributes: DbtTest.Attributes = Field( default_factory=lambda: DbtTest.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 .dbt_model import DbtModel # noqa: E402, F401 from .dbt_model_column import DbtModelColumn # noqa: E402, F401 from .dbt_source import DbtSource # noqa: E402, F401 from .s_q_l import SQL # noqa: E402, F401