Edit on GitHub

sqlmesh.core.config.root 

  1from __future__ import annotations
  2
  3import pickle
  4import re
  5import typing as t
  6import zlib
  7
  8from pydantic import Field
  9from pydantic.functional_validators import BeforeValidator
 10from sqlglot import exp
 11from sqlglot.helper import first
 12from sqlglot.optimizer.normalize_identifiers import normalize_identifiers
 13
 14from sqlmesh.cicd.config import CICDBotConfig
 15from sqlmesh.core import constants as c
 16from sqlmesh.core.console import get_console
 17from sqlmesh.core.config.common import (
 18    EnvironmentSuffixTarget,
 19    TableNamingConvention,
 20    VirtualEnvironmentMode,
 21)
 22from sqlmesh.core.config.base import BaseConfig, UpdateStrategy
 23from sqlmesh.core.config.common import variables_validator, compile_regex_mapping
 24from sqlmesh.core.config.connection import (
 25    ConnectionConfig,
 26    DuckDBConnectionConfig,
 27    SerializableConnectionConfig,
 28    connection_config_validator,
 29)
 30from sqlmesh.core.config.format import FormatConfig
 31from sqlmesh.core.config.gateway import GatewayConfig
 32from sqlmesh.core.config.janitor import JanitorConfig
 33from sqlmesh.core.config.migration import MigrationConfig
 34from sqlmesh.core.config.model import ModelDefaultsConfig
 35from sqlmesh.core.config.naming import NameInferenceConfig as NameInferenceConfig
 36from sqlmesh.core.config.linter import LinterConfig as LinterConfig
 37from sqlmesh.core.config.plan import PlanConfig
 38from sqlmesh.core.config.run import RunConfig
 39from sqlmesh.core.config.dbt import DbtConfig
 40from sqlmesh.core.config.scheduler import (
 41    BuiltInSchedulerConfig,
 42    SchedulerConfig,
 43    scheduler_config_validator,
 44)
 45from sqlmesh.core.config.ui import UIConfig
 46from sqlmesh.core.loader import Loader, SqlMeshLoader
 47from sqlmesh.core.notification_target import NotificationTarget
 48from sqlmesh.core.user import User
 49from sqlmesh.utils.date import to_timestamp, now
 50from sqlmesh.utils.errors import ConfigError
 51from sqlmesh.utils.pydantic import model_validator
 52
 53
 54def validate_no_past_ttl(v: str) -> str:
 55    current_time = now()
 56    if to_timestamp(v, relative_base=current_time) < to_timestamp(current_time):
 57        raise ValueError(
 58            f"TTL '{v}' is in the past. Please specify a relative time in the future. Ex: `in 1 week` instead of `1 week`."
 59        )
 60    return v
 61
 62
 63def gateways_ensure_dict(value: t.Dict[str, t.Any]) -> t.Dict[str, t.Any]:
 64    try:
 65        if not isinstance(value, GatewayConfig):
 66            GatewayConfig.parse_obj(value)
 67        return {"": value}
 68    except Exception:
 69        # Normalize all gateway keys to lowercase for case-insensitive matching
 70        if isinstance(value, dict):
 71            return {k.lower(): v for k, v in value.items()}
 72        return value
 73
 74
 75def validate_regex_key_dict(value: t.Dict[str | re.Pattern, t.Any]) -> t.Dict[re.Pattern, t.Any]:
 76    return compile_regex_mapping(value)
 77
 78
 79def _canonicalize(obj: object) -> object:
 80    """Recursively convert an object into a canonical, order-stable form for hashing.
 81
 82    ``set``/``frozenset`` iteration order is not stable across Python processes, so
 83    pickling them directly yields non-deterministic bytes. That makes any hash derived
 84    from the pickle (e.g. ``Config.fingerprint``) change run-to-run, which silently
 85    invalidates on-disk caches keyed by the fingerprint. Sorting set members into a
 86    list restores determinism while preserving contents.
 87    """
 88    if isinstance(obj, (set, frozenset)):
 89        return sorted(map(_canonicalize, obj))  # type: ignore[type-var]
 90    if isinstance(obj, dict):
 91        return {k: _canonicalize(v) for k, v in obj.items()}
 92    if isinstance(obj, (list, tuple)):
 93        return type(obj)(map(_canonicalize, obj))
 94    return obj
 95
 96
 97if t.TYPE_CHECKING:
 98    from sqlmesh.core._typing import Self
 99
100    NoPastTTLString = str
101    GatewayDict = t.Dict[str, GatewayConfig]
102    RegexKeyDict = t.Dict[re.Pattern, str]
103else:
104    NoPastTTLString = t.Annotated[str, BeforeValidator(validate_no_past_ttl)]
105    GatewayDict = t.Annotated[t.Dict[str, GatewayConfig], BeforeValidator(gateways_ensure_dict)]
106    RegexKeyDict = t.Annotated[t.Dict[re.Pattern, str], BeforeValidator(validate_regex_key_dict)]
107
108
109class Config(BaseConfig):
110    """An object used by a Context to configure your SQLMesh project.
111
112    Args:
113        gateways: Supported gateways and their configurations. Key represents a unique name of a gateway.
114        default_connection: The default connection to use if one is not specified in a gateway.
115        default_test_connection: The default connection to use for tests if one is not specified in a gateway.
116        default_scheduler: The default scheduler configuration to use if one is not specified in a gateway.
117        default_gateway: The default gateway.
118        notification_targets: The notification targets to use.
119        project: The project name of this config. Used for multi-repo setups.
120        snapshot_ttl: The period of time that a model snapshot that is not a part of any environment should exist before being deleted.
121        environment_ttl: The period of time that a development environment should exist before being deleted.
122        ignore_patterns: Files that match glob patterns specified in this list are ignored when scanning the project folder.
123        time_column_format: The default format to use for all model time columns. Defaults to %Y-%m-%d.
124            This time format uses python format codes. https://docs.python.org/3/library/datetime.html#strftime-and-strptime-format-codes.
125        users: A list of users that can be used for approvals/notifications.
126        username: Name of a single user who should receive approvals/notification, instead of all users in the `users` list.
127        pinned_environments: A list of development environment names that should not be deleted by the janitor task.
128        loader: Loader class used for loading project files.
129        loader_kwargs: Key-value arguments to pass to the loader instance.
130        env_vars: A dictionary of environmental variable names and values.
131        model_defaults: Default values for model definitions.
132        physical_schema_mapping: A mapping from regular expressions to names of schemas in which physical tables for corresponding models will be placed.
133        environment_suffix_target: Indicates whether to append the environment name to the schema or table name.
134        physical_table_naming_convention: Indicates how tables should be named at the physical layer
135        virtual_environment_mode: Indicates how environments should be handled.
136        gateway_managed_virtual_layer: Whether the models' views in the virtual layer are created by the model-specific gateway rather than the default gateway.
137        infer_python_dependencies: Whether to statically analyze Python code to automatically infer Python package requirements.
138        environment_catalog_mapping: A mapping from regular expressions to catalog names. The catalog name is used to determine the target catalog for a given environment.
139        default_target_environment: The name of the environment that will be the default target for the `sqlmesh plan` and `sqlmesh run` commands.
140        log_limit: The default number of logs to keep.
141        format: The formatting options for SQL code.
142        ui: The UI configuration for SQLMesh.
143        plan: The plan configuration.
144        migration: The migration configuration.
145        variables: A dictionary of variables that can be used in models / macros.
146        disable_anonymized_analytics: Whether to disable the anonymized analytics collection.
147        before_all: SQL statements or macros to be executed at the start of the `sqlmesh plan` and `sqlmesh run` commands.
148        after_all: SQL statements or macros to be executed at the end of the `sqlmesh plan` and `sqlmesh run` commands.
149        cache_dir: The directory to store the SQLMesh cache. Defaults to .cache in the project folder.
150    """
151
152    gateways: GatewayDict = {"": GatewayConfig()}
153    default_connection: t.Optional[SerializableConnectionConfig] = None
154    default_test_connection_: t.Optional[SerializableConnectionConfig] = Field(
155        default=None, alias="default_test_connection"
156    )
157    default_scheduler: SchedulerConfig = BuiltInSchedulerConfig()
158    default_gateway: str = ""
159    notification_targets: t.List[NotificationTarget] = []
160    project: str = ""
161    snapshot_ttl: NoPastTTLString = c.DEFAULT_SNAPSHOT_TTL
162    environment_ttl: t.Optional[NoPastTTLString] = c.DEFAULT_ENVIRONMENT_TTL
163    ignore_patterns: t.List[str] = c.IGNORE_PATTERNS
164    time_column_format: str = c.DEFAULT_TIME_COLUMN_FORMAT
165    users: t.List[User] = []
166    model_defaults: ModelDefaultsConfig = ModelDefaultsConfig()
167    pinned_environments: t.Set[str] = set()
168    loader: t.Type[Loader] = SqlMeshLoader
169    loader_kwargs: t.Dict[str, t.Any] = {}
170    env_vars: t.Dict[str, str] = {}
171    username: str = ""
172    physical_schema_mapping: RegexKeyDict = {}
173    environment_suffix_target: EnvironmentSuffixTarget = EnvironmentSuffixTarget.default
174    physical_table_naming_convention: TableNamingConvention = TableNamingConvention.default
175    virtual_environment_mode: VirtualEnvironmentMode = VirtualEnvironmentMode.default
176    gateway_managed_virtual_layer: bool = False
177    infer_python_dependencies: bool = True
178    environment_catalog_mapping: RegexKeyDict = {}
179    default_target_environment: str = c.PROD
180    log_limit: int = c.DEFAULT_LOG_LIMIT
181    cicd_bot: t.Optional[CICDBotConfig] = None
182    run: RunConfig = RunConfig()
183    format: FormatConfig = FormatConfig()
184    ui: UIConfig = UIConfig()
185    plan: PlanConfig = PlanConfig()
186    migration: MigrationConfig = MigrationConfig()
187    model_naming: NameInferenceConfig = NameInferenceConfig()
188    variables: t.Dict[str, t.Any] = {}
189    disable_anonymized_analytics: bool = False
190    before_all: t.Optional[t.List[str]] = None
191    after_all: t.Optional[t.List[str]] = None
192    linter: LinterConfig = LinterConfig()
193    janitor: JanitorConfig = JanitorConfig()
194    cache_dir: t.Optional[str] = None
195    dbt: t.Optional[DbtConfig] = None
196
197    _FIELD_UPDATE_STRATEGY: t.ClassVar[t.Dict[str, UpdateStrategy]] = {
198        "gateways": UpdateStrategy.NESTED_UPDATE,
199        "notification_targets": UpdateStrategy.EXTEND,
200        "ignore_patterns": UpdateStrategy.EXTEND,
201        "users": UpdateStrategy.EXTEND,
202        "model_defaults": UpdateStrategy.NESTED_UPDATE,
203        "auto_categorize_changes": UpdateStrategy.NESTED_UPDATE,
204        "pinned_environments": UpdateStrategy.EXTEND,
205        "physical_schema_override": UpdateStrategy.KEY_UPDATE,
206        "run": UpdateStrategy.NESTED_UPDATE,
207        "format": UpdateStrategy.NESTED_UPDATE,
208        "ui": UpdateStrategy.NESTED_UPDATE,
209        "loader_kwargs": UpdateStrategy.KEY_UPDATE,
210        "plan": UpdateStrategy.NESTED_UPDATE,
211        "before_all": UpdateStrategy.EXTEND,
212        "after_all": UpdateStrategy.EXTEND,
213        "linter": UpdateStrategy.NESTED_UPDATE,
214        "dbt": UpdateStrategy.NESTED_UPDATE,
215    }
216
217    _connection_config_validator = connection_config_validator
218    _scheduler_config_validator = scheduler_config_validator  # type: ignore
219    _variables_validator = variables_validator
220
221    @model_validator(mode="before")
222    def _normalize_and_validate_fields(cls, data: t.Any) -> t.Any:
223        if not isinstance(data, dict):
224            return data
225
226        if "gateways" not in data and "gateway" in data:
227            data["gateways"] = data.pop("gateway")
228
229        for plan_deprecated in ("auto_categorize_changes", "include_unmodified"):
230            if plan_deprecated in data:
231                raise ConfigError(
232                    f"The `{plan_deprecated}` config is deprecated. Please use the `plan.{plan_deprecated}` config instead."
233                )
234
235        if "physical_schema_override" in data:
236            get_console().log_warning(
237                "`physical_schema_override` is deprecated. Please use `physical_schema_mapping` instead."
238            )
239
240            if "physical_schema_mapping" in data:
241                raise ConfigError(
242                    "Only one of `physical_schema_override` and `physical_schema_mapping` can be specified."
243                )
244
245            physical_schema_override: t.Dict[str, str] = data.pop("physical_schema_override")
246            # translate physical_schema_override to physical_schema_mapping
247            data["physical_schema_mapping"] = {
248                f"^{k}$": v for k, v in physical_schema_override.items()
249            }
250
251        return data
252
253    @model_validator(mode="after")
254    def _normalize_fields_after(self) -> Self:
255        dialect = self.model_defaults.dialect
256
257        def _normalize_identifiers(key: str) -> None:
258            setattr(
259                self,
260                key,
261                {
262                    k: normalize_identifiers(v, dialect=dialect).name
263                    for k, v in getattr(self, key, {}).items()
264                },
265            )
266
267        if (
268            self.environment_suffix_target == EnvironmentSuffixTarget.CATALOG
269            and self.environment_catalog_mapping
270        ):
271            raise ConfigError(
272                f"'environment_suffix_target: catalog' is mutually exclusive with 'environment_catalog_mapping'.\n"
273                "Please specify one or the other"
274            )
275
276        if self.plan.use_finalized_state and not self.virtual_environment_mode.is_full:
277            raise ConfigError(
278                "Using the finalized state is only supported when `virtual_environment_mode` is set to `full`."
279            )
280
281        if self.environment_catalog_mapping:
282            _normalize_identifiers("environment_catalog_mapping")
283        if self.physical_schema_mapping:
284            _normalize_identifiers("physical_schema_mapping")
285
286        return self
287
288    @model_validator(mode="after")
289    def _inherit_project_config_in_cicd_bot(self) -> Self:
290        if self.cicd_bot:
291            # inherit the project-level settings into the CICD bot if they have not been explicitly overridden
292            if self.cicd_bot.auto_categorize_changes_ is None:
293                self.cicd_bot.auto_categorize_changes_ = self.plan.auto_categorize_changes
294
295            if self.cicd_bot.pr_include_unmodified_ is None:
296                self.cicd_bot.pr_include_unmodified_ = self.plan.include_unmodified
297
298        return self
299
300    def get_default_test_connection(
301        self,
302        default_catalog: t.Optional[str] = None,
303        default_catalog_dialect: t.Optional[str] = None,
304    ) -> ConnectionConfig:
305        return self.default_test_connection_ or DuckDBConnectionConfig(
306            catalogs=(
307                None
308                if default_catalog is None
309                else {
310                    # transpile catalog name from main connection dialect to DuckDB
311                    exp.parse_identifier(default_catalog, dialect=default_catalog_dialect).sql(
312                        dialect="duckdb"
313                    ): ":memory:"
314                }
315            )
316        )
317
318    def get_gateway(self, name: t.Optional[str] = None) -> GatewayConfig:
319        if isinstance(self.gateways, dict):
320            if name is None:
321                if self.default_gateway:
322                    # Normalize default_gateway name to lowercase for lookup
323                    default_key = self.default_gateway.lower()
324                    if default_key not in self.gateways:
325                        raise ConfigError(f"Missing gateway with name '{self.default_gateway}'")
326                    return self.gateways[default_key]
327
328                if "" in self.gateways:
329                    return self.gateways[""]
330
331                return first(self.gateways.values())
332
333            # Normalize lookup name to lowercase since gateway keys are already lowercase
334            lookup_key = name.lower()
335            if lookup_key not in self.gateways:
336                raise ConfigError(f"Missing gateway with name '{name}'.")
337
338            return self.gateways[lookup_key]
339        if name is not None:
340            raise ConfigError("Gateway name is not supported when only one gateway is configured.")
341        return self.gateways
342
343    def get_connection(self, gateway_name: t.Optional[str] = None) -> ConnectionConfig:
344        connection = self.get_gateway(gateway_name).connection or self.default_connection
345        if connection is None:
346            msg = f" for gateway '{gateway_name}'" if gateway_name else ""
347            raise ConfigError(f"No connection configured{msg}.")
348        return connection
349
350    def get_state_connection(
351        self, gateway_name: t.Optional[str] = None
352    ) -> t.Optional[ConnectionConfig]:
353        return self.get_gateway(gateway_name).state_connection
354
355    def get_test_connection(
356        self,
357        gateway_name: t.Optional[str] = None,
358        default_catalog: t.Optional[str] = None,
359        default_catalog_dialect: t.Optional[str] = None,
360    ) -> ConnectionConfig:
361        return self.get_gateway(gateway_name).test_connection or self.get_default_test_connection(
362            default_catalog=default_catalog, default_catalog_dialect=default_catalog_dialect
363        )
364
365    def get_scheduler(self, gateway_name: t.Optional[str] = None) -> SchedulerConfig:
366        return self.get_gateway(gateway_name).scheduler or self.default_scheduler
367
368    def get_state_schema(self, gateway_name: t.Optional[str] = None) -> t.Optional[str]:
369        return self.get_gateway(gateway_name).state_schema
370
371    @property
372    def default_gateway_name(self) -> str:
373        if self.default_gateway:
374            return self.default_gateway
375        if "" in self.gateways:
376            return ""
377        return first(self.gateways)
378
379    @property
380    def dialect(self) -> t.Optional[str]:
381        return self.model_defaults.dialect
382
383    @property
384    def fingerprint(self) -> str:
385        return str(
386            zlib.crc32(
387                pickle.dumps(_canonicalize(self.dict(exclude={"loader", "notification_targets"})))
388            )
389        )
def validate_no_past_ttl(v: str) -> str: 
55def validate_no_past_ttl(v: str) -> str:
56    current_time = now()
57    if to_timestamp(v, relative_base=current_time) < to_timestamp(current_time):
58        raise ValueError(
59            f"TTL '{v}' is in the past. Please specify a relative time in the future. Ex: `in 1 week` instead of `1 week`."
60        )
61    return v
def gateways_ensure_dict(value: Dict[str, Any]) -> Dict[str, Any]: 
64def gateways_ensure_dict(value: t.Dict[str, t.Any]) -> t.Dict[str, t.Any]:
65    try:
66        if not isinstance(value, GatewayConfig):
67            GatewayConfig.parse_obj(value)
68        return {"": value}
69    except Exception:
70        # Normalize all gateway keys to lowercase for case-insensitive matching
71        if isinstance(value, dict):
72            return {k.lower(): v for k, v in value.items()}
73        return value
def validate_regex_key_dict(value: Dict[str | re.Pattern, Any]) -> Dict[re.Pattern, Any]: 
76def validate_regex_key_dict(value: t.Dict[str | re.Pattern, t.Any]) -> t.Dict[re.Pattern, t.Any]:
77    return compile_regex_mapping(value)
class Config(sqlmesh.core.config.base.BaseConfig): 
110class Config(BaseConfig):
111    """An object used by a Context to configure your SQLMesh project.
112
113    Args:
114        gateways: Supported gateways and their configurations. Key represents a unique name of a gateway.
115        default_connection: The default connection to use if one is not specified in a gateway.
116        default_test_connection: The default connection to use for tests if one is not specified in a gateway.
117        default_scheduler: The default scheduler configuration to use if one is not specified in a gateway.
118        default_gateway: The default gateway.
119        notification_targets: The notification targets to use.
120        project: The project name of this config. Used for multi-repo setups.
121        snapshot_ttl: The period of time that a model snapshot that is not a part of any environment should exist before being deleted.
122        environment_ttl: The period of time that a development environment should exist before being deleted.
123        ignore_patterns: Files that match glob patterns specified in this list are ignored when scanning the project folder.
124        time_column_format: The default format to use for all model time columns. Defaults to %Y-%m-%d.
125            This time format uses python format codes. https://docs.python.org/3/library/datetime.html#strftime-and-strptime-format-codes.
126        users: A list of users that can be used for approvals/notifications.
127        username: Name of a single user who should receive approvals/notification, instead of all users in the `users` list.
128        pinned_environments: A list of development environment names that should not be deleted by the janitor task.
129        loader: Loader class used for loading project files.
130        loader_kwargs: Key-value arguments to pass to the loader instance.
131        env_vars: A dictionary of environmental variable names and values.
132        model_defaults: Default values for model definitions.
133        physical_schema_mapping: A mapping from regular expressions to names of schemas in which physical tables for corresponding models will be placed.
134        environment_suffix_target: Indicates whether to append the environment name to the schema or table name.
135        physical_table_naming_convention: Indicates how tables should be named at the physical layer
136        virtual_environment_mode: Indicates how environments should be handled.
137        gateway_managed_virtual_layer: Whether the models' views in the virtual layer are created by the model-specific gateway rather than the default gateway.
138        infer_python_dependencies: Whether to statically analyze Python code to automatically infer Python package requirements.
139        environment_catalog_mapping: A mapping from regular expressions to catalog names. The catalog name is used to determine the target catalog for a given environment.
140        default_target_environment: The name of the environment that will be the default target for the `sqlmesh plan` and `sqlmesh run` commands.
141        log_limit: The default number of logs to keep.
142        format: The formatting options for SQL code.
143        ui: The UI configuration for SQLMesh.
144        plan: The plan configuration.
145        migration: The migration configuration.
146        variables: A dictionary of variables that can be used in models / macros.
147        disable_anonymized_analytics: Whether to disable the anonymized analytics collection.
148        before_all: SQL statements or macros to be executed at the start of the `sqlmesh plan` and `sqlmesh run` commands.
149        after_all: SQL statements or macros to be executed at the end of the `sqlmesh plan` and `sqlmesh run` commands.
150        cache_dir: The directory to store the SQLMesh cache. Defaults to .cache in the project folder.
151    """
152
153    gateways: GatewayDict = {"": GatewayConfig()}
154    default_connection: t.Optional[SerializableConnectionConfig] = None
155    default_test_connection_: t.Optional[SerializableConnectionConfig] = Field(
156        default=None, alias="default_test_connection"
157    )
158    default_scheduler: SchedulerConfig = BuiltInSchedulerConfig()
159    default_gateway: str = ""
160    notification_targets: t.List[NotificationTarget] = []
161    project: str = ""
162    snapshot_ttl: NoPastTTLString = c.DEFAULT_SNAPSHOT_TTL
163    environment_ttl: t.Optional[NoPastTTLString] = c.DEFAULT_ENVIRONMENT_TTL
164    ignore_patterns: t.List[str] = c.IGNORE_PATTERNS
165    time_column_format: str = c.DEFAULT_TIME_COLUMN_FORMAT
166    users: t.List[User] = []
167    model_defaults: ModelDefaultsConfig = ModelDefaultsConfig()
168    pinned_environments: t.Set[str] = set()
169    loader: t.Type[Loader] = SqlMeshLoader
170    loader_kwargs: t.Dict[str, t.Any] = {}
171    env_vars: t.Dict[str, str] = {}
172    username: str = ""
173    physical_schema_mapping: RegexKeyDict = {}
174    environment_suffix_target: EnvironmentSuffixTarget = EnvironmentSuffixTarget.default
175    physical_table_naming_convention: TableNamingConvention = TableNamingConvention.default
176    virtual_environment_mode: VirtualEnvironmentMode = VirtualEnvironmentMode.default
177    gateway_managed_virtual_layer: bool = False
178    infer_python_dependencies: bool = True
179    environment_catalog_mapping: RegexKeyDict = {}
180    default_target_environment: str = c.PROD
181    log_limit: int = c.DEFAULT_LOG_LIMIT
182    cicd_bot: t.Optional[CICDBotConfig] = None
183    run: RunConfig = RunConfig()
184    format: FormatConfig = FormatConfig()
185    ui: UIConfig = UIConfig()
186    plan: PlanConfig = PlanConfig()
187    migration: MigrationConfig = MigrationConfig()
188    model_naming: NameInferenceConfig = NameInferenceConfig()
189    variables: t.Dict[str, t.Any] = {}
190    disable_anonymized_analytics: bool = False
191    before_all: t.Optional[t.List[str]] = None
192    after_all: t.Optional[t.List[str]] = None
193    linter: LinterConfig = LinterConfig()
194    janitor: JanitorConfig = JanitorConfig()
195    cache_dir: t.Optional[str] = None
196    dbt: t.Optional[DbtConfig] = None
197
198    _FIELD_UPDATE_STRATEGY: t.ClassVar[t.Dict[str, UpdateStrategy]] = {
199        "gateways": UpdateStrategy.NESTED_UPDATE,
200        "notification_targets": UpdateStrategy.EXTEND,
201        "ignore_patterns": UpdateStrategy.EXTEND,
202        "users": UpdateStrategy.EXTEND,
203        "model_defaults": UpdateStrategy.NESTED_UPDATE,
204        "auto_categorize_changes": UpdateStrategy.NESTED_UPDATE,
205        "pinned_environments": UpdateStrategy.EXTEND,
206        "physical_schema_override": UpdateStrategy.KEY_UPDATE,
207        "run": UpdateStrategy.NESTED_UPDATE,
208        "format": UpdateStrategy.NESTED_UPDATE,
209        "ui": UpdateStrategy.NESTED_UPDATE,
210        "loader_kwargs": UpdateStrategy.KEY_UPDATE,
211        "plan": UpdateStrategy.NESTED_UPDATE,
212        "before_all": UpdateStrategy.EXTEND,
213        "after_all": UpdateStrategy.EXTEND,
214        "linter": UpdateStrategy.NESTED_UPDATE,
215        "dbt": UpdateStrategy.NESTED_UPDATE,
216    }
217
218    _connection_config_validator = connection_config_validator
219    _scheduler_config_validator = scheduler_config_validator  # type: ignore
220    _variables_validator = variables_validator
221
222    @model_validator(mode="before")
223    def _normalize_and_validate_fields(cls, data: t.Any) -> t.Any:
224        if not isinstance(data, dict):
225            return data
226
227        if "gateways" not in data and "gateway" in data:
228            data["gateways"] = data.pop("gateway")
229
230        for plan_deprecated in ("auto_categorize_changes", "include_unmodified"):
231            if plan_deprecated in data:
232                raise ConfigError(
233                    f"The `{plan_deprecated}` config is deprecated. Please use the `plan.{plan_deprecated}` config instead."
234                )
235
236        if "physical_schema_override" in data:
237            get_console().log_warning(
238                "`physical_schema_override` is deprecated. Please use `physical_schema_mapping` instead."
239            )
240
241            if "physical_schema_mapping" in data:
242                raise ConfigError(
243                    "Only one of `physical_schema_override` and `physical_schema_mapping` can be specified."
244                )
245
246            physical_schema_override: t.Dict[str, str] = data.pop("physical_schema_override")
247            # translate physical_schema_override to physical_schema_mapping
248            data["physical_schema_mapping"] = {
249                f"^{k}$": v for k, v in physical_schema_override.items()
250            }
251
252        return data
253
254    @model_validator(mode="after")
255    def _normalize_fields_after(self) -> Self:
256        dialect = self.model_defaults.dialect
257
258        def _normalize_identifiers(key: str) -> None:
259            setattr(
260                self,
261                key,
262                {
263                    k: normalize_identifiers(v, dialect=dialect).name
264                    for k, v in getattr(self, key, {}).items()
265                },
266            )
267
268        if (
269            self.environment_suffix_target == EnvironmentSuffixTarget.CATALOG
270            and self.environment_catalog_mapping
271        ):
272            raise ConfigError(
273                f"'environment_suffix_target: catalog' is mutually exclusive with 'environment_catalog_mapping'.\n"
274                "Please specify one or the other"
275            )
276
277        if self.plan.use_finalized_state and not self.virtual_environment_mode.is_full:
278            raise ConfigError(
279                "Using the finalized state is only supported when `virtual_environment_mode` is set to `full`."
280            )
281
282        if self.environment_catalog_mapping:
283            _normalize_identifiers("environment_catalog_mapping")
284        if self.physical_schema_mapping:
285            _normalize_identifiers("physical_schema_mapping")
286
287        return self
288
289    @model_validator(mode="after")
290    def _inherit_project_config_in_cicd_bot(self) -> Self:
291        if self.cicd_bot:
292            # inherit the project-level settings into the CICD bot if they have not been explicitly overridden
293            if self.cicd_bot.auto_categorize_changes_ is None:
294                self.cicd_bot.auto_categorize_changes_ = self.plan.auto_categorize_changes
295
296            if self.cicd_bot.pr_include_unmodified_ is None:
297                self.cicd_bot.pr_include_unmodified_ = self.plan.include_unmodified
298
299        return self
300
301    def get_default_test_connection(
302        self,
303        default_catalog: t.Optional[str] = None,
304        default_catalog_dialect: t.Optional[str] = None,
305    ) -> ConnectionConfig:
306        return self.default_test_connection_ or DuckDBConnectionConfig(
307            catalogs=(
308                None
309                if default_catalog is None
310                else {
311                    # transpile catalog name from main connection dialect to DuckDB
312                    exp.parse_identifier(default_catalog, dialect=default_catalog_dialect).sql(
313                        dialect="duckdb"
314                    ): ":memory:"
315                }
316            )
317        )
318
319    def get_gateway(self, name: t.Optional[str] = None) -> GatewayConfig:
320        if isinstance(self.gateways, dict):
321            if name is None:
322                if self.default_gateway:
323                    # Normalize default_gateway name to lowercase for lookup
324                    default_key = self.default_gateway.lower()
325                    if default_key not in self.gateways:
326                        raise ConfigError(f"Missing gateway with name '{self.default_gateway}'")
327                    return self.gateways[default_key]
328
329                if "" in self.gateways:
330                    return self.gateways[""]
331
332                return first(self.gateways.values())
333
334            # Normalize lookup name to lowercase since gateway keys are already lowercase
335            lookup_key = name.lower()
336            if lookup_key not in self.gateways:
337                raise ConfigError(f"Missing gateway with name '{name}'.")
338
339            return self.gateways[lookup_key]
340        if name is not None:
341            raise ConfigError("Gateway name is not supported when only one gateway is configured.")
342        return self.gateways
343
344    def get_connection(self, gateway_name: t.Optional[str] = None) -> ConnectionConfig:
345        connection = self.get_gateway(gateway_name).connection or self.default_connection
346        if connection is None:
347            msg = f" for gateway '{gateway_name}'" if gateway_name else ""
348            raise ConfigError(f"No connection configured{msg}.")
349        return connection
350
351    def get_state_connection(
352        self, gateway_name: t.Optional[str] = None
353    ) -> t.Optional[ConnectionConfig]:
354        return self.get_gateway(gateway_name).state_connection
355
356    def get_test_connection(
357        self,
358        gateway_name: t.Optional[str] = None,
359        default_catalog: t.Optional[str] = None,
360        default_catalog_dialect: t.Optional[str] = None,
361    ) -> ConnectionConfig:
362        return self.get_gateway(gateway_name).test_connection or self.get_default_test_connection(
363            default_catalog=default_catalog, default_catalog_dialect=default_catalog_dialect
364        )
365
366    def get_scheduler(self, gateway_name: t.Optional[str] = None) -> SchedulerConfig:
367        return self.get_gateway(gateway_name).scheduler or self.default_scheduler
368
369    def get_state_schema(self, gateway_name: t.Optional[str] = None) -> t.Optional[str]:
370        return self.get_gateway(gateway_name).state_schema
371
372    @property
373    def default_gateway_name(self) -> str:
374        if self.default_gateway:
375            return self.default_gateway
376        if "" in self.gateways:
377            return ""
378        return first(self.gateways)
379
380    @property
381    def dialect(self) -> t.Optional[str]:
382        return self.model_defaults.dialect
383
384    @property
385    def fingerprint(self) -> str:
386        return str(
387            zlib.crc32(
388                pickle.dumps(_canonicalize(self.dict(exclude={"loader", "notification_targets"})))
389            )
390        )

An object used by a Context to configure your SQLMesh project.

Arguments:
  • gateways: Supported gateways and their configurations. Key represents a unique name of a gateway.
  • default_connection: The default connection to use if one is not specified in a gateway.
  • default_test_connection: The default connection to use for tests if one is not specified in a gateway.
  • default_scheduler: The default scheduler configuration to use if one is not specified in a gateway.
  • default_gateway: The default gateway.
  • notification_targets: The notification targets to use.
  • project: The project name of this config. Used for multi-repo setups.
  • snapshot_ttl: The period of time that a model snapshot that is not a part of any environment should exist before being deleted.
  • environment_ttl: The period of time that a development environment should exist before being deleted.
  • ignore_patterns: Files that match glob patterns specified in this list are ignored when scanning the project folder.
  • time_column_format: The default format to use for all model time columns. Defaults to %Y-%m-%d. This time format uses python format codes. https://docs.python.org/3/library/datetime.html#strftime-and-strptime-format-codes.
  • users: A list of users that can be used for approvals/notifications.
  • username: Name of a single user who should receive approvals/notification, instead of all users in the users list.
  • pinned_environments: A list of development environment names that should not be deleted by the janitor task.
  • loader: Loader class used for loading project files.
  • loader_kwargs: Key-value arguments to pass to the loader instance.
  • env_vars: A dictionary of environmental variable names and values.
  • model_defaults: Default values for model definitions.
  • physical_schema_mapping: A mapping from regular expressions to names of schemas in which physical tables for corresponding models will be placed.
  • environment_suffix_target: Indicates whether to append the environment name to the schema or table name.
  • physical_table_naming_convention: Indicates how tables should be named at the physical layer
  • virtual_environment_mode: Indicates how environments should be handled.
  • gateway_managed_virtual_layer: Whether the models' views in the virtual layer are created by the model-specific gateway rather than the default gateway.
  • infer_python_dependencies: Whether to statically analyze Python code to automatically infer Python package requirements.
  • environment_catalog_mapping: A mapping from regular expressions to catalog names. The catalog name is used to determine the target catalog for a given environment.
  • default_target_environment: The name of the environment that will be the default target for the sqlmesh plan and sqlmesh run commands.
  • log_limit: The default number of logs to keep.
  • format: The formatting options for SQL code.
  • ui: The UI configuration for SQLMesh.
  • plan: The plan configuration.
  • migration: The migration configuration.
  • variables: A dictionary of variables that can be used in models / macros.
  • disable_anonymized_analytics: Whether to disable the anonymized analytics collection.
  • before_all: SQL statements or macros to be executed at the start of the sqlmesh plan and sqlmesh run commands.
  • after_all: SQL statements or macros to be executed at the end of the sqlmesh plan and sqlmesh run commands.
  • cache_dir: The directory to store the SQLMesh cache. Defaults to .cache in the project folder.
gateways: Annotated[Dict[str, sqlmesh.core.config.gateway.GatewayConfig], BeforeValidator(func=<function gateways_ensure_dict at 0x7a2abec8eb90>, json_schema_input_type=PydanticUndefined)]
default_connection: Optional[Annotated[sqlmesh.core.config.connection.ConnectionConfig, SerializeAsAny()]]
default_test_connection_: Optional[Annotated[sqlmesh.core.config.connection.ConnectionConfig, SerializeAsAny()]]
default_gateway: str
project: str
snapshot_ttl: typing.Annotated[str, BeforeValidator(func=<function validate_no_past_ttl at 0x7a2abec4ad40>, json_schema_input_type=PydanticUndefined)]
environment_ttl: Optional[Annotated[str, BeforeValidator(func=<function validate_no_past_ttl at 0x7a2abec4ad40>, json_schema_input_type=PydanticUndefined)]]
ignore_patterns: List[str]
time_column_format: str
users: List[sqlmesh.core.user.User]
pinned_environments: Set[str]
loader_kwargs: Dict[str, Any]
env_vars: Dict[str, str]
username: str
physical_schema_mapping: Annotated[Dict[re.Pattern, str], BeforeValidator(func=<function validate_regex_key_dict at 0x7a2abecac1f0>, json_schema_input_type=PydanticUndefined)]
physical_table_naming_convention: sqlmesh.core.config.common.TableNamingConvention
gateway_managed_virtual_layer: bool
infer_python_dependencies: bool
environment_catalog_mapping: Annotated[Dict[re.Pattern, str], BeforeValidator(func=<function validate_regex_key_dict at 0x7a2abecac1f0>, json_schema_input_type=PydanticUndefined)]
default_target_environment: str
log_limit: int
variables: Dict[str, Any]
disable_anonymized_analytics: bool
before_all: Optional[List[str]]
after_all: Optional[List[str]]
cache_dir: Optional[str]
def get_default_test_connection( self, default_catalog: Optional[str] = None, default_catalog_dialect: Optional[str] = None) -> sqlmesh.core.config.connection.ConnectionConfig: 
301    def get_default_test_connection(
302        self,
303        default_catalog: t.Optional[str] = None,
304        default_catalog_dialect: t.Optional[str] = None,
305    ) -> ConnectionConfig:
306        return self.default_test_connection_ or DuckDBConnectionConfig(
307            catalogs=(
308                None
309                if default_catalog is None
310                else {
311                    # transpile catalog name from main connection dialect to DuckDB
312                    exp.parse_identifier(default_catalog, dialect=default_catalog_dialect).sql(
313                        dialect="duckdb"
314                    ): ":memory:"
315                }
316            )
317        )
def get_gateway( self, name: Optional[str] = None) -> sqlmesh.core.config.gateway.GatewayConfig: 
319    def get_gateway(self, name: t.Optional[str] = None) -> GatewayConfig:
320        if isinstance(self.gateways, dict):
321            if name is None:
322                if self.default_gateway:
323                    # Normalize default_gateway name to lowercase for lookup
324                    default_key = self.default_gateway.lower()
325                    if default_key not in self.gateways:
326                        raise ConfigError(f"Missing gateway with name '{self.default_gateway}'")
327                    return self.gateways[default_key]
328
329                if "" in self.gateways:
330                    return self.gateways[""]
331
332                return first(self.gateways.values())
333
334            # Normalize lookup name to lowercase since gateway keys are already lowercase
335            lookup_key = name.lower()
336            if lookup_key not in self.gateways:
337                raise ConfigError(f"Missing gateway with name '{name}'.")
338
339            return self.gateways[lookup_key]
340        if name is not None:
341            raise ConfigError("Gateway name is not supported when only one gateway is configured.")
342        return self.gateways
def get_connection( self, gateway_name: Optional[str] = None) -> sqlmesh.core.config.connection.ConnectionConfig: 
344    def get_connection(self, gateway_name: t.Optional[str] = None) -> ConnectionConfig:
345        connection = self.get_gateway(gateway_name).connection or self.default_connection
346        if connection is None:
347            msg = f" for gateway '{gateway_name}'" if gateway_name else ""
348            raise ConfigError(f"No connection configured{msg}.")
349        return connection
def get_state_connection( self, gateway_name: Optional[str] = None) -> Optional[sqlmesh.core.config.connection.ConnectionConfig]: 
351    def get_state_connection(
352        self, gateway_name: t.Optional[str] = None
353    ) -> t.Optional[ConnectionConfig]:
354        return self.get_gateway(gateway_name).state_connection
def get_test_connection( self, gateway_name: Optional[str] = None, default_catalog: Optional[str] = None, default_catalog_dialect: Optional[str] = None) -> sqlmesh.core.config.connection.ConnectionConfig: 
356    def get_test_connection(
357        self,
358        gateway_name: t.Optional[str] = None,
359        default_catalog: t.Optional[str] = None,
360        default_catalog_dialect: t.Optional[str] = None,
361    ) -> ConnectionConfig:
362        return self.get_gateway(gateway_name).test_connection or self.get_default_test_connection(
363            default_catalog=default_catalog, default_catalog_dialect=default_catalog_dialect
364        )
def get_scheduler( self, gateway_name: Optional[str] = None) -> sqlmesh.core.config.scheduler.SchedulerConfig: 
366    def get_scheduler(self, gateway_name: t.Optional[str] = None) -> SchedulerConfig:
367        return self.get_gateway(gateway_name).scheduler or self.default_scheduler
def get_state_schema(self, gateway_name: Optional[str] = None) -> Optional[str]: 
369    def get_state_schema(self, gateway_name: t.Optional[str] = None) -> t.Optional[str]:
370        return self.get_gateway(gateway_name).state_schema
default_gateway_name: str 
372    @property
373    def default_gateway_name(self) -> str:
374        if self.default_gateway:
375            return self.default_gateway
376        if "" in self.gateways:
377            return ""
378        return first(self.gateways)
dialect: Optional[str] 
380    @property
381    def dialect(self) -> t.Optional[str]:
382        return self.model_defaults.dialect
fingerprint: str 
384    @property
385    def fingerprint(self) -> str:
386        return str(
387            zlib.crc32(
388                pickle.dumps(_canonicalize(self.dict(exclude={"loader", "notification_targets"})))
389            )
390        )
model_config = {'json_encoders': {<class 'sqlglot.expressions.core.Expr'>: <function _expression_encoder>, <class 'sqlglot.expressions.datatypes.DataType'>: <function _expression_encoder>, <class 'sqlglot.expressions.query.Tuple'>: <function _expression_encoder>, typing.Union[sqlglot.expressions.query.Query, sqlmesh.core.dialect.JinjaQuery]: <function _expression_encoder>, typing.Union[sqlglot.expressions.query.Query, sqlmesh.core.dialect.JinjaQuery, sqlmesh.core.dialect.MacroFunc]: <function _expression_encoder>, <class 'datetime.tzinfo'>: <function PydanticModel.<lambda>>}, 'arbitrary_types_allowed': True, 'extra': 'forbid', 'protected_namespaces': ()}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

Inherited Members
pydantic.main.BaseModel
BaseModel
model_fields
model_computed_fields
model_extra
model_fields_set
model_construct
model_copy
model_dump
model_dump_json
model_json_schema
model_parametrized_name
model_post_init
model_rebuild
model_validate
model_validate_json
model_validate_strings
parse_file
from_orm
construct
schema
schema_json
validate
update_forward_refs
sqlmesh.core.config.base.BaseConfig
update_with
sqlmesh.utils.pydantic.PydanticModel
dict
json
copy
fields_set
parse_obj
parse_raw
missing_required_fields
extra_fields
all_fields
all_field_infos
required_fields