sqlmesh.dbt.source

  1from __future__ import annotations
  2
  3import typing as t
  4
  5from pydantic import Field
  6
  7from sqlmesh.core.config.base import UpdateStrategy
  8from sqlmesh.dbt.column import ColumnConfig
  9from sqlmesh.dbt.common import GeneralConfig
 10from sqlmesh.dbt.relation import RelationType
 11from sqlmesh.dbt.util import DBT_VERSION
 12from sqlmesh.utils import AttributeDict
 13from sqlmesh.utils.errors import ConfigError
 14
 15if t.TYPE_CHECKING:
 16    from sqlmesh.dbt.context import DbtContext
 17
 18
 19class SourceConfig(GeneralConfig):
 20    """
 21    Args:
 22        name: The name of the table
 23        source_name: The name of the source that defines the table
 24        database: Name of the database where the table is stored. By default, the project's target database is used.
 25        schema: The scehma name as stored in the database. If not specified, the source name is used.
 26        identifier: The table name as stored in the database. If not specified, the source table name is used
 27        loader: Describes the tool that loads the source into the warehouse
 28        overrides: Override a source defined in the specified package
 29        freshness: Dictionary specifying maximum time, since the most recent record, to consider the source fresh
 30        loaded_at_field: Column name or expression that returns a timestamp indicating freshness
 31        quoting: Define which components of the qualified name (database, schema, identifier) to quote when resolving the source() method
 32        external: Dictionary of metadata properties specific to sources that point to external tables
 33        columns: Columns within the source
 34    """
 35
 36    # DBT configuration fields
 37    name: str = ""
 38    source_name_: str = Field("", alias="source_name")
 39    fqn_: t.List[str] = Field(default_factory=list, alias="fqn")
 40    database: t.Optional[str] = None
 41    schema_: t.Optional[str] = Field(None, alias="schema")
 42    identifier: t.Optional[str] = None
 43    loader: t.Optional[str] = None
 44    overrides: t.Optional[str] = None
 45    freshness: t.Optional[t.Dict[str, t.Any]] = {}
 46    loaded_at_field: t.Optional[str] = None
 47    quoting: t.Dict[str, t.Optional[bool]] = {}
 48    external: t.Optional[t.Dict[str, t.Any]] = {}
 49    source_meta: t.Optional[t.Dict[str, t.Any]] = {}
 50    columns: t.Dict[str, ColumnConfig] = {}
 51    event_time: t.Optional[str] = None
 52
 53    _canonical_name: t.Optional[str] = None
 54
 55    _FIELD_UPDATE_STRATEGY: t.ClassVar[t.Dict[str, UpdateStrategy]] = {
 56        **GeneralConfig._FIELD_UPDATE_STRATEGY,
 57        **{"columns": UpdateStrategy.KEY_EXTEND},
 58    }
 59
 60    @property
 61    def table_name(self) -> t.Optional[str]:
 62        return self.identifier or self.name
 63
 64    @property
 65    def config_name(self) -> str:
 66        return f"{self.source_name_}.{self.name}"
 67
 68    @property
 69    def fqn(self) -> str:
 70        return ".".join(self.fqn_)
 71
 72    def canonical_name(self, context: DbtContext) -> str:
 73        if self._canonical_name is None:
 74            source = context.get_callable_macro("source")
 75            if not source:
 76                raise ConfigError("'source' macro not found.")
 77
 78            try:
 79                relation = source(self.source_name_, self.name)
 80            except Exception as e:
 81                raise ConfigError(
 82                    f"'source' macro failed for '{self.config_name}' with exception '{e}'."
 83                )
 84
 85            relation = relation.quote(
 86                database=False,
 87                schema=False,
 88                identifier=False,
 89            )
 90            if relation.database == context.target.database:
 91                relation = relation.include(database=False)
 92            self._canonical_name = relation.render()
 93        return self._canonical_name
 94
 95    @property
 96    def relation_info(self) -> AttributeDict:
 97        extras = {}
 98        external_location = (
 99            self.source_meta.get("external_location", None) if self.source_meta else None
100        )
101        if external_location:
102            extras["external"] = external_location.replace("{name}", self.table_name)
103
104        if DBT_VERSION >= (1, 9, 0) and self.event_time:
105            extras["event_time_filter"] = {
106                "field_name": self.event_time,
107            }
108
109        return AttributeDict(
110            {
111                "database": self.database,
112                "schema": self.schema_,
113                "identifier": self.table_name,
114                "type": RelationType.External.value,
115                "quote_policy": AttributeDict(self.quoting),
116                **extras,
117            }
118        )