sqlmesh.dbt.target

   1from __future__ import annotations
   2
   3import abc
   4import typing as t
   5from pathlib import Path
   6
   7from dbt.adapters.base import BaseRelation, Column
   8from pydantic import Field, AliasChoices
   9
  10from sqlmesh.core.console import get_console
  11from sqlmesh.core.config.connection import (
  12    AthenaConnectionConfig,
  13    BigQueryConnectionConfig,
  14    BigQueryConnectionMethod,
  15    BigQueryPriority,
  16    ClickhouseConnectionConfig,
  17    ConnectionConfig,
  18    DatabricksConnectionConfig,
  19    DuckDBConnectionConfig,
  20    MSSQLConnectionConfig,
  21    PostgresConnectionConfig,
  22    RedshiftConnectionConfig,
  23    SnowflakeConnectionConfig,
  24    TrinoAuthenticationMethod,
  25    TrinoConnectionConfig,
  26)
  27from sqlmesh.core.model import (
  28    IncrementalByTimeRangeKind,
  29    IncrementalByUniqueKeyKind,
  30    IncrementalUnmanagedKind,
  31)
  32from sqlmesh.core.schema_diff import NestedSupport
  33from sqlmesh.dbt.common import DbtConfig
  34from sqlmesh.dbt.relation import Policy
  35from sqlmesh.dbt.util import DBT_VERSION
  36from sqlmesh.utils import AttributeDict, classproperty
  37from sqlmesh.utils.errors import ConfigError
  38from sqlmesh.utils.pydantic import field_validator, model_validator
  39
  40IncrementalKind = t.Union[
  41    t.Type[IncrementalByUniqueKeyKind],
  42    t.Type[IncrementalByTimeRangeKind],
  43    t.Type[IncrementalUnmanagedKind],
  44]
  45
  46# We only serialize a subset of fields in order to avoid persisting sensitive information
  47SERIALIZABLE_FIELDS = {
  48    # core
  49    "name",
  50    "schema_",
  51    "type",
  52    "threads",
  53    # snowflake
  54    "database",
  55    "warehouse",
  56    "user",
  57    "role",
  58    "account",
  59    # postgres/redshift
  60    "dbname",
  61    "host",
  62    "port",
  63    # bigquery
  64    "project",
  65    "dataset",
  66}
  67
  68SCHEMA_DIFFER_OVERRIDES = {
  69    "schema_differ_overrides": {
  70        "treat_alter_data_type_as_destructive": True,
  71        "nested_support": NestedSupport.IGNORE,
  72    }
  73}
  74
  75
  76def with_schema_differ_overrides(
  77    func: t.Callable[..., ConnectionConfig],
  78) -> t.Callable[..., ConnectionConfig]:
  79    """Decorator that merges default config with kwargs."""
  80
  81    def wrapper(self: TargetConfig, **kwargs: t.Any) -> ConnectionConfig:
  82        merged_kwargs = {**SCHEMA_DIFFER_OVERRIDES, **kwargs}
  83        return func(self, **merged_kwargs)
  84
  85    return wrapper
  86
  87
  88class TargetConfig(abc.ABC, DbtConfig):
  89    """
  90    Configuration for DBT profile target
  91
  92    Args:
  93        type: The type of the data warehouse
  94        name: The name of this target
  95        database: Name of the database
  96        schema_: Name of the schema
  97        threads: The number of threads to run on
  98    """
  99
 100    # dbt
 101    type: str = "none"
 102    name: str
 103    database: str
 104    schema_: str = Field(alias="schema")
 105    threads: int = 1
 106    profile_name: t.Optional[str] = None
 107
 108    @classmethod
 109    def load(cls, data: t.Dict[str, t.Any]) -> TargetConfig:
 110        """
 111        Loads the configuration from the yaml provided for a profile target
 112
 113        Args:
 114            data: The yaml for the project's target output
 115
 116        Returns:
 117            The configuration of the provided profile target
 118        """
 119        db_type = data["type"]
 120        if db_type == "databricks":
 121            return DatabricksConfig(**data)
 122        if db_type == "duckdb":
 123            return DuckDbConfig(**data)
 124        if db_type == "postgres":
 125            return PostgresConfig(**data)
 126        if db_type == "redshift":
 127            return RedshiftConfig(**data)
 128        if db_type == "snowflake":
 129            return SnowflakeConfig(**data)
 130        if db_type == "bigquery":
 131            return BigQueryConfig(**data)
 132        if db_type == "sqlserver":
 133            return MSSQLConfig(**data)
 134        if db_type == "trino":
 135            return TrinoConfig(**data)
 136        if db_type == "clickhouse":
 137            return ClickhouseConfig(**data)
 138        if db_type == "athena":
 139            return AthenaConfig(**data)
 140
 141        raise ConfigError(f"{db_type} not supported.")
 142
 143    def default_incremental_strategy(self, kind: IncrementalKind) -> str:
 144        """The default incremental strategy for the db"""
 145        raise NotImplementedError
 146
 147    @with_schema_differ_overrides
 148    def to_sqlmesh(self, **kwargs: t.Any) -> ConnectionConfig:
 149        """Converts target config to SQLMesh connection config"""
 150        raise NotImplementedError
 151
 152    def attribute_dict(self) -> AttributeDict:
 153        fields = self.dict(include=SERIALIZABLE_FIELDS).copy()
 154        fields["target_name"] = self.name
 155        return AttributeDict(fields)
 156
 157    @classproperty
 158    def quote_policy(cls) -> Policy:
 159        return Policy()
 160
 161    @property
 162    def extra(self) -> t.Set[str]:
 163        return self.extra_fields(set(self.dict()))
 164
 165    @classproperty
 166    def relation_class(cls) -> t.Type[BaseRelation]:
 167        return BaseRelation
 168
 169    @classproperty
 170    def column_class(cls) -> t.Type[Column]:
 171        return Column
 172
 173    @property
 174    def dialect(self) -> str:
 175        return self.type
 176
 177
 178DUCKDB_IN_MEMORY = ":memory:"
 179
 180
 181class DuckDbConfig(TargetConfig):
 182    """
 183    Connection config for DuckDb target
 184
 185    Args:
 186        path: Location of the database file. If not specified, an in memory database is used.
 187        extensions: A list of autoloadable extensions to load.
 188        settings: A dictionary of settings to pass into the duckdb connector.
 189        secrets: A list of secrets to pass to the secret manager in the duckdb connector.
 190        filesystems: A list of `fsspec` filesystems to register in the duckdb connection.
 191    """
 192
 193    type: t.Literal["duckdb"] = "duckdb"
 194    database: str = "main"
 195    schema_: str = Field(default="main", alias="schema")
 196    path: str = DUCKDB_IN_MEMORY
 197    extensions: t.Optional[t.List[str]] = None
 198    settings: t.Optional[t.Dict[str, t.Any]] = None
 199    secrets: t.Optional[t.List[t.Dict[str, t.Any]]] = None
 200    filesystems: t.Optional[t.List[t.Dict[str, t.Any]]] = None
 201
 202    @model_validator(mode="before")
 203    def validate_authentication(cls, data: t.Any) -> t.Any:
 204        if not isinstance(data, dict):
 205            return data
 206
 207        if "database" not in data and DBT_VERSION >= (1, 5, 0):
 208            path = data.get("path")
 209            data["database"] = (
 210                "memory"
 211                if path is None or path == DUCKDB_IN_MEMORY
 212                else Path(t.cast(str, path)).stem
 213            )
 214
 215        if "threads" in data and t.cast(int, data["threads"]) > 1:
 216            get_console().log_warning("DuckDB does not support concurrency - setting threads to 1.")
 217
 218        return data
 219
 220    def default_incremental_strategy(self, kind: IncrementalKind) -> str:
 221        return "delete+insert"
 222
 223    @classproperty
 224    def relation_class(cls) -> t.Type[BaseRelation]:
 225        from dbt.adapters.duckdb.relation import DuckDBRelation
 226
 227        return DuckDBRelation
 228
 229    @with_schema_differ_overrides
 230    def to_sqlmesh(self, **kwargs: t.Any) -> ConnectionConfig:
 231        if self.extensions is not None:
 232            kwargs["extensions"] = self.extensions
 233        if self.settings is not None:
 234            kwargs["connector_config"] = self.settings
 235        if self.secrets is not None:
 236            kwargs["secrets"] = self.secrets
 237        if self.filesystems is not None:
 238            kwargs["filesystems"] = self.filesystems
 239        return DuckDBConnectionConfig(
 240            database=self.path,
 241            concurrent_tasks=1,
 242            **kwargs,
 243        )
 244
 245
 246class SnowflakeConfig(TargetConfig):
 247    """
 248    Project connection and operational configuration for the Snowflake target
 249
 250    Args:
 251        account: Snowflake account
 252        warehouse: Name of the warehouse
 253        user: Name of the user
 254        password: User's password
 255        role: Role of the user
 256        client_session_keep_alive: A boolean flag to extend the duration of the Snowflake session beyond 4 hours
 257        query_tag: tag for the query in Snowflake
 258        connect_retries: Number of times to retry if the Snowflake connector encounters an error
 259        connect_timeout: Number of seconds to wait between failed attempts
 260        retry_on_database_errors: A boolean flag to retry if a Snowflake connector Database error is encountered
 261        retry_all: A boolean flag to retry on all Snowflake connector errors
 262        authenticator: SSO authentication: Snowflake authentication method
 263        private_key: Key pair authentication: Private key
 264        private_key_path: Key pair authentication: Path to the private key, used instead of private_key
 265        private_key_passphrase: Key pair authentication: passphrase used to decrypt private key (if encrypted)
 266        token: OAuth authentication: The Snowflake OAuth 2.0 access token
 267    """
 268
 269    type: t.Literal["snowflake"] = "snowflake"
 270    account: str
 271    user: str
 272
 273    # User and password authentication
 274    password: t.Optional[str] = None
 275
 276    # SSO authentication
 277    authenticator: t.Optional[str] = None
 278
 279    # Key Pair Auth
 280    private_key: t.Optional[str] = None
 281    private_key_path: t.Optional[str] = None
 282    private_key_passphrase: t.Optional[str] = None
 283
 284    # TODO add other forms of authentication
 285
 286    # oauth access token
 287    token: t.Optional[str] = None
 288
 289    # Optional
 290    warehouse: t.Optional[str] = None
 291    role: t.Optional[str] = None
 292    client_session_keep_alive: bool = False
 293    query_tag: t.Optional[str] = None
 294    connect_retries: int = 0
 295    connect_timeout: int = 10
 296    retry_on_database_errors: bool = False
 297    retry_all: bool = False
 298
 299    @model_validator(mode="before")
 300    @classmethod
 301    def validate_authentication(cls, data: t.Any) -> t.Any:
 302        if not isinstance(data, dict) or (
 303            data.get("password")
 304            or data.get("authenticator")
 305            or data.get("private_key")
 306            or data.get("private_key_path")
 307        ):
 308            return data
 309
 310        raise ConfigError("No supported Snowflake authentication method found in target profile.")
 311
 312    def default_incremental_strategy(self, kind: IncrementalKind) -> str:
 313        return "merge"
 314
 315    @classproperty
 316    def relation_class(cls) -> t.Type[BaseRelation]:
 317        from dbt.adapters.snowflake import SnowflakeRelation
 318
 319        return SnowflakeRelation
 320
 321    @classproperty
 322    def column_class(cls) -> t.Type[Column]:
 323        from dbt.adapters.snowflake import SnowflakeColumn
 324
 325        return SnowflakeColumn
 326
 327    @with_schema_differ_overrides
 328    def to_sqlmesh(self, **kwargs: t.Any) -> ConnectionConfig:
 329        return SnowflakeConnectionConfig(
 330            user=self.user,
 331            password=self.password,
 332            authenticator=self.authenticator,
 333            account=self.account,
 334            warehouse=self.warehouse,
 335            database=self.database,
 336            role=self.role,
 337            concurrent_tasks=self.threads,
 338            token=self.token,
 339            private_key=self.private_key,
 340            private_key_path=self.private_key_path,
 341            private_key_passphrase=self.private_key_passphrase,
 342            **kwargs,
 343        )
 344
 345    @classproperty
 346    def quote_policy(cls) -> Policy:
 347        return Policy(database=False, schema=False, identifier=False)
 348
 349
 350class PostgresConfig(TargetConfig):
 351    """
 352    Project connection and operational configuration for the Postgres target
 353
 354    Args:
 355        host: The Postgres host to connect to
 356        user: Name of the user
 357        password: User's password
 358        port: The port to connect to
 359        dbname: Name of the database
 360        keepalives_idle: Seconds between TCP keepalive packets
 361        connect_timeout: Number of seconds to wait between failed attempts
 362        retries: Number of times to retry if the Postgres connector encounters an error
 363        search_path: Overrides the default search path
 364        role: Role of the user
 365        sslmode: SSL Mode used to connect to the database
 366    """
 367
 368    type: t.Literal["postgres"] = "postgres"
 369    host: str
 370    user: str
 371    password: str = Field(validation_alias=AliasChoices("pass", "password"))
 372    port: int
 373    dbname: str
 374    keepalives_idle: t.Optional[int] = None
 375    connect_timeout: int = 10
 376    retries: int = 1  # Currently Unsupported by SQLMesh
 377    search_path: t.Optional[str] = None  # Currently Unsupported by SQLMesh
 378    role: t.Optional[str] = None
 379    sslmode: t.Optional[str] = None
 380
 381    @model_validator(mode="before")
 382    @classmethod
 383    def validate_database(cls, data: t.Any) -> t.Any:
 384        if not isinstance(data, dict):
 385            return data
 386
 387        data["database"] = data.get("database") or data.get("dbname")
 388        if not data["database"]:
 389            raise ConfigError("Either database or dbname must be set")
 390
 391        return data
 392
 393    @field_validator("port")
 394    @classmethod
 395    def _validate_port(cls, v: t.Union[int, str]) -> int:
 396        return int(v)
 397
 398    def default_incremental_strategy(self, kind: IncrementalKind) -> str:
 399        return "delete+insert" if kind is IncrementalByUniqueKeyKind else "append"
 400
 401    @with_schema_differ_overrides
 402    def to_sqlmesh(self, **kwargs: t.Any) -> ConnectionConfig:
 403        return PostgresConnectionConfig(
 404            host=self.host,
 405            user=self.user,
 406            password=self.password,
 407            port=self.port,
 408            database=self.dbname,
 409            keepalives_idle=self.keepalives_idle,
 410            concurrent_tasks=self.threads,
 411            connect_timeout=self.connect_timeout,
 412            role=self.role,
 413            sslmode=self.sslmode,
 414            **kwargs,
 415        )
 416
 417
 418class RedshiftConfig(TargetConfig):
 419    """
 420    Project connection and operational configuration for the Redshift target
 421
 422    Args:
 423        host: The Redshift host to connect to
 424        user: Name of the user
 425        password: User's password
 426        port: The port to connect to
 427        dbname: Name of the database
 428        connect_timeout: Number of seconds to wait between failed attempts
 429        ra3_node: Enables cross-database sources
 430        search_path: Overrides the default search path
 431        sslmode: SSL Mode used to connect to the database
 432    """
 433
 434    # TODO add other forms of authentication
 435    type: t.Literal["redshift"] = "redshift"
 436    host: str
 437    user: str
 438    password: str = Field(validation_alias=AliasChoices("pass", "password"))
 439    port: int
 440    dbname: str
 441    connect_timeout: t.Optional[int] = None
 442    ra3_node: bool = True
 443    search_path: t.Optional[str] = None
 444    sslmode: t.Optional[str] = None
 445
 446    @model_validator(mode="before")
 447    @classmethod
 448    def validate_database(cls, data: t.Any) -> t.Any:
 449        if not isinstance(data, dict):
 450            return data
 451
 452        data["database"] = data.get("database") or data.get("dbname")
 453        if not data["database"]:
 454            raise ConfigError("Either database or dbname must be set")
 455
 456        return data
 457
 458    def default_incremental_strategy(self, kind: IncrementalKind) -> str:
 459        return "append"
 460
 461    @classproperty
 462    def relation_class(cls) -> t.Type[BaseRelation]:
 463        from dbt.adapters.redshift import RedshiftRelation
 464
 465        return RedshiftRelation
 466
 467    @classproperty
 468    def column_class(cls) -> t.Type[Column]:
 469        if DBT_VERSION < (1, 6, 0):
 470            from dbt.adapters.redshift import RedshiftColumn  # type: ignore
 471
 472            return RedshiftColumn
 473        return super(RedshiftConfig, cls).column_class
 474
 475    @with_schema_differ_overrides
 476    def to_sqlmesh(self, **kwargs: t.Any) -> ConnectionConfig:
 477        return RedshiftConnectionConfig(
 478            user=self.user,
 479            password=self.password,
 480            database=self.database,
 481            host=self.host,
 482            port=self.port,
 483            sslmode=self.sslmode,
 484            timeout=self.connect_timeout,
 485            concurrent_tasks=self.threads,
 486            **kwargs,
 487        )
 488
 489
 490class DatabricksConfig(TargetConfig):
 491    """
 492    Project connection and operational configuration for the Databricks target
 493
 494    Args:
 495        catalog: Catalog name to use for Unity Catalog
 496        host: The Databricks host to connect to
 497        http_path: The Databricks compute resources URL
 498        token: Personal access token
 499        database: Name of the database. Not applicable for Databricks and ignored
 500    """
 501
 502    type: t.Literal["databricks"] = "databricks"
 503    host: str
 504    http_path: str
 505    token: t.Optional[str] = None  # only required if auth_type is not set to 'oauth'
 506    database: t.Optional[str] = Field(alias="catalog")  # type: ignore
 507    auth_type: t.Optional[str] = None
 508    client_id: t.Optional[str] = None
 509    client_secret: t.Optional[str] = None
 510
 511    def default_incremental_strategy(self, kind: IncrementalKind) -> str:
 512        return "merge"
 513
 514    @classproperty
 515    def relation_class(cls) -> t.Type[BaseRelation]:
 516        from dbt.adapters.databricks.relation import DatabricksRelation
 517
 518        return DatabricksRelation
 519
 520    @classproperty
 521    def column_class(cls) -> t.Type[Column]:
 522        from dbt.adapters.databricks.column import DatabricksColumn
 523
 524        return DatabricksColumn
 525
 526    @with_schema_differ_overrides
 527    def to_sqlmesh(self, **kwargs: t.Any) -> ConnectionConfig:
 528        return DatabricksConnectionConfig(
 529            server_hostname=self.host,
 530            http_path=self.http_path,
 531            access_token=self.token,
 532            concurrent_tasks=self.threads,
 533            catalog=self.database,
 534            auth_type="databricks-oauth" if self.auth_type == "oauth" else self.auth_type,
 535            oauth_client_id=self.client_id,
 536            oauth_client_secret=self.client_secret,
 537            **kwargs,
 538        )
 539
 540
 541class BigQueryConfig(TargetConfig):
 542    """
 543    Project connection and operational configuration for the BigQuery target
 544
 545    Args:
 546        type: The type of the target (bigquery)
 547        method: The BigQuery authentication method to use
 548        project: The BigQuery project to connect to
 549        location: The BigQuery location to connect to
 550        keyfile: The path to the BigQuery keyfile
 551        keyfile_json: The BigQuery keyfile as a JSON string
 552        token: The BigQuery token
 553        refresh_token: The BigQuery refresh token
 554        client_id: The BigQuery client ID
 555        client_secret: The BigQuery client secret
 556        token_uri: The BigQuery token URI
 557        scopes: The BigQuery scopes
 558        impersonated_service_account: The service account to impersonate
 559        job_creation_timeout_seconds: The maximum amount of time, in seconds, to wait for the underlying job to be created
 560        job_execution_timeout_seconds: The maximum amount of time, in seconds, to wait for the underlying job to complete
 561        timeout_seconds: Alias for job_execution_timeout_seconds
 562        job_retries: The number of times to retry the underlying job if it fails
 563        retries: Alias for job_retries
 564        job_retry_deadline_seconds: Total number of seconds to wait while retrying the same query
 565        priority: The priority of the underlying job
 566        maximum_bytes_billed: The maximum number of bytes to be billed for the underlying job
 567    """
 568
 569    type: t.Literal["bigquery"] = "bigquery"
 570    method: t.Optional[str] = BigQueryConnectionMethod.OAUTH
 571    dataset: t.Optional[str] = None
 572    project: t.Optional[str] = None
 573    execution_project: t.Optional[str] = None
 574    quota_project: t.Optional[str] = None
 575    location: t.Optional[str] = None
 576    keyfile: t.Optional[str] = None
 577    keyfile_json: t.Optional[t.Dict[str, t.Any]] = None
 578    token: t.Optional[str] = None
 579    refresh_token: t.Optional[str] = None
 580    client_id: t.Optional[str] = None
 581    client_secret: t.Optional[str] = None
 582    token_uri: t.Optional[str] = None
 583    scopes: t.Tuple[str, ...] = (
 584        "https://www.googleapis.com/auth/bigquery",
 585        "https://www.googleapis.com/auth/cloud-platform",
 586        "https://www.googleapis.com/auth/drive",
 587    )
 588    impersonated_service_account: t.Optional[str] = None
 589    job_creation_timeout_seconds: t.Optional[int] = None
 590    job_execution_timeout_seconds: t.Optional[int] = None
 591    timeout_seconds: t.Optional[int] = None  # To support legacy config
 592    job_retries: t.Optional[int] = None
 593    retries: int = 1  # To support legacy config
 594    job_retry_deadline_seconds: t.Optional[int] = None
 595    priority: BigQueryPriority = BigQueryPriority.INTERACTIVE
 596    maximum_bytes_billed: t.Optional[int] = None
 597
 598    @model_validator(mode="before")
 599    @classmethod
 600    def validate_fields(cls, data: t.Any) -> t.Any:
 601        if not isinstance(data, dict):
 602            return data
 603
 604        # dbt treats schema and dataset interchangeably
 605        schema = data.get("schema") or data.get("dataset")
 606        if not schema:
 607            raise ConfigError("Either schema or dataset must be set")
 608        data["dataset"] = data["schema"] = schema
 609
 610        # dbt treats database and project interchangeably
 611        database = data.get("database") or data.get("project")
 612        if not database:
 613            raise ConfigError("Either database or project must be set")
 614        data["database"] = data["project"] = database
 615
 616        return data
 617
 618    def default_incremental_strategy(self, kind: IncrementalKind) -> str:
 619        return "merge"
 620
 621    @classproperty
 622    def relation_class(cls) -> t.Type[BaseRelation]:
 623        from dbt.adapters.bigquery.relation import BigQueryRelation
 624
 625        return BigQueryRelation
 626
 627    @classproperty
 628    def column_class(cls) -> t.Type[Column]:
 629        from dbt.adapters.bigquery import BigQueryColumn
 630
 631        return BigQueryColumn
 632
 633    @with_schema_differ_overrides
 634    def to_sqlmesh(self, **kwargs: t.Any) -> ConnectionConfig:
 635        job_retries = self.job_retries if self.job_retries is not None else self.retries
 636        job_execution_timeout_seconds = (
 637            self.job_execution_timeout_seconds
 638            if self.job_execution_timeout_seconds is not None
 639            else self.timeout_seconds
 640        )
 641        return BigQueryConnectionConfig(
 642            method=self.method,
 643            project=self.database,
 644            execution_project=self.execution_project,
 645            quota_project=self.quota_project,
 646            location=self.location,
 647            concurrent_tasks=self.threads,
 648            keyfile=self.keyfile,
 649            keyfile_json=self.keyfile_json,
 650            token=self.token,
 651            refresh_token=self.refresh_token,
 652            client_id=self.client_id,
 653            client_secret=self.client_secret,
 654            token_uri=self.token_uri,
 655            scopes=self.scopes,
 656            impersonated_service_account=self.impersonated_service_account,
 657            job_creation_timeout_seconds=self.job_creation_timeout_seconds,
 658            job_execution_timeout_seconds=job_execution_timeout_seconds,
 659            job_retries=job_retries,
 660            job_retry_deadline_seconds=self.job_retry_deadline_seconds,
 661            priority=self.priority,
 662            maximum_bytes_billed=self.maximum_bytes_billed,
 663            **kwargs,
 664        )
 665
 666
 667class MSSQLConfig(TargetConfig):
 668    """
 669    Project connection and operational configuration for the SQL Server (MSSQL) target
 670
 671    Args:
 672        host: The MSSQL server host to connect to
 673        server: Alias for host
 674        port: The MSSQL server port to connect to
 675        user: User name for authentication
 676        username: Alias for user
 677        UID: Alias for user
 678        password: User password for authentication
 679        PWD: Alias for password
 680        login_timeout: The number of seconds to wait for a login to complete
 681        query_timeout: The number of seconds to wait for a query to complete
 682        authentication: The authentication method to use (only "sql" is supported)
 683        schema_authorization: The principal who should own created schemas, not supported by SQLMesh
 684        driver: ODBC driver to use, not used by SQLMesh
 685        encrypt: A boolean flag to enable server connection encryption, not used by SQLMesh
 686        trust_cert: A boolean flag to trust the server certificate, not used by SQLMesh
 687        retries: Number of times to retry if the SQL Server connector encounters an error, not used by SQLMesh
 688        windows_login: A boolean flag to use Windows Authentication, not used by SQLMesh
 689        tenant_id: The tenant ID of the Azure Active Directory instance, not used by SQLMesh
 690        client_id: The client ID of the Azure Active Directory service principal, not used by SQLMesh
 691        client_secret: The client secret of the Azure Active Directory service principal, not used by SQLMesh
 692    """
 693
 694    type: t.Literal["sqlserver"] = "sqlserver"
 695    host: t.Optional[str] = None
 696    server: t.Optional[str] = None
 697    port: int = 1433
 698    database: str = Field(default="master")
 699    schema_: str = Field(default="dbo", alias="schema")
 700    user: t.Optional[str] = None
 701    username: t.Optional[str] = None
 702    UID: t.Optional[str] = None
 703    password: t.Optional[str] = None
 704    PWD: t.Optional[str] = None
 705    threads: int = 4
 706    login_timeout: t.Optional[int] = None
 707    query_timeout: t.Optional[int] = None
 708    authentication: t.Optional[str] = "sql"
 709    schema_authorization: t.Optional[str] = None  # Not supported by SQLMesh
 710
 711    # Unused ODBC parameters (SQLMesh uses pymssql instead of ODBC)
 712    driver: t.Optional[str] = None
 713    encrypt: t.Optional[bool] = None
 714    trust_cert: t.Optional[bool] = None
 715    retries: t.Optional[int] = None
 716
 717    # Unused authentication parameters (not supported by pymssql)
 718    windows_login: t.Optional[bool] = None  # pymssql doesn't require this flag for Windows Auth
 719    tenant_id: t.Optional[str] = None  # Azure Active Directory auth
 720    client_id: t.Optional[str] = None  # Azure Active Directory auth
 721    client_secret: t.Optional[str] = None  # Azure Active Directory auth
 722
 723    @model_validator(mode="before")
 724    @classmethod
 725    def validate_alias_fields(cls, data: t.Any) -> t.Any:
 726        if not isinstance(data, dict):
 727            return data
 728
 729        data["host"] = data.get("host") or data.get("server")
 730        if not data["host"]:
 731            raise ConfigError("Either host or server must be set")
 732
 733        data["user"] = data.get("user") or data.get("username") or data.get("UID")
 734        if not data["user"]:
 735            raise ConfigError("One of user, username, or UID must be set")
 736
 737        data["password"] = data.get("password") or data.get("PWD")
 738        if not data["password"]:
 739            raise ConfigError("Either password or PWD must be set")
 740
 741        return data
 742
 743    @field_validator("authentication")
 744    @classmethod
 745    def _validate_authentication(cls, v: str) -> str:
 746        if v != "sql":
 747            raise ConfigError("Only SQL and Windows Authentication are supported for SQL Server")
 748        return v
 749
 750    @field_validator("port")
 751    @classmethod
 752    def _validate_port(cls, v: t.Union[int, str]) -> int:
 753        return int(v)
 754
 755    def default_incremental_strategy(self, kind: IncrementalKind) -> str:
 756        # https://github.com/microsoft/dbt-fabric/blob/main/dbt/include/fabric/macros/materializations/models/incremental/incremental_strategies.sql
 757        return "delete+insert" if kind is IncrementalByUniqueKeyKind else "append"
 758
 759    @classproperty
 760    def column_class(cls) -> t.Type[Column]:
 761        try:
 762            # 1.8.0+
 763            from dbt.adapters.sqlserver.sqlserver_column import SQLServerColumn
 764        except ImportError:
 765            # <1.8.0
 766            from dbt.adapters.sqlserver.sql_server_column import SQLServerColumn  # type: ignore
 767
 768        return SQLServerColumn
 769
 770    @property
 771    def dialect(self) -> str:
 772        return "tsql"
 773
 774    @with_schema_differ_overrides
 775    def to_sqlmesh(self, **kwargs: t.Any) -> ConnectionConfig:
 776        return MSSQLConnectionConfig(
 777            host=self.host,
 778            user=self.user,
 779            password=self.password,
 780            port=self.port,
 781            database=self.database,
 782            timeout=self.query_timeout,
 783            login_timeout=self.login_timeout,
 784            concurrent_tasks=self.threads,
 785            **kwargs,
 786        )
 787
 788
 789class TrinoConfig(TargetConfig):
 790    """
 791    Project connection and operational configuration for the Trino target.
 792
 793    Args:
 794        method: The Trino authentication method to use
 795        host: The server host to connect to
 796        port: The MSSQL server port to connect to
 797        database: Name of the Trino database/catalog
 798        schema: Name of the Trino schema
 799        user: User name for authentication
 800        password: User password for authentication
 801        roles: Trino catalog roles
 802        session_properties: Trino session properties
 803        retries: Number of times to retry if the Trino connector encounters an error
 804        timezone: The timezone to use for the Trino session
 805        http_headers: HTTP Headers to send alongside requests to Trino
 806        http_scheme: The HTTP scheme to use for requests to Trino (default: http, or https if kerberos, ldap or jwt auth)
 807        threads: The number of threads to run on
 808        impersonation_user:  LDAP authentication: override the provided username
 809        keytab: Kerberos authentication: Path to keytab
 810        krb5_config: Kerberos authentication: Path to config
 811        principal: Kerberos authentication: Principal
 812        service_name: Kerberos authentication: Service name
 813        hostname_override: Kerberos authentication: hostname for a host whose DNS name doesn't match
 814        mutual_authentication: Kerberos authentication: Boolean flag for mutual authentication.
 815        force_preemptive: Kerberos authentication: Boolean flag to preemptively initiate the GSS exchange.
 816        sanitize_mutual_error_response: Kerberos authentication: Boolean flag to strip content and headers from error responses.
 817        delegate: Kerberos authentication: Boolean flag for credential delegation (`GSS_C_DELEG_FLAG`)
 818        jwt_token: JWT authentication: JWT string
 819        client_certificate: Certification authentication: Path to client certificate
 820        client_private_key: Certification authentication: Path to client private key
 821        cert: Certification authentication: Full path to a certificate file
 822    """
 823
 824    _method_to_auth_enum: t.ClassVar[t.Dict[str, TrinoAuthenticationMethod]] = {
 825        "none": TrinoAuthenticationMethod.NO_AUTH,
 826        "ldap": TrinoAuthenticationMethod.LDAP,
 827        "kerberos": TrinoAuthenticationMethod.KERBEROS,
 828        "jwt": TrinoAuthenticationMethod.JWT,
 829        "certificate": TrinoAuthenticationMethod.CERTIFICATE,
 830        "oauth": TrinoAuthenticationMethod.OAUTH,
 831        "oauth_console": TrinoAuthenticationMethod.OAUTH,
 832    }
 833
 834    type: t.Literal["trino"] = "trino"
 835    host: str
 836    database: str
 837    schema_: str = Field(alias="schema")
 838    port: int = 443
 839    method: str
 840    user: t.Optional[str] = None
 841
 842    threads: int = 1
 843    roles: t.Optional[t.Dict[str, str]] = None
 844    session_properties: t.Optional[t.Dict[str, str]] = None
 845    retries: int = 3
 846    timezone: t.Optional[str] = None
 847    http_headers: t.Optional[t.Dict[str, str]] = None
 848    http_scheme: t.Optional[str] = None
 849    prepared_statements_enabled: bool = True  # not used by SQLMesh
 850
 851    # ldap authentication
 852    password: t.Optional[str] = None
 853    impersonation_user: t.Optional[str] = None
 854
 855    # kerberos authentication
 856    keytab: t.Optional[str] = None
 857    krb5_config: t.Optional[str] = None
 858    principal: t.Optional[str] = None
 859    service_name: str = "trino"
 860    hostname_override: t.Optional[str] = None
 861    mutual_authentication: bool = False
 862    force_preemptive: bool = False
 863    sanitize_mutual_error_response: bool = True
 864    delegate: bool = False
 865
 866    # jwt authentication
 867    jwt_token: t.Optional[str] = None
 868
 869    # certificate authentication
 870    client_certificate: t.Optional[str] = None
 871    client_private_key: t.Optional[str] = None
 872    cert: t.Optional[str] = None
 873
 874    def default_incremental_strategy(self, kind: IncrementalKind) -> str:
 875        return "append"
 876
 877    @classproperty
 878    def relation_class(cls) -> t.Type[BaseRelation]:
 879        from dbt.adapters.trino.relation import TrinoRelation
 880
 881        return TrinoRelation
 882
 883    @classproperty
 884    def column_class(cls) -> t.Type[Column]:
 885        from dbt.adapters.trino.column import TrinoColumn
 886
 887        return TrinoColumn
 888
 889    @with_schema_differ_overrides
 890    def to_sqlmesh(self, **kwargs: t.Any) -> ConnectionConfig:
 891        return TrinoConnectionConfig(
 892            method=self._method_to_auth_enum[self.method],
 893            host=self.host,
 894            user=self.user,
 895            catalog=self.database,
 896            port=self.port,
 897            http_scheme=self.http_scheme,
 898            roles=self.roles,
 899            http_headers=self.http_headers,
 900            session_properties=self.session_properties,
 901            retries=self.retries,
 902            timezone=self.timezone,
 903            password=self.password,
 904            impersonation_user=self.impersonation_user,
 905            keytab=self.keytab,
 906            krb5_config=self.krb5_config,
 907            principal=self.principal,
 908            service_name=self.service_name,
 909            hostname_override=self.hostname_override,
 910            mutual_authentication=self.mutual_authentication,
 911            force_preemptive=self.force_preemptive,
 912            sanitize_mutual_error_response=self.sanitize_mutual_error_response,
 913            delegate=self.delegate,
 914            jwt_token=self.jwt_token,
 915            client_certificate=self.client_certificate,
 916            client_private_key=self.client_private_key,
 917            cert=self.cert,
 918            concurrent_tasks=self.threads,
 919            **kwargs,
 920        )
 921
 922
 923class ClickhouseConfig(TargetConfig):
 924    """
 925    Project connection and operational configuration for the Clickhouse target
 926
 927    Args:
 928      host: [localhost]
 929      user: [default] # User for all database operations
 930      password: [<empty string>] # Password for the user
 931      secure: [False] # Use TLS (native protocol) or HTTPS (http protocol)
 932      port: [8123]  # If not set, defaults to 8123, 8443 depending on the secure and driver settings
 933      connect_timeout: [10] # Timeout in seconds to establish a connection to ClickHouse
 934      send_receive_timeout: [300] # Timeout in seconds to receive data from the ClickHouse server
 935      verify: [True] # Validate TLS certificate if using TLS/SSL
 936      cluster: [<empty string>] # If set, certain DDL/table operations will be executed with the `ON CLUSTER` clause using this cluster.
 937      custom_settings: [{}] # A dictionary/mapping of custom ClickHouse settings for the connection - default is empty.
 938      schema: [default] # ClickHouse database for dbt models, not used by SQLMesh
 939      driver: [http] # http or native.  If not set this will be autodetermined based on port setting, not used by SQLMesh
 940      retries: [1] # Number of times to retry a "retriable" database exception (such as a 503 'Service Unavailable' error), not used by SQLMesh
 941      compression: [<empty string>] # Use gzip compression if truthy (http), or compression type for a native connection, not used by SQLMesh
 942      cluster_mode: [False] # Use specific settings designed to improve operation on Replicated databases (recommended for ClickHouse Cloud), not used by SQLMesh
 943      use_lw_deletes: [False] # Use the strategy `delete+insert` as the default incremental strategy, not used by SQLMesh
 944      check_exchange: [True] # Validate that clickhouse support the atomic EXCHANGE TABLES command. Not used by SQLMesh.
 945      local_suffix: [_local] # Table suffix of local tables on shards for distributed materializations, not used by SQLMesh
 946      local_db_prefix: [<empty string>] # Database prefix of local tables on shards for distributed materializations, not used by SQLMesh
 947      allow_automatic_deduplication: [False] # Enable ClickHouse automatic deduplication for Replicated tables, not used by SQLMesh
 948      tcp_keepalive: [False] # Native client only, specify TCP keepalive configuration. Specify custom keepalive settings as [idle_time_sec, interval_sec, probes], not used by SQLMesh
 949      sync_request_timeout: [5] # Timeout for server ping, not used by SQLMesh
 950      compress_block_size: [1048576] # Compression block size if compression is enabled, not used by SQLMesh
 951    """
 952
 953    host: str = "localhost"
 954    user: str = Field(default="default", alias="username")
 955    password: str = ""
 956    port: t.Optional[int] = None
 957    cluster: t.Optional[str] = None
 958    schema_: str = Field(default="default", alias="schema")
 959    connect_timeout: int = 10
 960    send_receive_timeout: int = 300
 961    verify: bool = True
 962    compression: str = ""
 963    custom_settings: t.Optional[t.Dict[str, t.Any]] = None
 964
 965    # Not used by SQLMesh
 966    driver: t.Optional[str] = None
 967    secure: bool = False
 968    retries: int = 1
 969    database_engine: t.Optional[str] = None
 970    cluster_mode: bool = False
 971    sync_request_timeout: int = 5
 972    compress_block_size: int = 1048576
 973    check_exchange: bool = True
 974    use_lw_deletes: bool = False
 975    allow_automatic_deduplication: bool = False
 976    tcp_keepalive: t.Union[bool, t.Tuple[int, ...], t.List[int]] = False
 977    database: str = ""
 978    local_suffix: str = "local"
 979    local_db_prefix: str = ""
 980
 981    type: t.Literal["clickhouse"] = "clickhouse"
 982
 983    def default_incremental_strategy(self, kind: IncrementalKind) -> str:
 984        # dbt-clickhouse name for temp table swap. That is sqlmesh's default
 985        #   strategy so doesn't require special handling during conversion.
 986        return "legacy"
 987
 988    @classproperty
 989    def relation_class(cls) -> t.Type[BaseRelation]:
 990        from dbt.adapters.clickhouse.relation import ClickHouseRelation
 991
 992        return ClickHouseRelation
 993
 994    @classproperty
 995    def column_class(cls) -> t.Type[Column]:
 996        from dbt.adapters.clickhouse.column import ClickHouseColumn
 997
 998        return ClickHouseColumn
 999
1000    @with_schema_differ_overrides
1001    def to_sqlmesh(self, **kwargs: t.Any) -> ConnectionConfig:
1002        return ClickhouseConnectionConfig(
1003            host=self.host,
1004            username=self.user,
1005            password=self.password,
1006            port=self.port,
1007            cluster=self.cluster,
1008            connect_timeout=self.connect_timeout,
1009            send_receive_timeout=self.send_receive_timeout,
1010            verify=self.verify,
1011            compression_method=self.compression,
1012            connection_settings=self.custom_settings,
1013            **kwargs,
1014        )
1015
1016
1017class AthenaConfig(TargetConfig):
1018    """
1019    Project connection and operational configuration for the Athena target.
1020
1021    Args:
1022        s3_staging_dir: S3 location to store Athena query results and metadata
1023        s3_data_dir: Prefix for storing tables, if different from the connection's s3_staging_dir
1024        s3_data_naming: How to generate table paths in s3_data_dir
1025        s3_tmp_table_dir: Prefix for storing temporary tables, if different from the connection's s3_data_dir
1026        region_name: AWS region of your Athena instance
1027        schema: Specify the schema (Athena database) to build models into (lowercase only)
1028        database: Specify the database (Data catalog) to build models into (lowercase only)
1029        poll_interval: Interval in seconds to use for polling the status of query results in Athena
1030        debug_query_state: Flag if debug message with Athena query state is needed
1031        aws_access_key_id: Access key ID of the user performing requests
1032        aws_secret_access_key: Secret access key of the user performing requests
1033        aws_profile_name: Profile to use from your AWS shared credentials file
1034        work_group: Identifier of Athena workgroup
1035        skip_workgroup_check: Indicates if the WorkGroup check (additional AWS call) can be skipped
1036        num_retries: Number of times to retry a failing query
1037        num_boto3_retries: Number of times to retry boto3 requests (e.g. deleting S3 files for materialized tables)
1038        num_iceberg_retries: Number of times to retry iceberg commit queries to fix ICEBERG_COMMIT_ERROR
1039        spark_work_group: Identifier of Athena Spark workgroup for running Python models
1040        seed_s3_upload_args: Dictionary containing boto3 ExtraArgs when uploading to S3
1041        lf_tags_database: Default LF tags for new database if it's created by dbt
1042    """
1043
1044    type: t.Literal["athena"] = "athena"
1045    threads: int = 4
1046
1047    s3_staging_dir: t.Optional[str] = None
1048    s3_data_dir: t.Optional[str] = None
1049    s3_data_naming: t.Optional[str] = None
1050    s3_tmp_table_dir: t.Optional[str] = None
1051    poll_interval: t.Optional[int] = None
1052    debug_query_state: bool = False
1053    work_group: t.Optional[str] = None
1054    skip_workgroup_check: t.Optional[bool] = None
1055    spark_work_group: t.Optional[str] = None
1056
1057    aws_access_key_id: t.Optional[str] = None
1058    aws_secret_access_key: t.Optional[str] = None
1059    aws_profile_name: t.Optional[str] = None
1060    region_name: t.Optional[str] = None
1061
1062    num_retries: t.Optional[int] = None
1063    num_boto3_retries: t.Optional[int] = None
1064    num_iceberg_retries: t.Optional[int] = None
1065
1066    seed_s3_upload_args: t.Dict[str, str] = {}
1067    lf_tags_database: t.Dict[str, str] = {}
1068
1069    @classproperty
1070    def relation_class(cls) -> t.Type[BaseRelation]:
1071        from dbt.adapters.athena.relation import AthenaRelation
1072
1073        return AthenaRelation
1074
1075    @classproperty
1076    def column_class(cls) -> t.Type[Column]:
1077        from dbt.adapters.athena.column import AthenaColumn
1078
1079        return AthenaColumn
1080
1081    def default_incremental_strategy(self, kind: IncrementalKind) -> str:
1082        return "insert_overwrite"
1083
1084    @with_schema_differ_overrides
1085    def to_sqlmesh(self, **kwargs: t.Any) -> ConnectionConfig:
1086        return AthenaConnectionConfig(
1087            type="athena",
1088            aws_access_key_id=self.aws_access_key_id,
1089            aws_secret_access_key=self.aws_secret_access_key,
1090            region_name=self.region_name,
1091            work_group=self.work_group,
1092            s3_staging_dir=self.s3_staging_dir,
1093            s3_warehouse_location=self.s3_data_dir,
1094            schema_name=self.schema_,
1095            catalog_name=self.database,
1096            concurrent_tasks=self.threads,
1097            **kwargs,
1098        )
1099
1100
1101TARGET_TYPE_TO_CONFIG_CLASS: t.Dict[str, t.Type[TargetConfig]] = {
1102    "databricks": DatabricksConfig,
1103    "duckdb": DuckDbConfig,
1104    "postgres": PostgresConfig,
1105    "redshift": RedshiftConfig,
1106    "snowflake": SnowflakeConfig,
1107    "bigquery": BigQueryConfig,
1108    "sqlserver": MSSQLConfig,
1109    "tsql": MSSQLConfig,
1110    "trino": TrinoConfig,
1111    "athena": AthenaConfig,
1112    "clickhouse": ClickhouseConfig,
1113}