Edit on GitHub

sqlmesh.core.environment

View Source

  1from __future__ import annotations
  2
  3import json
  4import re
  5import typing as t
  6
  7from pydantic import Field
  8
  9from sqlmesh.core import constants as c
 10from sqlmesh.core.config import EnvironmentSuffixTarget
 11from sqlmesh.core.engine_adapter.base import EngineAdapter
 12from sqlmesh.core.macros import RuntimeStage
 13from sqlmesh.core.renderer import render_statements
 14from sqlmesh.core.snapshot import SnapshotId, SnapshotTableInfo, Snapshot
 15from sqlmesh.utils import word_characters_only
 16from sqlmesh.utils.date import TimeLike, now_timestamp
 17from sqlmesh.utils.errors import SQLMeshError
 18from sqlmesh.utils.jinja import JinjaMacroRegistry
 19from sqlmesh.utils.metaprogramming import Executable
 20from sqlmesh.utils.pydantic import PydanticModel, field_validator, ValidationInfo
 21
 22T = t.TypeVar("T", bound="EnvironmentNamingInfo")
 23PydanticType = t.TypeVar("PydanticType", bound="PydanticModel")
 24
 25
 26class EnvironmentNamingInfo(PydanticModel):
 27    """
 28    Information required for creating an object within an environment
 29
 30    Args:
 31        name: The name of the environment.
 32        suffix_target: Indicates whether to append the environment name to the schema or table name.
 33        catalog_name_override: The name of the catalog to use for this environment if an override was provided
 34        normalize_name: Indicates whether the environment's name will be normalized. For example, if it's
 35            `dev`, then it will become `DEV` when targeting Snowflake.
 36        gateway_managed: Determines whether the virtual layer's views are created by the model-specific
 37            gateways, otherwise the default gateway is used. Default: False.
 38    """
 39
 40    name: str = c.PROD
 41    suffix_target: EnvironmentSuffixTarget = Field(default=EnvironmentSuffixTarget.SCHEMA)
 42    catalog_name_override: t.Optional[str] = None
 43    normalize_name: bool = True
 44    gateway_managed: bool = False
 45
 46    @property
 47    def is_dev(self) -> bool:
 48        return self.name.lower() != c.PROD
 49
 50    @field_validator("name", mode="before")
 51    @classmethod
 52    def _sanitize_name(cls, v: str) -> str:
 53        return word_characters_only(v).lower()
 54
 55    @field_validator("normalize_name", "gateway_managed", mode="before")
 56    @classmethod
 57    def _validate_boolean_field(cls, v: t.Any, info: ValidationInfo) -> bool:
 58        if v is None:
 59            # Pydantic 2.13+ sets field_name to None during model_validate_json()
 60            return (info.field_name or "") == "normalize_name"
 61        return bool(v)
 62
 63    @t.overload
 64    @classmethod
 65    def sanitize_name(cls, v: str) -> str: ...
 66
 67    @t.overload
 68    @classmethod
 69    def sanitize_name(cls, v: Environment) -> Environment: ...
 70
 71    @classmethod
 72    def sanitize_name(cls, v: str | Environment) -> str | Environment:
 73        """
 74        Sanitizes the environment name so we create names that are valid names for database objects.
 75        This means alphanumeric and underscores only. Invalid characters are replaced with underscores.
 76        """
 77        if isinstance(v, Environment):
 78            return v
 79        if not isinstance(v, str):
 80            raise TypeError(f"Expected str or Environment, got {type(v).__name__}")
 81        return cls._sanitize_name(v)
 82
 83    @classmethod
 84    def sanitize_names(cls, values: t.Iterable[str]) -> t.Set[str]:
 85        return {cls.sanitize_name(value) for value in values}
 86
 87    @classmethod
 88    def from_environment_catalog_mapping(
 89        cls: t.Type[T],
 90        environment_catalog_mapping: t.Dict[re.Pattern, str],
 91        name: str = c.PROD,
 92        **kwargs: t.Any,
 93    ) -> T:
 94        construction_kwargs = dict(name=name, **kwargs)
 95        for re_pattern, catalog_name in environment_catalog_mapping.items():
 96            if re.match(re_pattern, name):
 97                return cls(
 98                    catalog_name_override=catalog_name,
 99                    **construction_kwargs,
100                )
101        return cls(**construction_kwargs)
102
103
104class EnvironmentSummary(PydanticModel):
105    """Represents summary information of an isolated environment.
106
107    Args:
108        name: The name of the environment.
109        start_at: The start time of the environment.
110        end_at: The end time of the environment.
111        plan_id: The ID of the plan that last updated this environment.
112        previous_plan_id: The ID of the previous plan that updated this environment.
113        expiration_ts: The timestamp when this environment will expire.
114        finalized_ts: The timestamp when this environment was finalized.
115    """
116
117    name: str
118    start_at: TimeLike
119    end_at: t.Optional[TimeLike] = None
120    plan_id: str
121    previous_plan_id: t.Optional[str] = None
122    expiration_ts: t.Optional[int] = None
123    finalized_ts: t.Optional[int] = None
124
125    @property
126    def expired(self) -> bool:
127        return self.expiration_ts is not None and self.expiration_ts <= now_timestamp()
128
129
130class Environment(EnvironmentNamingInfo, EnvironmentSummary):
131    """Represents an isolated environment.
132
133    Environments are isolated workspaces that hold pointers to physical tables.
134
135    Args:
136        snapshots: The snapshots that are part of this environment.
137        promoted_snapshot_ids: The IDs of the snapshots that are promoted in this environment
138            (i.e. for which the views are created). If not specified, all snapshots are promoted.
139        previous_finalized_snapshots: Snapshots that were part of this environment last time it was finalized.
140        requirements: A mapping of library versions for all the snapshots in this environment.
141    """
142
143    snapshots_: t.List[t.Any] = Field(alias="snapshots")
144    promoted_snapshot_ids_: t.Optional[t.List[t.Any]] = Field(
145        default=None, alias="promoted_snapshot_ids"
146    )
147    previous_finalized_snapshots_: t.Optional[t.List[t.Any]] = Field(
148        default=None, alias="previous_finalized_snapshots"
149    )
150    requirements: t.Dict[str, str] = {}
151
152    @field_validator("snapshots_", "previous_finalized_snapshots_", mode="before")
153    @classmethod
154    def _load_snapshots(cls, v: str | t.List[t.Any] | None) -> t.List[t.Any] | None:
155        if isinstance(v, str):
156            return json.loads(v)
157        if v and not isinstance(next(iter(v)), (dict, SnapshotTableInfo)):
158            raise ValueError("Must be a list of SnapshotTableInfo dicts or objects")
159        return v
160
161    @field_validator("promoted_snapshot_ids_", mode="before")
162    @classmethod
163    def _load_snapshot_ids(cls, v: str | t.List[t.Any] | None) -> t.List[t.Any] | None:
164        if isinstance(v, str):
165            return json.loads(v)
166        if v and not isinstance(next(iter(v)), (dict, SnapshotId)):
167            raise ValueError("Must be a list of SnapshotId dicts or objects")
168        return v
169
170    @field_validator("requirements", mode="before")
171    def _load_requirements(cls, v: t.Any) -> t.Any:
172        if isinstance(v, str):
173            v = json.loads(v)
174        return v or {}
175
176    @property
177    def snapshots(self) -> t.List[SnapshotTableInfo]:
178        return self._convert_list_to_models_and_store("snapshots_", SnapshotTableInfo) or []
179
180    def snapshot_dicts(self) -> t.List[dict]:
181        return self._convert_list_to_dicts(self.snapshots_)
182
183    @property
184    def promoted_snapshot_ids(self) -> t.Optional[t.List[SnapshotId]]:
185        return self._convert_list_to_models_and_store("promoted_snapshot_ids_", SnapshotId)
186
187    def promoted_snapshot_id_dicts(self) -> t.List[dict]:
188        return self._convert_list_to_dicts(self.promoted_snapshot_ids_)
189
190    @property
191    def promoted_snapshots(self) -> t.List[SnapshotTableInfo]:
192        if self.promoted_snapshot_ids is None:
193            return self.snapshots
194
195        promoted_snapshot_ids = set(self.promoted_snapshot_ids)
196        return [s for s in self.snapshots if s.snapshot_id in promoted_snapshot_ids]
197
198    @property
199    def previous_finalized_snapshots(self) -> t.Optional[t.List[SnapshotTableInfo]]:
200        return self._convert_list_to_models_and_store(
201            "previous_finalized_snapshots_", SnapshotTableInfo
202        )
203
204    def previous_finalized_snapshot_dicts(self) -> t.List[dict]:
205        return self._convert_list_to_dicts(self.previous_finalized_snapshots_)
206
207    @property
208    def finalized_or_current_snapshots(self) -> t.List[SnapshotTableInfo]:
209        return (
210            self.snapshots
211            if self.finalized_ts
212            else self.previous_finalized_snapshots or self.snapshots
213        )
214
215    @property
216    def naming_info(self) -> EnvironmentNamingInfo:
217        return EnvironmentNamingInfo(
218            name=self.name,
219            suffix_target=self.suffix_target,
220            catalog_name_override=self.catalog_name_override,
221            normalize_name=self.normalize_name,
222            gateway_managed=self.gateway_managed,
223        )
224
225    @property
226    def summary(self) -> EnvironmentSummary:
227        return EnvironmentSummary(
228            name=self.name,
229            start_at=self.start_at,
230            end_at=self.end_at,
231            plan_id=self.plan_id,
232            previous_plan_id=self.previous_plan_id,
233            expiration_ts=self.expiration_ts,
234            finalized_ts=self.finalized_ts,
235        )
236
237    def can_partially_promote(self, existing_environment: Environment) -> bool:
238        """Returns True if the existing environment can be partially promoted to the current environment.
239
240        Partial promotion means that we don't need to re-create views for snapshots that are already promoted in the
241        target environment.
242        """
243        return (
244            bool(existing_environment.finalized_ts)
245            and not existing_environment.expired
246            and existing_environment.gateway_managed == self.gateway_managed
247            and existing_environment.name == c.PROD
248        )
249
250    def _convert_list_to_models_and_store(
251        self, field: str, type_: t.Type[PydanticType]
252    ) -> t.Optional[t.List[PydanticType]]:
253        value = getattr(self, field)
254        if value and not isinstance(value[0], type_):
255            value = [type_.parse_obj(obj) for obj in value]
256            setattr(self, field, value)
257        return value
258
259    def _convert_list_to_dicts(self, value: t.Optional[t.List[t.Any]]) -> t.List[dict]:
260        if not value:
261            return []
262        return value if isinstance(value[0], dict) else [v.dict() for v in value]
263
264
265class EnvironmentStatements(PydanticModel):
266    before_all: t.List[str]
267    after_all: t.List[str]
268    python_env: t.Dict[str, Executable]
269    jinja_macros: t.Optional[JinjaMacroRegistry] = None
270    project: t.Optional[str] = None
271
272    def render_before_all(
273        self,
274        dialect: str,
275        default_catalog: t.Optional[str] = None,
276        **render_kwargs: t.Any,
277    ) -> t.List[str]:
278        return self.render(RuntimeStage.BEFORE_ALL, dialect, default_catalog, **render_kwargs)
279
280    def render_after_all(
281        self,
282        dialect: str,
283        default_catalog: t.Optional[str] = None,
284        **render_kwargs: t.Any,
285    ) -> t.List[str]:
286        return self.render(RuntimeStage.AFTER_ALL, dialect, default_catalog, **render_kwargs)
287
288    def render(
289        self,
290        runtime_stage: RuntimeStage,
291        dialect: str,
292        default_catalog: t.Optional[str] = None,
293        **render_kwargs: t.Any,
294    ) -> t.List[str]:
295        return render_statements(
296            statements=getattr(self, runtime_stage.value),
297            dialect=dialect,
298            default_catalog=default_catalog,
299            python_env=self.python_env,
300            jinja_macros=self.jinja_macros,
301            runtime_stage=runtime_stage,
302            **render_kwargs,
303        )
304
305
306def execute_environment_statements(
307    adapter: EngineAdapter,
308    environment_statements: t.List[EnvironmentStatements],
309    runtime_stage: RuntimeStage,
310    environment_naming_info: EnvironmentNamingInfo,
311    default_catalog: t.Optional[str] = None,
312    snapshots: t.Optional[t.Dict[str, Snapshot]] = None,
313    start: t.Optional[TimeLike] = None,
314    end: t.Optional[TimeLike] = None,
315    execution_time: t.Optional[TimeLike] = None,
316    selected_models: t.Optional[t.Set[str]] = None,
317) -> None:
318    try:
319        rendered_expressions = [
320            expr
321            for statements in environment_statements
322            for expr in statements.render(
323                runtime_stage=runtime_stage,
324                dialect=adapter.dialect,
325                default_catalog=default_catalog,
326                snapshots=snapshots,
327                start=start,
328                end=end,
329                execution_time=execution_time,
330                environment_naming_info=environment_naming_info,
331                engine_adapter=adapter,
332                selected_models=selected_models,
333            )
334        ]
335    except Exception as e:
336        raise SQLMeshError(
337            f"An error occurred during rendering of the '{runtime_stage.value}' statements:\n\n{e}"
338        )
339    if rendered_expressions:
340        with adapter.transaction():
341            for expr in rendered_expressions:
342                try:
343                    adapter.execute(expr)
344                except Exception as e:
345                    raise SQLMeshError(
346                        f"An error occurred during execution of the following '{runtime_stage.value}' statement:\n\n{expr}\n\n{e}"
347                    )

class EnvironmentNamingInfo(sqlmesh.utils.pydantic.PydanticModel): View Source

 27class EnvironmentNamingInfo(PydanticModel):
 28    """
 29    Information required for creating an object within an environment
 30
 31    Args:
 32        name: The name of the environment.
 33        suffix_target: Indicates whether to append the environment name to the schema or table name.
 34        catalog_name_override: The name of the catalog to use for this environment if an override was provided
 35        normalize_name: Indicates whether the environment's name will be normalized. For example, if it's
 36            `dev`, then it will become `DEV` when targeting Snowflake.
 37        gateway_managed: Determines whether the virtual layer's views are created by the model-specific
 38            gateways, otherwise the default gateway is used. Default: False.
 39    """
 40
 41    name: str = c.PROD
 42    suffix_target: EnvironmentSuffixTarget = Field(default=EnvironmentSuffixTarget.SCHEMA)
 43    catalog_name_override: t.Optional[str] = None
 44    normalize_name: bool = True
 45    gateway_managed: bool = False
 46
 47    @property
 48    def is_dev(self) -> bool:
 49        return self.name.lower() != c.PROD
 50
 51    @field_validator("name", mode="before")
 52    @classmethod
 53    def _sanitize_name(cls, v: str) -> str:
 54        return word_characters_only(v).lower()
 55
 56    @field_validator("normalize_name", "gateway_managed", mode="before")
 57    @classmethod
 58    def _validate_boolean_field(cls, v: t.Any, info: ValidationInfo) -> bool:
 59        if v is None:
 60            # Pydantic 2.13+ sets field_name to None during model_validate_json()
 61            return (info.field_name or "") == "normalize_name"
 62        return bool(v)
 63
 64    @t.overload
 65    @classmethod
 66    def sanitize_name(cls, v: str) -> str: ...
 67
 68    @t.overload
 69    @classmethod
 70    def sanitize_name(cls, v: Environment) -> Environment: ...
 71
 72    @classmethod
 73    def sanitize_name(cls, v: str | Environment) -> str | Environment:
 74        """
 75        Sanitizes the environment name so we create names that are valid names for database objects.
 76        This means alphanumeric and underscores only. Invalid characters are replaced with underscores.
 77        """
 78        if isinstance(v, Environment):
 79            return v
 80        if not isinstance(v, str):
 81            raise TypeError(f"Expected str or Environment, got {type(v).__name__}")
 82        return cls._sanitize_name(v)
 83
 84    @classmethod
 85    def sanitize_names(cls, values: t.Iterable[str]) -> t.Set[str]:
 86        return {cls.sanitize_name(value) for value in values}
 87
 88    @classmethod
 89    def from_environment_catalog_mapping(
 90        cls: t.Type[T],
 91        environment_catalog_mapping: t.Dict[re.Pattern, str],
 92        name: str = c.PROD,
 93        **kwargs: t.Any,
 94    ) -> T:
 95        construction_kwargs = dict(name=name, **kwargs)
 96        for re_pattern, catalog_name in environment_catalog_mapping.items():
 97            if re.match(re_pattern, name):
 98                return cls(
 99                    catalog_name_override=catalog_name,
100                    **construction_kwargs,
101                )
102        return cls(**construction_kwargs)

Information required for creating an object within an environment

Arguments:

name: The name of the environment.
suffix_target: Indicates whether to append the environment name to the schema or table name.
catalog_name_override: The name of the catalog to use for this environment if an override was provided
normalize_name: Indicates whether the environment's name will be normalized. For example, if it's dev, then it will become DEV when targeting Snowflake.
gateway_managed: Determines whether the virtual layer's views are created by the model-specific gateways, otherwise the default gateway is used. Default: False.

name: str

suffix_target: sqlmesh.core.config.common.EnvironmentSuffixTarget

catalog_name_override: Optional[str]

normalize_name: bool

gateway_managed: bool

is_dev: bool View Source

47    @property
48    def is_dev(self) -> bool:
49        return self.name.lower() != c.PROD

@classmethod

def sanitize_name( cls, v: str | Environment) -> str | Environment: View Source

72    @classmethod
73    def sanitize_name(cls, v: str | Environment) -> str | Environment:
74        """
75        Sanitizes the environment name so we create names that are valid names for database objects.
76        This means alphanumeric and underscores only. Invalid characters are replaced with underscores.
77        """
78        if isinstance(v, Environment):
79            return v
80        if not isinstance(v, str):
81            raise TypeError(f"Expected str or Environment, got {type(v).__name__}")
82        return cls._sanitize_name(v)

Sanitizes the environment name so we create names that are valid names for database objects. This means alphanumeric and underscores only. Invalid characters are replaced with underscores.

@classmethod

def sanitize_names(cls, values: Iterable[str]) -> Set[str]: View Source

84    @classmethod
85    def sanitize_names(cls, values: t.Iterable[str]) -> t.Set[str]:
86        return {cls.sanitize_name(value) for value in values}

@classmethod

def from_environment_catalog_mapping( cls: Type[~T], environment_catalog_mapping: Dict[re.Pattern, str], name: str = 'prod', **kwargs: Any) -> ~T: View Source

 88    @classmethod
 89    def from_environment_catalog_mapping(
 90        cls: t.Type[T],
 91        environment_catalog_mapping: t.Dict[re.Pattern, str],
 92        name: str = c.PROD,
 93        **kwargs: t.Any,
 94    ) -> T:
 95        construction_kwargs = dict(name=name, **kwargs)
 96        for re_pattern, catalog_name in environment_catalog_mapping.items():
 97            if re.match(re_pattern, name):
 98                return cls(
 99                    catalog_name_override=catalog_name,
100                    **construction_kwargs,
101                )
102        return cls(**construction_kwargs)

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.utils.pydantic.PydanticModel: dict; json; copy; fields_set; parse_obj; parse_raw; missing_required_fields; extra_fields; all_fields; all_field_infos; required_fields

class EnvironmentSummary(sqlmesh.utils.pydantic.PydanticModel): View Source

105class EnvironmentSummary(PydanticModel):
106    """Represents summary information of an isolated environment.
107
108    Args:
109        name: The name of the environment.
110        start_at: The start time of the environment.
111        end_at: The end time of the environment.
112        plan_id: The ID of the plan that last updated this environment.
113        previous_plan_id: The ID of the previous plan that updated this environment.
114        expiration_ts: The timestamp when this environment will expire.
115        finalized_ts: The timestamp when this environment was finalized.
116    """
117
118    name: str
119    start_at: TimeLike
120    end_at: t.Optional[TimeLike] = None
121    plan_id: str
122    previous_plan_id: t.Optional[str] = None
123    expiration_ts: t.Optional[int] = None
124    finalized_ts: t.Optional[int] = None
125
126    @property
127    def expired(self) -> bool:
128        return self.expiration_ts is not None and self.expiration_ts <= now_timestamp()

Represents summary information of an isolated environment.

Arguments:

name: The name of the environment.
start_at: The start time of the environment.
end_at: The end time of the environment.
plan_id: The ID of the plan that last updated this environment.
previous_plan_id: The ID of the previous plan that updated this environment.
expiration_ts: The timestamp when this environment will expire.
finalized_ts: The timestamp when this environment was finalized.

name: str

start_at: Union[datetime.date, datetime.datetime, str, int, float]

end_at: Union[datetime.date, datetime.datetime, str, int, float, NoneType]

plan_id: str

previous_plan_id: Optional[str]

expiration_ts: Optional[int]

finalized_ts: Optional[int]

expired: bool View Source

126    @property
127    def expired(self) -> bool:
128        return self.expiration_ts is not None and self.expiration_ts <= now_timestamp()

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.utils.pydantic.PydanticModel: dict; json; copy; fields_set; parse_obj; parse_raw; missing_required_fields; extra_fields; all_fields; all_field_infos; required_fields

class EnvironmentStatements(sqlmesh.utils.pydantic.PydanticModel): View Source

266class EnvironmentStatements(PydanticModel):
267    before_all: t.List[str]
268    after_all: t.List[str]
269    python_env: t.Dict[str, Executable]
270    jinja_macros: t.Optional[JinjaMacroRegistry] = None
271    project: t.Optional[str] = None
272
273    def render_before_all(
274        self,
275        dialect: str,
276        default_catalog: t.Optional[str] = None,
277        **render_kwargs: t.Any,
278    ) -> t.List[str]:
279        return self.render(RuntimeStage.BEFORE_ALL, dialect, default_catalog, **render_kwargs)
280
281    def render_after_all(
282        self,
283        dialect: str,
284        default_catalog: t.Optional[str] = None,
285        **render_kwargs: t.Any,
286    ) -> t.List[str]:
287        return self.render(RuntimeStage.AFTER_ALL, dialect, default_catalog, **render_kwargs)
288
289    def render(
290        self,
291        runtime_stage: RuntimeStage,
292        dialect: str,
293        default_catalog: t.Optional[str] = None,
294        **render_kwargs: t.Any,
295    ) -> t.List[str]:
296        return render_statements(
297            statements=getattr(self, runtime_stage.value),
298            dialect=dialect,
299            default_catalog=default_catalog,
300            python_env=self.python_env,
301            jinja_macros=self.jinja_macros,
302            runtime_stage=runtime_stage,
303            **render_kwargs,
304        )

!!! abstract "Usage Documentation" Models

A base class for creating Pydantic models.

Attributes:

__class_vars__: The names of the class variables defined on the model.
__private_attributes__: Metadata about the private attributes of the model.
__signature__: The synthesized __init__ [Signature][inspect.Signature] of the model.
__pydantic_complete__: Whether model building is completed, or if there are still undefined fields.
__pydantic_core_schema__: The core schema of the model.
__pydantic_custom_init__: Whether the model has a custom __init__ function.
__pydantic_decorators__: Metadata containing the decorators defined on the model. This replaces Model.__validators__ and Model.__root_validators__ from Pydantic V1.
__pydantic_generic_metadata__: A dictionary containing metadata about generic Pydantic models. The origin and args items map to the [__origin__][genericalias.__origin__] and [__args__][genericalias.__args__] attributes of [generic aliases][types-genericalias], and the parameter item maps to the __parameter__ attribute of generic classes.
__pydantic_parent_namespace__: Parent namespace of the model, used for automatic rebuilding of models.
__pydantic_post_init__: The name of the post-init method for the model, if defined.
__pydantic_root_model__: Whether the model is a [RootModel][pydantic.root_model.RootModel].
__pydantic_serializer__: The pydantic-core SchemaSerializer used to dump instances of the model.
__pydantic_validator__: The pydantic-core SchemaValidator used to validate instances of the model.
__pydantic_fields__: A dictionary of field names and their corresponding [FieldInfo][pydantic.fields.FieldInfo] objects.
__pydantic_computed_fields__: A dictionary of computed field names and their corresponding [ComputedFieldInfo][pydantic.fields.ComputedFieldInfo] objects.
__pydantic_extra__: A dictionary containing extra values, if [extra][pydantic.config.ConfigDict.extra] is set to 'allow'.
__pydantic_fields_set__: The names of fields explicitly set during instantiation.
__pydantic_private__: Values of private attributes set on the model instance.

before_all: List[str]

after_all: List[str]

python_env: Dict[str, sqlmesh.utils.metaprogramming.Executable]

jinja_macros: Optional[sqlmesh.utils.jinja.JinjaMacroRegistry]

project: Optional[str]

def render_before_all( self, dialect: str, default_catalog: Optional[str] = None, **render_kwargs: Any) -> List[str]: View Source

273    def render_before_all(
274        self,
275        dialect: str,
276        default_catalog: t.Optional[str] = None,
277        **render_kwargs: t.Any,
278    ) -> t.List[str]:
279        return self.render(RuntimeStage.BEFORE_ALL, dialect, default_catalog, **render_kwargs)

def render_after_all( self, dialect: str, default_catalog: Optional[str] = None, **render_kwargs: Any) -> List[str]: View Source

281    def render_after_all(
282        self,
283        dialect: str,
284        default_catalog: t.Optional[str] = None,
285        **render_kwargs: t.Any,
286    ) -> t.List[str]:
287        return self.render(RuntimeStage.AFTER_ALL, dialect, default_catalog, **render_kwargs)

def render( self, runtime_stage: sqlmesh.core.macros.RuntimeStage, dialect: str, default_catalog: Optional[str] = None, **render_kwargs: Any) -> List[str]: View Source

289    def render(
290        self,
291        runtime_stage: RuntimeStage,
292        dialect: str,
293        default_catalog: t.Optional[str] = None,
294        **render_kwargs: t.Any,
295    ) -> t.List[str]:
296        return render_statements(
297            statements=getattr(self, runtime_stage.value),
298            dialect=dialect,
299            default_catalog=default_catalog,
300            python_env=self.python_env,
301            jinja_macros=self.jinja_macros,
302            runtime_stage=runtime_stage,
303            **render_kwargs,
304        )

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.utils.pydantic.PydanticModel: dict; json; copy; fields_set; parse_obj; parse_raw; missing_required_fields; extra_fields; all_fields; all_field_infos; required_fields

def execute_environment_statements( adapter: sqlmesh.core.engine_adapter.base.EngineAdapter, environment_statements: List[EnvironmentStatements], runtime_stage: sqlmesh.core.macros.RuntimeStage, environment_naming_info: EnvironmentNamingInfo, default_catalog: Optional[str] = None, snapshots: Optional[Dict[str, sqlmesh.core.snapshot.definition.Snapshot]] = None, start: Union[datetime.date, datetime.datetime, str, int, float, NoneType] = None, end: Union[datetime.date, datetime.datetime, str, int, float, NoneType] = None, execution_time: Union[datetime.date, datetime.datetime, str, int, float, NoneType] = None, selected_models: Optional[Set[str]] = None) -> None: View Source

307def execute_environment_statements(
308    adapter: EngineAdapter,
309    environment_statements: t.List[EnvironmentStatements],
310    runtime_stage: RuntimeStage,
311    environment_naming_info: EnvironmentNamingInfo,
312    default_catalog: t.Optional[str] = None,
313    snapshots: t.Optional[t.Dict[str, Snapshot]] = None,
314    start: t.Optional[TimeLike] = None,
315    end: t.Optional[TimeLike] = None,
316    execution_time: t.Optional[TimeLike] = None,
317    selected_models: t.Optional[t.Set[str]] = None,
318) -> None:
319    try:
320        rendered_expressions = [
321            expr
322            for statements in environment_statements
323            for expr in statements.render(
324                runtime_stage=runtime_stage,
325                dialect=adapter.dialect,
326                default_catalog=default_catalog,
327                snapshots=snapshots,
328                start=start,
329                end=end,
330                execution_time=execution_time,
331                environment_naming_info=environment_naming_info,
332                engine_adapter=adapter,
333                selected_models=selected_models,
334            )
335        ]
336    except Exception as e:
337        raise SQLMeshError(
338            f"An error occurred during rendering of the '{runtime_stage.value}' statements:\n\n{e}"
339        )
340    if rendered_expressions:
341        with adapter.transaction():
342            for expr in rendered_expressions:
343                try:
344                    adapter.execute(expr)
345                except Exception as e:
346                    raise SQLMeshError(
347                        f"An error occurred during execution of the following '{runtime_stage.value}' statement:\n\n{expr}\n\n{e}"
348                    )