sqlmesh.core.config.common

  1from __future__ import annotations
  2
  3import typing as t
  4from enum import Enum
  5import re
  6
  7from sqlmesh.utils import classproperty
  8from sqlmesh.utils.errors import ConfigError
  9from sqlmesh.utils.pydantic import field_validator
 10
 11# Config files that can be present in the project dir
 12ALL_CONFIG_FILENAMES = ("config.py", "config.yml", "config.yaml", "sqlmesh.yml", "sqlmesh.yaml")
 13
 14# For personal paths (~/.sqlmesh/) where python config is not supported
 15YAML_CONFIG_FILENAMES = tuple(n for n in ALL_CONFIG_FILENAMES if not n.endswith(".py"))
 16
 17# Note: is here to prevent having to import from sqlmesh.dbt.loader which introduces a dependency
 18# on dbt-core in a native project
 19DBT_PROJECT_FILENAME = "dbt_project.yml"
 20
 21
 22class EnvironmentSuffixTarget(str, Enum):
 23    # Intended to create virtual environments in their own schemas, with names like "<model_schema_name>__<env name>". The view name is untouched.
 24    # For example, a model named 'sqlmesh_example.full_model' created in an environment called 'dev'
 25    # would have its virtual layer view created as 'sqlmesh_example__dev.full_model'
 26    SCHEMA = "schema"
 27
 28    # Intended to create virtual environments in the same schema as their production counterparts by adjusting the table name.
 29    # For example, a model named 'sqlmesh_example.full_model' created in an environment called 'dev'
 30    # would have its virtual layer view created as "sqlmesh_example.full_model__dev"
 31    TABLE = "table"
 32
 33    # Intended to create virtual environments in their own catalogs to preserve the schema and view name of the models
 34    # For example, a model named 'sqlmesh_example.full_model' created in an environment called 'dev'
 35    # with a default catalog of "warehouse" would have its virtual layer view created as "warehouse__dev.sqlmesh_example.full_model"
 36    # note: this only works for engines that can query across catalogs
 37    CATALOG = "catalog"
 38
 39    @property
 40    def is_schema(self) -> bool:
 41        return self == EnvironmentSuffixTarget.SCHEMA
 42
 43    @property
 44    def is_table(self) -> bool:
 45        return self == EnvironmentSuffixTarget.TABLE
 46
 47    @property
 48    def is_catalog(self) -> bool:
 49        return self == EnvironmentSuffixTarget.CATALOG
 50
 51    @classproperty
 52    def default(cls) -> EnvironmentSuffixTarget:
 53        return EnvironmentSuffixTarget.SCHEMA
 54
 55    def __str__(self) -> str:
 56        return self.name
 57
 58    def __repr__(self) -> str:
 59        return str(self)
 60
 61
 62class VirtualEnvironmentMode(str, Enum):
 63    """Mode for virtual environment behavior.
 64
 65    FULL: Use full virtual environment functionality with versioned table names and virtual layer updates.
 66    DEV_ONLY: Bypass virtual environments in production, using original unversioned model names.
 67    """
 68
 69    FULL = "full"
 70    DEV_ONLY = "dev_only"
 71
 72    @property
 73    def is_full(self) -> bool:
 74        return self == VirtualEnvironmentMode.FULL
 75
 76    @property
 77    def is_dev_only(self) -> bool:
 78        return self == VirtualEnvironmentMode.DEV_ONLY
 79
 80    @classproperty
 81    def default(cls) -> VirtualEnvironmentMode:
 82        return VirtualEnvironmentMode.FULL
 83
 84    def __str__(self) -> str:
 85        return self.name
 86
 87    def __repr__(self) -> str:
 88        return str(self)
 89
 90
 91class TableNamingConvention(str, Enum):
 92    # Causes table names at the physical layer to follow the convention:
 93    # <schema-name>__<table-name>__<fingerprint>
 94    SCHEMA_AND_TABLE = "schema_and_table"
 95
 96    # Causes table names at the physical layer to follow the convention:
 97    # <table-name>__<fingerprint>
 98    TABLE_ONLY = "table_only"
 99
100    # Takes the table name that would be returned from SCHEMA_AND_TABLE and wraps it in md5()
101    # to generate a hash and prefixes the has with `sqlmesh_md5__`, for the following reasons:
102    # - at a glance, you can still see it's managed by sqlmesh and that md5 was used to generate the hash
103    # - unquoted identifiers that start with numbers can trip up DB engine parsers, so having a text prefix prevents this
104    # This causes table names at the physical layer to follow the convention:
105    # sqlmesh_md5__3b07384d113edec49eaa6238ad5ff00d
106    HASH_MD5 = "hash_md5"
107
108    @classproperty
109    def default(cls) -> TableNamingConvention:
110        return TableNamingConvention.SCHEMA_AND_TABLE
111
112    def __str__(self) -> str:
113        return self.name
114
115    def __repr__(self) -> str:
116        return str(self)
117
118
119def _concurrent_tasks_validator(v: t.Any) -> int:
120    if isinstance(v, str):
121        v = int(v)
122    if not isinstance(v, int) or v <= 0:
123        raise ConfigError(
124            f"The number of concurrent tasks must be an integer value greater than 0. '{v}' was provided"
125        )
126    return v
127
128
129concurrent_tasks_validator = field_validator(
130    "backfill_concurrent_tasks",
131    "ddl_concurrent_tasks",
132    "concurrent_tasks",
133    mode="before",
134    check_fields=False,
135)(_concurrent_tasks_validator)
136
137
138def _http_headers_validator(v: t.Any) -> t.Any:
139    if isinstance(v, dict):
140        return [(key, value) for key, value in v.items()]
141    return v
142
143
144http_headers_validator = field_validator(
145    "http_headers",
146    mode="before",
147    check_fields=False,
148)(_http_headers_validator)
149
150
151def _variables_validator(value: t.Dict[str, t.Any]) -> t.Dict[str, t.Any]:
152    if not isinstance(value, dict):
153        raise ConfigError(f"Variables must be a dictionary, not {type(value)}")
154
155    def _validate_type(v: t.Any) -> None:
156        if isinstance(v, list):
157            for item in v:
158                _validate_type(item)
159        elif isinstance(v, dict):
160            for item in v.values():
161                _validate_type(item)
162        elif v is not None and not isinstance(v, (str, int, float, bool)):
163            raise ConfigError(f"Unsupported variable value type: {type(v)}")
164
165    _validate_type(value)
166    return {k.lower(): v for k, v in value.items()}
167
168
169variables_validator = field_validator(
170    "variables",
171    mode="before",
172    check_fields=False,
173)(_variables_validator)
174
175
176def compile_regex_mapping(value: t.Dict[str | re.Pattern, t.Any]) -> t.Dict[re.Pattern, t.Any]:
177    """
178    Utility function to compile a dict of { "string regex pattern" : "string value" } into { "<re.Pattern>": "string value" }
179    """
180    compiled_regexes = {}
181    for k, v in value.items():
182        try:
183            compiled_regexes[re.compile(k)] = v
184        except re.error:
185            raise ConfigError(f"`{k}` is not a valid regular expression.")
186    return compiled_regexes