sqlmesh.core.config.common

 1from __future__ import annotations
 2
 3import typing as t
 4from enum import Enum
 5
 6from sqlmesh.utils import classproperty
 7from sqlmesh.utils.errors import ConfigError
 8from sqlmesh.utils.pydantic import field_validator
 9
10
11class EnvironmentSuffixTarget(str, Enum):
12    SCHEMA = "schema"
13    TABLE = "table"
14
15    @property
16    def is_schema(self) -> bool:
17        return self == EnvironmentSuffixTarget.SCHEMA
18
19    @property
20    def is_table(self) -> bool:
21        return self == EnvironmentSuffixTarget.TABLE
22
23    @classproperty
24    def default(cls) -> EnvironmentSuffixTarget:
25        return EnvironmentSuffixTarget.SCHEMA
26
27    def __str__(self) -> str:
28        return self.name
29
30    def __repr__(self) -> str:
31        return str(self)
32
33
34def _concurrent_tasks_validator(v: t.Any) -> int:
35    if isinstance(v, str):
36        v = int(v)
37    if not isinstance(v, int) or v <= 0:
38        raise ConfigError(
39            f"The number of concurrent tasks must be an integer value greater than 0. '{v}' was provided"
40        )
41    return v
42
43
44concurrent_tasks_validator = field_validator(
45    "backfill_concurrent_tasks",
46    "ddl_concurrent_tasks",
47    "concurrent_tasks",
48    mode="before",
49    check_fields=False,
50)(_concurrent_tasks_validator)
51
52
53def _http_headers_validator(v: t.Any) -> t.Any:
54    if isinstance(v, dict):
55        return [(key, value) for key, value in v.items()]
56    return v
57
58
59http_headers_validator = field_validator(
60    "http_headers",
61    mode="before",
62    check_fields=False,
63)(_http_headers_validator)
64
65
66def _variables_validator(value: t.Dict[str, t.Any]) -> t.Dict[str, t.Any]:
67    if not isinstance(value, dict):
68        raise ConfigError(f"Variables must be a dictionary, not {type(value)}")
69
70    def _validate_type(v: t.Any) -> None:
71        if isinstance(v, list):
72            for item in v:
73                _validate_type(item)
74        elif isinstance(v, dict):
75            for item in v.values():
76                _validate_type(item)
77        elif v is not None and not isinstance(v, (str, int, float, bool)):
78            raise ConfigError(f"Unsupported variable value type: {type(v)}")
79
80    _validate_type(value)
81    return {k.lower(): v for k, v in value.items()}
82
83
84variables_validator = field_validator(
85    "variables",
86    mode="before",
87    check_fields=False,
88)(_variables_validator)