sqlmesh.core.model.definition
1from __future__ import annotations 2 3import json 4import logging 5import types 6import re 7import typing as t 8from functools import cached_property, partial 9from pathlib import Path 10 11from pydantic import Field 12from sqlglot import diff, exp 13from sqlglot.diff import Insert 14from sqlglot.helper import seq_get 15from sqlglot.optimizer.qualify_columns import quote_identifiers 16from sqlglot.optimizer.simplify import gen 17from sqlglot.optimizer.normalize_identifiers import normalize_identifiers 18from sqlglot.schema import MappingSchema, nested_set 19from sqlglot.time import format_time 20 21from sqlmesh.core import constants as c 22from sqlmesh.core import dialect as d 23from sqlmesh.core.audit import Audit, ModelAudit 24from sqlmesh.core.node import IntervalUnit 25from sqlmesh.core.macros import MacroRegistry, macro 26from sqlmesh.core.model.common import ( 27 ParsableSql, 28 make_python_env, 29 parse_dependencies, 30 parse_strings_with_macro_refs, 31 single_value_or_tuple, 32 sorted_python_env_payloads, 33 validate_extra_and_required_fields, 34) 35from sqlmesh.core.model.meta import ModelMeta 36from sqlmesh.core.model.kind import ( 37 ExternalKind, 38 ModelKindName, 39 SeedKind, 40 ModelKind, 41 FullKind, 42 create_model_kind, 43 CustomKind, 44) 45from sqlmesh.core.model.seed import CsvSeedReader, Seed, create_seed 46from sqlmesh.core.renderer import ExpressionRenderer, QueryRenderer 47from sqlmesh.core.signal import SignalRegistry 48from sqlmesh.utils import columns_to_types_all_known, str_to_bool, UniqueKeyDict 49from sqlmesh.utils.cron import CroniterCache 50from sqlmesh.utils.date import TimeLike, make_inclusive, to_datetime, to_time_column 51from sqlmesh.utils.errors import ConfigError, SQLMeshError, raise_config_error, PythonModelEvalError 52from sqlmesh.utils.hashing import hash_data 53from sqlmesh.utils.jinja import JinjaMacroRegistry, extract_macro_references_and_variables 54from sqlmesh.utils.pydantic import PydanticModel, PRIVATE_FIELDS 55from sqlmesh.utils.metaprogramming import ( 56 Executable, 57 SqlValue, 58 build_env, 59 prepare_env, 60 serialize_env, 61 format_evaluated_code_exception, 62) 63 64if t.TYPE_CHECKING: 65 from sqlglot.dialects.dialect import DialectType 66 from sqlmesh.core.node import _Node 67 from sqlmesh.core._typing import Self, TableName, SessionProperties 68 from sqlmesh.core.context import ExecutionContext 69 from sqlmesh.core.engine_adapter import EngineAdapter 70 from sqlmesh.core.engine_adapter._typing import QueryOrDF 71 from sqlmesh.core.engine_adapter.shared import DataObjectType 72 from sqlmesh.core.linter.rule import Rule 73 from sqlmesh.core.snapshot import DeployabilityIndex, Node, Snapshot 74 from sqlmesh.utils.jinja import MacroReference 75 76 77logger = logging.getLogger(__name__) 78 79 80PROPERTIES = {"physical_properties", "session_properties", "virtual_properties"} 81 82RUNTIME_RENDERED_MODEL_FIELDS = { 83 "audits", 84 "signals", 85 "merge_filter", 86} | PROPERTIES 87 88CRON_SHORTCUTS = { 89 "@midnight", 90 "@hourly", 91 "@daily", 92 "@weekly", 93 "@monthly", 94 "@yearly", 95 "@annually", 96} 97 98 99class _Model(ModelMeta, frozen=True): 100 """Model is the core abstraction for user defined datasets. 101 102 A model consists of logic that fetches the data (a SQL query, a Python script or a seed) and metadata 103 associated with it. Models can be run on arbitrary cadences and support incremental or full refreshes. 104 Models can also be materialized into physical tables or shared across other models as temporary views. 105 106 Example: 107 MODEL ( 108 name sushi.order_items, 109 owner jen, 110 cron '@daily', 111 start '2020-01-01', 112 partitioned_by ds 113 ); 114 115 @DEF(var, 'my_var'); 116 117 SELECT 118 1 AS column_a # my first column, 119 @var AS my_column # my second column, 120 ; 121 122 Args: 123 name: The name of the model, which is of the form [catalog].[db].table. 124 The catalog and db are optional. 125 dialect: The SQL dialect that the model's query is written in. By default, 126 this is assumed to be the dialect of the context. 127 owner: The owner of the model. 128 cron: A cron string specifying how often the model should be refreshed, leveraging the 129 [croniter](https://github.com/kiorky/croniter) library. 130 description: The optional model description. 131 stamp: An optional arbitrary string sequence used to create new model versions without making 132 changes to any of the functional components of the definition. 133 start: The earliest date that the model will be backfilled for. If this is None, 134 then the date is inferred by taking the most recent start date of its ancestors. 135 The start date can be a static datetime or a relative datetime like "1 year ago" 136 end: The date that the model will be backfilled up until. Follows the same syntax as 'start', 137 should be omitted if there is no end date. 138 lookback: The number of previous incremental intervals in the lookback window. 139 table_format: The table format used to manage the physical table files defined by `storage_format`, only applicable in certain engines. 140 (eg, 'iceberg', 'delta', 'hudi') 141 storage_format: The storage format used to store the physical table, only applicable in certain engines. 142 (eg. 'parquet', 'orc') 143 partitioned_by: The partition columns or engine specific expressions, only applicable in certain engines. (eg. (ds, hour)) 144 clustered_by: The cluster columns or engine specific expressions, only applicable in certain engines. (eg. (ds, hour)) 145 python_env: Dictionary containing all global variables needed to render the model's macros. 146 mapping_schema: The schema of table names to column and types. 147 extract_dependencies_from_query: Whether to extract additional dependencies from the rendered model's query. 148 physical_schema_override: The desired physical schema name override. 149 """ 150 151 python_env: t.Dict[str, Executable] = {} 152 jinja_macros: JinjaMacroRegistry = JinjaMacroRegistry() 153 audit_definitions: t.Dict[str, ModelAudit] = {} 154 mapping_schema: t.Dict[str, t.Any] = {} 155 extract_dependencies_from_query: bool = True 156 pre_statements_: t.Optional[t.List[ParsableSql]] = Field(default=None, alias="pre_statements") 157 post_statements_: t.Optional[t.List[ParsableSql]] = Field(default=None, alias="post_statements") 158 on_virtual_update_: t.Optional[t.List[ParsableSql]] = Field( 159 default=None, alias="on_virtual_update" 160 ) 161 162 _full_depends_on: t.Optional[t.Set[str]] = None 163 _statement_renderer_cache: t.Dict[int, ExpressionRenderer] = {} 164 _is_metadata_only_change_cache: t.Dict[int, bool] = {} 165 166 _expressions_validator = ParsableSql.validator() 167 168 def __getstate__(self) -> t.Dict[t.Any, t.Any]: 169 state = super().__getstate__() 170 private = state[PRIVATE_FIELDS] 171 private["_statement_renderer_cache"] = {} 172 return state 173 174 def copy(self, **kwargs: t.Any) -> Self: 175 model = super().copy(**kwargs) 176 model._statement_renderer_cache = {} 177 return model 178 179 def render( 180 self, 181 *, 182 context: ExecutionContext, 183 start: t.Optional[TimeLike] = None, 184 end: t.Optional[TimeLike] = None, 185 execution_time: t.Optional[TimeLike] = None, 186 **kwargs: t.Any, 187 ) -> t.Iterator[QueryOrDF]: 188 """Renders the content of this model in a form of either a SELECT query, executing which the data for this model can 189 be fetched, or a dataframe object which contains the data itself. 190 191 The type of the returned object (query or dataframe) depends on whether the model was sourced from a SQL query, 192 a Python script or a pre-built dataset (seed). 193 194 Args: 195 context: The execution context used for fetching data. 196 start: The start date/time of the run. 197 end: The end date/time of the run. 198 execution_time: The date/time time reference to use for execution time. 199 200 Returns: 201 A generator which yields either a query object or one of the supported dataframe objects. 202 """ 203 yield self.render_query_or_raise( 204 start=start, 205 end=end, 206 execution_time=execution_time, 207 snapshots=context.snapshots, 208 deployability_index=context.deployability_index, 209 engine_adapter=context.engine_adapter, 210 **kwargs, 211 ) 212 213 def render_definition( 214 self, 215 include_python: bool = True, 216 include_defaults: bool = False, 217 render_query: bool = False, 218 ) -> t.List[exp.Expr]: 219 """Returns the original list of sql expressions comprising the model definition. 220 221 Args: 222 include_python: Whether or not to include Python code in the rendered definition. 223 """ 224 expressions = [] 225 comment = None 226 for field_name, field_info in ModelMeta.all_field_infos().items(): 227 field_value = getattr(self, field_name) 228 229 if (include_defaults and field_value) or field_value != field_info.default: 230 if field_name == "description": 231 comment = field_value 232 elif field_name == "kind": 233 expressions.append( 234 exp.Property( 235 this="kind", 236 value=field_value.to_expression(dialect=self.dialect), 237 ) 238 ) 239 elif field_name == "name": 240 expressions.append( 241 exp.Property( 242 this=field_name, 243 value=exp.to_table(field_value, dialect=self.dialect), 244 ) 245 ) 246 elif field_name not in ("default_catalog", "enabled", "ignored_rules_"): 247 expressions.append( 248 exp.Property( 249 this=field_info.alias or field_name, 250 value=META_FIELD_CONVERTER.get(field_name, exp.to_identifier)( 251 field_value 252 ), 253 ) 254 ) 255 256 model = d.Model(expressions=expressions) 257 model.comments = [comment] if comment else None 258 259 jinja_expressions = [] 260 python_expressions = [] 261 if include_python: 262 python_env = d.PythonCode(expressions=sorted_python_env_payloads(self.python_env)) 263 if python_env.expressions: 264 python_expressions.append(python_env) 265 266 jinja_expressions = self.jinja_macros.to_expressions() 267 268 return [ 269 model, 270 *python_expressions, 271 *jinja_expressions, 272 ] 273 274 def render_query( 275 self, 276 *, 277 start: t.Optional[TimeLike] = None, 278 end: t.Optional[TimeLike] = None, 279 execution_time: t.Optional[TimeLike] = None, 280 snapshots: t.Optional[t.Dict[str, Snapshot]] = None, 281 table_mapping: t.Optional[t.Dict[str, str]] = None, 282 expand: t.Iterable[str] = tuple(), 283 deployability_index: t.Optional[DeployabilityIndex] = None, 284 engine_adapter: t.Optional[EngineAdapter] = None, 285 **kwargs: t.Any, 286 ) -> t.Optional[exp.Query]: 287 """Renders a model's query, expanding macros with provided kwargs, and optionally expanding referenced models. 288 289 Args: 290 start: The start datetime to render. Defaults to epoch start. 291 end: The end datetime to render. Defaults to epoch start. 292 execution_time: The date/time time reference to use for execution time. 293 snapshots: All upstream snapshots (by name) to use for expansion and mapping of physical locations. 294 table_mapping: Table mapping of physical locations. Takes precedence over snapshot mappings. 295 expand: Expand referenced models as subqueries. This is used to bypass backfills when running queries 296 that depend on materialized tables. Model definitions are inlined and can thus be run end to 297 end on the fly. 298 deployability_index: Determines snapshots that are deployable in the context of this render. 299 kwargs: Additional kwargs to pass to the renderer. 300 301 Returns: 302 The rendered expression. 303 """ 304 return exp.select( 305 *( 306 exp.cast(exp.Null(), column_type, copy=False).as_(name, copy=False, quoted=True) 307 for name, column_type in (self.columns_to_types or {}).items() 308 ), 309 copy=False, 310 ).from_(exp.values([tuple([1])], alias="t", columns=["dummy"]), copy=False) 311 312 def render_query_or_raise( 313 self, 314 *, 315 start: t.Optional[TimeLike] = None, 316 end: t.Optional[TimeLike] = None, 317 execution_time: t.Optional[TimeLike] = None, 318 snapshots: t.Optional[t.Dict[str, Snapshot]] = None, 319 table_mapping: t.Optional[t.Dict[str, str]] = None, 320 expand: t.Iterable[str] = tuple(), 321 deployability_index: t.Optional[DeployabilityIndex] = None, 322 engine_adapter: t.Optional[EngineAdapter] = None, 323 **kwargs: t.Any, 324 ) -> exp.Query: 325 """Same as `render_query()` but raises an exception if the query can't be rendered. 326 327 Args: 328 start: The start datetime to render. Defaults to epoch start. 329 end: The end datetime to render. Defaults to epoch start. 330 execution_time: The date/time time reference to use for execution time. 331 snapshots: All upstream snapshots (by model name) to use for expansion and mapping of physical locations. 332 table_mapping: Table mapping of physical locations. Takes precedence over snapshot mappings. 333 expand: Expand referenced models as subqueries. This is used to bypass backfills when running queries 334 that depend on materialized tables. Model definitions are inlined and can thus be run end to 335 end on the fly. 336 deployability_index: Determines snapshots that are deployable in the context of this render. 337 kwargs: Additional kwargs to pass to the renderer. 338 339 Returns: 340 The rendered expression. 341 """ 342 query = self.render_query( 343 start=start, 344 end=end, 345 execution_time=execution_time, 346 snapshots=snapshots, 347 table_mapping=table_mapping, 348 expand=expand, 349 deployability_index=deployability_index, 350 engine_adapter=engine_adapter, 351 **kwargs, 352 ) 353 if query is None: 354 raise SQLMeshError(f"Failed to render query for model '{self.name}'.") 355 return query 356 357 def render_pre_statements( 358 self, 359 *, 360 start: t.Optional[TimeLike] = None, 361 end: t.Optional[TimeLike] = None, 362 execution_time: t.Optional[TimeLike] = None, 363 snapshots: t.Optional[t.Collection[Snapshot]] = None, 364 expand: t.Iterable[str] = tuple(), 365 deployability_index: t.Optional[DeployabilityIndex] = None, 366 engine_adapter: t.Optional[EngineAdapter] = None, 367 inside_transaction: t.Optional[bool] = True, 368 **kwargs: t.Any, 369 ) -> t.List[exp.Expr]: 370 """Renders pre-statements for a model. 371 372 Pre-statements are statements that preceded the model's SELECT query. 373 374 Args: 375 start: The start datetime to render. Defaults to epoch start. 376 end: The end datetime to render. Defaults to epoch start. 377 execution_time: The date/time time reference to use for execution time. 378 snapshots: All upstream snapshots (by model name) to use for expansion and mapping of physical locations. 379 expand: Expand referenced models as subqueries. This is used to bypass backfills when running queries 380 that depend on materialized tables. Model definitions are inlined and can thus be run end to 381 end on the fly. 382 deployability_index: Determines snapshots that are deployable in the context of this render. 383 kwargs: Additional kwargs to pass to the renderer. 384 385 Returns: 386 The list of rendered expressions. 387 """ 388 return self._render_statements( 389 [ 390 stmt 391 for stmt in self.pre_statements 392 if stmt.args.get("transaction", True) == inside_transaction 393 ], 394 start=start, 395 end=end, 396 execution_time=execution_time, 397 snapshots=snapshots, 398 expand=expand, 399 deployability_index=deployability_index, 400 engine_adapter=engine_adapter, 401 **kwargs, 402 ) 403 404 def render_post_statements( 405 self, 406 *, 407 start: t.Optional[TimeLike] = None, 408 end: t.Optional[TimeLike] = None, 409 execution_time: t.Optional[TimeLike] = None, 410 snapshots: t.Optional[t.Dict[str, Snapshot]] = None, 411 expand: t.Iterable[str] = tuple(), 412 deployability_index: t.Optional[DeployabilityIndex] = None, 413 engine_adapter: t.Optional[EngineAdapter] = None, 414 inside_transaction: t.Optional[bool] = True, 415 **kwargs: t.Any, 416 ) -> t.List[exp.Expr]: 417 """Renders post-statements for a model. 418 419 Post-statements are statements that follow after the model's SELECT query. 420 421 Args: 422 start: The start datetime to render. Defaults to epoch start. 423 end: The end datetime to render. Defaults to epoch start. 424 execution_time: The date/time time reference to use for execution time. 425 snapshots: All upstream snapshots (by model name) to use for expansion and mapping of physical locations. 426 expand: Expand referenced models as subqueries. This is used to bypass backfills when running queries 427 that depend on materialized tables. Model definitions are inlined and can thus be run end to 428 end on the fly. 429 deployability_index: Determines snapshots that are deployable in the context of this render. 430 inside_transaction: Whether to render hooks with transaction=True (inside) or transaction=False (outside). 431 kwargs: Additional kwargs to pass to the renderer. 432 433 Returns: 434 The list of rendered expressions. 435 """ 436 return self._render_statements( 437 [ 438 stmt 439 for stmt in self.post_statements 440 if stmt.args.get("transaction", True) == inside_transaction 441 ], 442 start=start, 443 end=end, 444 execution_time=execution_time, 445 snapshots=snapshots, 446 expand=expand, 447 deployability_index=deployability_index, 448 engine_adapter=engine_adapter, 449 **kwargs, 450 ) 451 452 def render_on_virtual_update( 453 self, 454 *, 455 start: t.Optional[TimeLike] = None, 456 end: t.Optional[TimeLike] = None, 457 execution_time: t.Optional[TimeLike] = None, 458 snapshots: t.Optional[t.Dict[str, Snapshot]] = None, 459 expand: t.Iterable[str] = tuple(), 460 deployability_index: t.Optional[DeployabilityIndex] = None, 461 engine_adapter: t.Optional[EngineAdapter] = None, 462 **kwargs: t.Any, 463 ) -> t.List[exp.Expr]: 464 return self._render_statements( 465 self.on_virtual_update, 466 start=start, 467 end=end, 468 execution_time=execution_time, 469 snapshots=snapshots, 470 expand=expand, 471 deployability_index=deployability_index, 472 engine_adapter=engine_adapter, 473 **kwargs, 474 ) 475 476 def render_audit_query( 477 self, 478 audit: Audit, 479 *, 480 start: t.Optional[TimeLike] = None, 481 end: t.Optional[TimeLike] = None, 482 execution_time: t.Optional[TimeLike] = None, 483 snapshots: t.Optional[t.Dict[str, Snapshot]] = None, 484 deployability_index: t.Optional[DeployabilityIndex] = None, 485 **kwargs: t.Any, 486 ) -> exp.Query: 487 from sqlmesh.core.snapshot import DeployabilityIndex 488 489 deployability_index = deployability_index or DeployabilityIndex.all_deployable() 490 snapshot = (snapshots or {}).get(self.fqn) 491 492 this_model = kwargs.pop("this_model", None) or ( 493 snapshot.table_name(deployability_index.is_deployable(snapshot)) 494 if snapshot 495 else self.fqn 496 ) 497 498 columns_to_types: t.Optional[t.Dict[str, t.Any]] = None 499 if "engine_adapter" in kwargs: 500 try: 501 columns_to_types = kwargs["engine_adapter"].columns(this_model) 502 except Exception: 503 pass 504 505 if self.time_column: 506 low, high = [ 507 self.convert_to_time_column(dt, columns_to_types) 508 for dt in make_inclusive(start or c.EPOCH, end or c.EPOCH, self.dialect) 509 ] 510 where = self.time_column.column.between(low, high) 511 else: 512 where = None 513 514 # The model's name is already normalized, but in case of snapshots we also prepend a 515 # case-sensitive physical schema name, so we quote here to ensure that we won't have 516 # a broken schema reference after the resulting query is normalized in `render`. 517 quoted_model_name = quote_identifiers( 518 exp.to_table(this_model, dialect=self.dialect), dialect=self.dialect 519 ) 520 521 query_renderer = QueryRenderer( 522 audit.query, 523 audit.dialect or self.dialect, 524 audit.macro_definitions, 525 path=audit._path or Path(), 526 jinja_macro_registry=audit.jinja_macros, 527 python_env=self.python_env, 528 only_execution_time=self.kind.only_execution_time, 529 default_catalog=self.default_catalog, 530 ) 531 532 rendered_query = query_renderer.render( 533 start=start, 534 end=end, 535 execution_time=execution_time, 536 snapshots=snapshots, 537 deployability_index=deployability_index, 538 **{ 539 **audit.defaults, 540 "this_model": exp.select("*").from_(quoted_model_name).where(where).subquery() 541 if where is not None 542 else quoted_model_name, 543 **kwargs, 544 }, # type: ignore 545 ) 546 547 if rendered_query is None: 548 raise SQLMeshError( 549 f"Failed to render query for audit '{audit.name}', model '{self.name}'." 550 ) 551 552 return rendered_query 553 554 @property 555 def pre_statements(self) -> t.List[exp.Expr]: 556 return self._get_parsed_statements("pre_statements_") 557 558 @property 559 def post_statements(self) -> t.List[exp.Expr]: 560 return self._get_parsed_statements("post_statements_") 561 562 @property 563 def on_virtual_update(self) -> t.List[exp.Expr]: 564 return self._get_parsed_statements("on_virtual_update_") 565 566 @property 567 def macro_definitions(self) -> t.List[d.MacroDef]: 568 """All macro definitions from the list of expressions.""" 569 return [ 570 s 571 for s in self.pre_statements + self.post_statements + self.on_virtual_update 572 if isinstance(s, d.MacroDef) 573 ] 574 575 def _get_parsed_statements(self, attr_name: str) -> t.List[exp.Expr]: 576 value = getattr(self, attr_name) 577 if not value: 578 return [] 579 result = [] 580 for v in value: 581 parsed = v.parse(self.dialect) 582 if getattr(v, "transaction", None) is not None: 583 parsed.set("transaction", v.transaction) 584 if not isinstance(parsed, exp.Semicolon): 585 result.append(parsed) 586 return result 587 588 def _render_statements( 589 self, 590 statements: t.Iterable[exp.Expr], 591 **kwargs: t.Any, 592 ) -> t.List[exp.Expr]: 593 rendered = ( 594 self._statement_renderer(statement).render(**kwargs) 595 for statement in statements 596 if not isinstance(statement, d.MacroDef) 597 ) 598 return [r for expressions in rendered if expressions for r in expressions] 599 600 def _statement_renderer(self, expression: exp.Expr) -> ExpressionRenderer: 601 expression_key = id(expression) 602 if expression_key not in self._statement_renderer_cache: 603 self._statement_renderer_cache[expression_key] = ExpressionRenderer( 604 expression, 605 self.dialect, 606 self.macro_definitions, 607 path=self._path, 608 jinja_macro_registry=self.jinja_macros, 609 python_env=self.python_env, 610 only_execution_time=False, 611 default_catalog=self.default_catalog, 612 model=self, 613 ) 614 return self._statement_renderer_cache[expression_key] 615 616 def render_signals( 617 self, 618 *, 619 start: t.Optional[TimeLike] = None, 620 end: t.Optional[TimeLike] = None, 621 execution_time: t.Optional[TimeLike] = None, 622 ) -> t.List[t.Dict[str, str | int | float | bool]]: 623 """Renders external; signals defined for this model. 624 625 Args: 626 start: The start datetime to render. Defaults to epoch start. 627 end: The end datetime to render. Defaults to epoch start. 628 execution_time: The date/time time reference to use for execution time. 629 630 Returns: 631 The list of rendered expressions. 632 """ 633 634 def _render(e: exp.Expr) -> str | int | float | bool: 635 rendered_exprs = ( 636 self._create_renderer(e).render(start=start, end=end, execution_time=execution_time) 637 or [] 638 ) 639 if len(rendered_exprs) != 1: 640 raise SQLMeshError(f"Expected one expression but got {len(rendered_exprs)}") 641 642 rendered = rendered_exprs[0] 643 if rendered.is_int: 644 return int(rendered.this) 645 if rendered.is_number: 646 return float(rendered.this) 647 if isinstance(rendered, (exp.Literal, exp.Boolean)): 648 return rendered.this 649 return rendered.sql(dialect=self.dialect) 650 651 # airflow only 652 return [ 653 {k: _render(v) for k, v in signal.items()} for name, signal in self.signals if not name 654 ] 655 656 def render_signal_calls(self) -> EvaluatableSignals: 657 python_env = self.python_env 658 env = prepare_env(python_env) 659 signals_to_kwargs = { 660 name: { 661 k: seq_get(self._create_renderer(v).render() or [], 0) for k, v in kwargs.items() 662 } 663 for name, kwargs in self.signals 664 if name 665 } 666 667 return EvaluatableSignals( 668 signals_to_kwargs=signals_to_kwargs, 669 python_env=python_env, 670 prepared_python_env=env, 671 ) 672 673 def render_merge_filter( 674 self, 675 *, 676 start: t.Optional[TimeLike] = None, 677 end: t.Optional[TimeLike] = None, 678 execution_time: t.Optional[TimeLike] = None, 679 ) -> t.Optional[exp.Expr]: 680 if self.merge_filter is None: 681 return None 682 rendered_exprs = ( 683 self._create_renderer(self.merge_filter).render( 684 start=start, end=end, execution_time=execution_time 685 ) 686 or [] 687 ) 688 if len(rendered_exprs) != 1: 689 raise SQLMeshError(f"Expected one expression but got {len(rendered_exprs)}") 690 return rendered_exprs[0].transform(d.replace_merge_table_aliases, dialect=self.dialect) 691 692 def _render_properties( 693 self, properties: t.Dict[str, exp.Expr] | SessionProperties, **render_kwargs: t.Any 694 ) -> t.Dict[str, t.Any]: 695 def _render(expression: exp.Expr) -> exp.Expr | None: 696 # note: we use the _statement_renderer instead of _create_renderer because it sets model_fqn which 697 # in turn makes @this_model available in the evaluation context 698 rendered_exprs = self._statement_renderer(expression).render(**render_kwargs) 699 700 # Inform instead of raising for cases where a property is conditionally assigned 701 if not rendered_exprs or rendered_exprs[0].sql().lower() in {"none", "null"}: 702 logger.info( 703 f"Rendering '{expression.sql(dialect=self.dialect)}' did not return an expression" 704 ) 705 return None 706 707 if len(rendered_exprs) != 1: 708 raise SQLMeshError( 709 f"Expected one result when rendering '{expression.sql(dialect=self.dialect)}' but got {len(rendered_exprs)}" 710 ) 711 712 return rendered_exprs[0] 713 714 return { 715 k: rendered 716 for k, v in properties.items() 717 if (rendered := (_render(v) if isinstance(v, exp.Expr) else v)) 718 } 719 720 def render_physical_properties(self, **render_kwargs: t.Any) -> t.Dict[str, t.Any]: 721 rendered = self._render_properties(properties=self.physical_properties, **render_kwargs) 722 723 # Some engines (e.g. StarRocks) accept properties whose values reference other models and 724 # need the physical table name rather than the logical view SQLMesh exposes. Resolve those. 725 engine_adapter = render_kwargs.get("engine_adapter") 726 resolve_keys: t.FrozenSet[str] = getattr( 727 engine_adapter, "RESOLVE_TABLE_REFS_IN_PHYSICAL_PROPERTIES", frozenset() 728 ) 729 keys_to_resolve = [key for key in resolve_keys if key in rendered] 730 if keys_to_resolve: 731 # Local import: sqlmesh.core.snapshot.definition imports _Model, so importing 732 # to_table_mapping at module scope would be circular. 733 from sqlmesh.core.snapshot.definition import to_table_mapping 734 735 table_mapping = to_table_mapping( 736 (render_kwargs.get("snapshots") or {}).values(), 737 render_kwargs.get("deployability_index"), 738 ) 739 for key in keys_to_resolve: 740 rendered[key] = _resolve_model_refs_to_physical_tables( 741 rendered[key], table_mapping, self.dialect 742 ) 743 744 return rendered 745 746 def render_virtual_properties(self, **render_kwargs: t.Any) -> t.Dict[str, t.Any]: 747 return self._render_properties(properties=self.virtual_properties, **render_kwargs) 748 749 def render_session_properties(self, **render_kwargs: t.Any) -> t.Dict[str, t.Any]: 750 return self._render_properties(properties=self.session_properties, **render_kwargs) 751 752 def _create_renderer(self, expression: exp.Expr) -> ExpressionRenderer: 753 return ExpressionRenderer( 754 expression, 755 self.dialect, 756 [], 757 path=self._path, 758 jinja_macro_registry=self.jinja_macros, 759 python_env=self.python_env, 760 only_execution_time=False, 761 quote_identifiers=False, 762 ) 763 764 def ctas_query(self, **render_kwarg: t.Any) -> exp.Query: 765 """Return a dummy query to do a CTAS. 766 767 If a model's column types are unknown, the only way to create the table is to 768 run the fully expanded query. This can be expensive so we add a WHERE FALSE to all 769 SELECTS and hopefully the optimizer is smart enough to not do anything. 770 771 Args: 772 render_kwarg: Additional kwargs to pass to the renderer. 773 Return: 774 The mocked out ctas query. 775 """ 776 query = self.render_query_or_raise(**render_kwarg).limit(0) 777 778 for select_or_set_op in query.find_all(exp.Select, exp.SetOperation): 779 if isinstance(select_or_set_op, exp.Select) and select_or_set_op.args.get("from_"): 780 select_or_set_op.where(exp.false(), copy=False) 781 782 if self.managed_columns: 783 query.select( 784 *[ 785 exp.alias_(exp.cast(exp.Null(), to=col_type), col) 786 for col, col_type in self.managed_columns.items() 787 if col not in query.named_selects 788 ], 789 append=True, 790 copy=False, 791 ) 792 return query 793 794 def text_diff(self, other: Node, rendered: bool = False) -> str: 795 """Produce a text diff against another node. 796 797 Args: 798 other: The node to diff against. 799 rendered: Whether the diff should compare raw vs rendered models 800 801 Returns: 802 A unified text diff showing additions and deletions. 803 """ 804 if not isinstance(other, _Model): 805 raise SQLMeshError( 806 f"Cannot diff model '{self.name} against a non-model node '{other.name}'" 807 ) 808 809 text_diff = d.text_diff( 810 self.render_definition(render_query=rendered), 811 other.render_definition(render_query=rendered), 812 self.dialect, 813 other.dialect, 814 ).strip() 815 816 if not text_diff and not rendered: 817 text_diff = d.text_diff( 818 self.render_definition(render_query=True), 819 other.render_definition(render_query=True), 820 self.dialect, 821 other.dialect, 822 ).strip() 823 824 return text_diff 825 826 def set_time_format(self, default_time_format: str = c.DEFAULT_TIME_COLUMN_FORMAT) -> None: 827 """Sets the default time format for a model. 828 829 Args: 830 default_time_format: A python time format used as the default format when none is provided. 831 """ 832 if not self.time_column: 833 return 834 835 if self.time_column.format: 836 # Transpile the time column format into the generic dialect 837 formatted_time = format_time( 838 self.time_column.format, 839 d.Dialect.get_or_raise(self.dialect).TIME_MAPPING, 840 ) 841 assert formatted_time is not None 842 self.time_column.format = formatted_time 843 else: 844 self.time_column.format = default_time_format 845 846 def convert_to_time_column( 847 self, time: TimeLike, columns_to_types: t.Optional[t.Dict[str, exp.DataType]] = None 848 ) -> exp.Expr: 849 """Convert a TimeLike object to the same time format and type as the model's time column.""" 850 if self.time_column: 851 if columns_to_types is None: 852 columns_to_types = self.columns_to_types_or_raise 853 854 if self.time_column.column.name not in columns_to_types: 855 raise ConfigError( 856 f"Time column '{self.time_column.column.sql(dialect=self.dialect)}' not found in model '{self.name}'." 857 ) 858 859 time_column_type = columns_to_types[self.time_column.column.name] 860 861 return to_time_column( 862 time, 863 time_column_type, 864 self.dialect, 865 self.time_column.format, 866 ) 867 return exp.convert(time) 868 869 def set_mapping_schema(self, schema: t.Dict) -> None: 870 self.mapping_schema.clear() 871 self.mapping_schema.update(schema) 872 873 def update_schema(self, schema: MappingSchema) -> None: 874 """Updates the schema for this model's dependencies based on the given mapping schema.""" 875 for dep in self.depends_on: 876 table = exp.to_table(dep) 877 mapping_schema = schema.find(table) 878 879 if mapping_schema: 880 nested_set( 881 self.mapping_schema, 882 tuple(part.sql(copy=False) for part in table.parts), 883 {col: dtype.sql(dialect=self.dialect) for col, dtype in mapping_schema.items()}, 884 ) 885 886 @property 887 def depends_on(self) -> t.Set[str]: 888 """All of the upstream dependencies referenced in the model's query, excluding self references. 889 890 Returns: 891 A list of all the upstream table names. 892 """ 893 return self.full_depends_on - {self.fqn} 894 895 @property 896 def columns_to_types(self) -> t.Optional[t.Dict[str, exp.DataType]]: 897 """Returns the mapping of column names to types of this model.""" 898 if self.columns_to_types_ is None: 899 return None 900 return {**self.columns_to_types_, **self.managed_columns} 901 902 @property 903 def columns_to_types_or_raise(self) -> t.Dict[str, exp.DataType]: 904 """Returns the mapping of column names to types of this model or raise if not available.""" 905 columns_to_types = self.columns_to_types 906 if columns_to_types is None: 907 raise SQLMeshError(f"Column information is not available for model '{self.name}'") 908 return columns_to_types 909 910 @property 911 def annotated(self) -> bool: 912 """Checks if all column projection types of this model are known.""" 913 if self.columns_to_types is None: 914 return False 915 columns_to_types = { 916 k: v for k, v in self.columns_to_types.items() if k not in self.managed_columns 917 } 918 if not columns_to_types: 919 return False 920 return columns_to_types_all_known(columns_to_types) 921 922 @property 923 def sorted_python_env(self) -> t.List[t.Tuple[str, Executable]]: 924 """Returns the python env sorted by executable kind and then var name.""" 925 return sorted(self.python_env.items(), key=lambda x: (x[1].kind, x[0])) 926 927 @property 928 def view_name(self) -> str: 929 return self.fully_qualified_table.name 930 931 @property 932 def schema_name(self) -> str: 933 return self.fully_qualified_table.db or c.DEFAULT_SCHEMA 934 935 @property 936 def physical_schema(self) -> str: 937 return self.physical_schema_override or f"{c.SQLMESH}__{self.schema_name}" 938 939 @property 940 def is_sql(self) -> bool: 941 return False 942 943 @property 944 def is_python(self) -> bool: 945 return False 946 947 @property 948 def is_seed(self) -> bool: 949 return False 950 951 @property 952 def depends_on_self(self) -> bool: 953 return self.fqn in self.full_depends_on 954 955 @property 956 def forward_only(self) -> bool: 957 return getattr(self.kind, "forward_only", False) 958 959 @property 960 def disable_restatement(self) -> bool: 961 return getattr(self.kind, "disable_restatement", False) 962 963 @property 964 def auto_restatement_intervals(self) -> t.Optional[int]: 965 return getattr(self.kind, "auto_restatement_intervals", None) 966 967 @property 968 def auto_restatement_cron(self) -> t.Optional[str]: 969 return getattr(self.kind, "auto_restatement_cron", None) 970 971 def auto_restatement_croniter(self, value: TimeLike) -> CroniterCache: 972 cron = self.auto_restatement_cron 973 if cron is None: 974 raise SQLMeshError("Auto restatement cron is not set.") 975 return CroniterCache(cron, value) 976 977 @property 978 def wap_supported(self) -> bool: 979 return self.kind.is_materialized and (self.storage_format or "").lower() == "iceberg" 980 981 def validate_definition(self) -> None: 982 """Validates the model's definition. 983 984 Raises: 985 ConfigError 986 """ 987 988 for field in ("partitioned_by", "clustered_by"): 989 values = getattr(self, field) 990 991 if values: 992 values = [ 993 col.name 994 for expr in values 995 if not ( 996 field == "clustered_by" 997 and (self.dialect or "").lower() == "databricks" 998 and isinstance(expr, exp.Var) 999 and expr.name.upper() in c.LIQUID_CLUSTERING_KEYWORDS 1000 ) 1001 for col in t.cast( 1002 exp.Expr, exp.maybe_parse(expr, dialect=self.dialect) 1003 ).find_all(exp.Column) 1004 ] 1005 1006 unique_keys = set(values) 1007 1008 if len(values) != len(unique_keys): 1009 raise_config_error( 1010 f"All keys in '{field}' must be unique in the model definition", 1011 self._path, 1012 ) 1013 1014 columns_to_types = self.columns_to_types 1015 if columns_to_types is not None: 1016 missing_keys = unique_keys - set(columns_to_types) 1017 if missing_keys: 1018 missing_keys_str = ", ".join(f"'{k}'" for k in sorted(missing_keys)) 1019 raise_config_error( 1020 f"{field} keys [{missing_keys_str}] are missing in the model definition", 1021 self._path, 1022 ) 1023 1024 if self.kind.is_incremental_by_time_range and not self.time_column: 1025 raise_config_error( 1026 "Incremental by time range models must have a time_column field", 1027 self._path, 1028 ) 1029 1030 if ( 1031 self.kind.is_incremental_unmanaged 1032 and getattr(self.kind, "insert_overwrite", False) 1033 and not self.partitioned_by_ 1034 ): 1035 raise_config_error( 1036 "Unmanaged incremental models with insert / overwrite enabled must specify the partitioned_by field", 1037 self._path, 1038 ) 1039 1040 if self.kind.is_managed: 1041 # TODO: would this sort of logic be better off moved into the Kind? 1042 if self.dialect == "snowflake" and "target_lag" not in self.physical_properties: 1043 raise_config_error( 1044 "Snowflake managed tables must specify the 'target_lag' physical property", 1045 self._path, 1046 ) 1047 1048 if self.physical_version is not None and not self.forward_only: 1049 raise_config_error( 1050 "Pinning a physical version is only supported for forward only models", 1051 self._path, 1052 ) 1053 1054 # The following attributes should be set only for SQL models 1055 if not self.is_sql: 1056 if self.optimize_query: 1057 raise_config_error( 1058 "SQLMesh query optimizer can only be enabled for SQL models", 1059 self._path, 1060 ) 1061 1062 if isinstance(self.kind, CustomKind): 1063 from sqlmesh.core.snapshot.evaluator import get_custom_materialization_type_or_raise 1064 1065 # Will raise if the custom materialization points to an invalid class 1066 get_custom_materialization_type_or_raise(self.kind.materialization) 1067 1068 # Embedded model kind shouldn't have audits 1069 if self.kind.name == ModelKindName.EMBEDDED and self.audits: 1070 raise_config_error( 1071 "Audits are not supported for embedded models", 1072 self._path, 1073 ) 1074 1075 def is_breaking_change(self, previous: Model) -> t.Optional[bool]: 1076 """Determines whether this model is a breaking change in relation to the `previous` model. 1077 1078 Args: 1079 previous: The previous model to compare against. 1080 1081 Returns: 1082 True if this model instance represents a breaking change, False if it's a non-breaking change 1083 and None if the nature of the change can't be determined. 1084 """ 1085 raise NotImplementedError 1086 1087 def is_metadata_only_change(self, other: _Node) -> bool: 1088 if self._is_metadata_only_change_cache.get(id(other), None) is not None: 1089 return self._is_metadata_only_change_cache[id(other)] 1090 1091 is_metadata_change = True 1092 if ( 1093 not isinstance(other, _Model) 1094 or self.metadata_hash == other.metadata_hash 1095 or self._data_hash_values_no_sql != other._data_hash_values_no_sql 1096 ): 1097 is_metadata_change = False 1098 else: 1099 this_statements = [ 1100 s 1101 for s in [*self.pre_statements, *self.post_statements] 1102 if not self._is_metadata_statement(s) 1103 ] 1104 other_statements = [ 1105 s 1106 for s in [*other.pre_statements, *other.post_statements] 1107 if not other._is_metadata_statement(s) 1108 ] 1109 if len(this_statements) != len(other_statements): 1110 is_metadata_change = False 1111 else: 1112 for this_statement, other_statement in zip(this_statements, other_statements): 1113 this_rendered = ( 1114 self._statement_renderer(this_statement).render() or this_statement 1115 ) 1116 other_rendered = ( 1117 other._statement_renderer(other_statement).render() or other_statement 1118 ) 1119 if this_rendered != other_rendered: 1120 is_metadata_change = False 1121 break 1122 1123 self._is_metadata_only_change_cache[id(other)] = is_metadata_change 1124 return is_metadata_change 1125 1126 @property 1127 def data_hash(self) -> str: 1128 """ 1129 Computes the data hash for the node. 1130 1131 Returns: 1132 The data hash for the node. 1133 """ 1134 if self._data_hash is None: 1135 self._data_hash = hash_data(self._data_hash_values) 1136 return self._data_hash 1137 1138 @property 1139 def _data_hash_values(self) -> t.List[str]: 1140 return self._data_hash_values_no_sql + self._data_hash_values_sql 1141 1142 @property 1143 def _data_hash_values_sql(self) -> t.List[str]: 1144 data = [] 1145 1146 for statements in [self.pre_statements_, self.post_statements_]: 1147 for statement in statements or []: 1148 data.append(statement.sql) 1149 1150 return data 1151 1152 @property 1153 def _data_hash_values_no_sql(self) -> t.List[str]: 1154 data = [ 1155 str( # Exclude metadata only macro funcs 1156 [(k, v) for k, v in self.sorted_python_env if not v.is_metadata] 1157 ), 1158 *self.kind.data_hash_values, 1159 self.table_format, 1160 self.storage_format, 1161 str(self.lookback), 1162 *(gen(expr) for expr in (self.partitioned_by or [])), 1163 *(gen(expr) for expr in (self.clustered_by or [])), 1164 self.stamp, 1165 self.physical_schema, 1166 self.physical_version, 1167 self.gateway, 1168 self.interval_unit.value if self.interval_unit is not None else None, 1169 str(self.optimize_query) if self.optimize_query is not None else None, 1170 self.virtual_environment_mode.value, 1171 ] 1172 1173 for column_name, column_type in (self.columns_to_types_ or {}).items(): 1174 data.append(column_name) 1175 data.append(column_type.sql(dialect=self.dialect)) 1176 1177 for key, value in (self.physical_properties or {}).items(): 1178 data.append(key) 1179 data.append(gen(value)) 1180 1181 return data # type: ignore 1182 1183 def _audit_metadata_hash_values(self) -> t.List[str]: 1184 from sqlmesh.core.audit.builtin import BUILT_IN_AUDITS 1185 1186 metadata = [] 1187 1188 for audit_name, audit_args in sorted(self.audits, key=lambda a: a[0]): 1189 metadata.append(audit_name) 1190 if audit_name not in BUILT_IN_AUDITS: 1191 audit = self.audit_definitions[audit_name] 1192 metadata.extend( 1193 [ 1194 audit.query_.sql, 1195 audit.dialect, 1196 str(audit.skip), 1197 str(audit.blocking), 1198 ] 1199 ) 1200 for arg_name, arg_value in audit_args.items(): 1201 metadata.append(arg_name) 1202 metadata.append(gen(arg_value)) 1203 1204 return metadata 1205 1206 def audit_metadata_hash(self) -> str: 1207 return hash_data(self._audit_metadata_hash_values()) 1208 1209 @property 1210 def metadata_hash(self) -> str: 1211 """ 1212 Computes the metadata hash for the node. 1213 1214 Returns: 1215 The metadata hash for the node. 1216 """ 1217 if self._metadata_hash is None: 1218 metadata = [ 1219 self.dialect, 1220 self.owner, 1221 self.description, 1222 json.dumps(self.column_descriptions, sort_keys=True), 1223 self.cron, 1224 self.cron_tz.key if self.cron_tz else None, 1225 str(self.start) if self.start else None, 1226 str(self.end) if self.end else None, 1227 str(self.retention) if self.retention else None, 1228 str(self.batch_size) if self.batch_size is not None else None, 1229 str(self.batch_concurrency) if self.batch_concurrency is not None else None, 1230 json.dumps(self.mapping_schema, sort_keys=True), 1231 *sorted(self.tags), 1232 *sorted(ref.json(sort_keys=True) for ref in self.all_references), 1233 *self.kind.metadata_hash_values, 1234 self.project, 1235 str(self.allow_partials), 1236 gen(self.session_properties_) if self.session_properties_ else None, 1237 *[gen(g) for g in self.grains], 1238 *self._audit_metadata_hash_values(), 1239 json.dumps(self.grants, sort_keys=True) if self.grants else None, 1240 self.grants_target_layer, 1241 ] 1242 1243 for key, value in (self.virtual_properties or {}).items(): 1244 metadata.append(key) 1245 metadata.append(gen(value)) 1246 1247 for signal_name, args in sorted(self.signals, key=lambda x: x[0]): 1248 metadata.append(signal_name) 1249 for k, v in sorted(args.items()): 1250 metadata.append(f"{k}:{gen(v)}") 1251 1252 if self.dbt_node_info: 1253 metadata.append(self.dbt_node_info.json(sort_keys=True)) 1254 1255 metadata.extend(self._additional_metadata) 1256 1257 self._metadata_hash = hash_data(metadata) 1258 return self._metadata_hash 1259 1260 @property 1261 def is_model(self) -> bool: 1262 """Return True if this is a model node""" 1263 return True 1264 1265 @property 1266 def grants_table_type(self) -> DataObjectType: 1267 """Get the table type for grants application (TABLE, VIEW, MATERIALIZED_VIEW). 1268 1269 Returns: 1270 The DataObjectType that should be used when applying grants to this model. 1271 """ 1272 from sqlmesh.core.engine_adapter.shared import DataObjectType 1273 1274 if self.kind.is_view: 1275 if hasattr(self.kind, "materialized") and getattr(self.kind, "materialized", False): 1276 return DataObjectType.MATERIALIZED_VIEW 1277 return DataObjectType.VIEW 1278 if self.kind.is_managed: 1279 return DataObjectType.MANAGED_TABLE 1280 # All other materialized models are tables 1281 return DataObjectType.TABLE 1282 1283 @property 1284 def _additional_metadata(self) -> t.List[str]: 1285 additional_metadata = [] 1286 1287 metadata_only_macros = [(k, v) for k, v in self.sorted_python_env if v.is_metadata] 1288 if metadata_only_macros: 1289 additional_metadata.append(str(metadata_only_macros)) 1290 1291 for statements in [self.pre_statements_, self.post_statements_, self.on_virtual_update_]: 1292 for statement in statements or []: 1293 additional_metadata.append(statement.sql) 1294 1295 return additional_metadata 1296 1297 def _is_metadata_statement(self, statement: exp.Expr) -> bool: 1298 if isinstance(statement, d.MacroDef): 1299 return True 1300 if isinstance(statement, d.MacroFunc): 1301 target_macro = macro.get_registry().get(statement.name) 1302 if target_macro: 1303 return target_macro.metadata_only 1304 target_macro = self.python_env.get(statement.name) 1305 if target_macro: 1306 return bool(target_macro.is_metadata) 1307 return False 1308 1309 @property 1310 def full_depends_on(self) -> t.Set[str]: 1311 if not self.extract_dependencies_from_query: 1312 return self.depends_on_ or set() 1313 if self._full_depends_on is None: 1314 depends_on = self.depends_on_ or set() 1315 1316 query = self.render_query(needs_optimization=False) 1317 if query is not None: 1318 depends_on |= d.find_tables( 1319 query, default_catalog=self.default_catalog, dialect=self.dialect 1320 ) 1321 self._full_depends_on = depends_on 1322 1323 return self._full_depends_on 1324 1325 @property 1326 def partitioned_by(self) -> t.List[exp.Expr]: 1327 """Columns to partition the model by, including the time column if it is not already included.""" 1328 if self.time_column and not self._is_time_column_in_partitioned_by: 1329 # This allows the user to opt out of automatic time_column injection 1330 # by setting `partition_by_time_column false` on the model kind 1331 if ( 1332 hasattr(self.kind, "partition_by_time_column") 1333 and self.kind.partition_by_time_column 1334 ): 1335 return [ 1336 TIME_COL_PARTITION_FUNC.get(self.dialect, lambda x, y: x)( 1337 self.time_column.column, self.columns_to_types 1338 ), 1339 *self.partitioned_by_, 1340 ] 1341 return self.partitioned_by_ 1342 1343 @property 1344 def partition_interval_unit(self) -> t.Optional[IntervalUnit]: 1345 """The interval unit to use for partitioning if applicable.""" 1346 # Only return the interval unit for partitioning if the partitioning 1347 # wasn't explicitly set by the user. Otherwise, the user-provided 1348 # value should always take precedence. 1349 if self.time_column and not self._is_time_column_in_partitioned_by: 1350 return self.interval_unit 1351 return None 1352 1353 @property 1354 def audits_with_args(self) -> t.List[t.Tuple[Audit, t.Dict[str, exp.Expr]]]: 1355 from sqlmesh.core.audit.builtin import BUILT_IN_AUDITS 1356 1357 audits_by_name = {**BUILT_IN_AUDITS, **self.audit_definitions} 1358 audits_with_args = [] 1359 added_audits = set() 1360 1361 for audit_name, audit_args in self.audits: 1362 audits_with_args.append((audits_by_name[audit_name], audit_args.copy())) 1363 added_audits.add(audit_name) 1364 1365 for audit_name in self.audit_definitions: 1366 if audit_name not in added_audits: 1367 audits_with_args.append((audits_by_name[audit_name], {})) 1368 1369 return audits_with_args 1370 1371 @property 1372 def _is_time_column_in_partitioned_by(self) -> bool: 1373 return self.time_column is not None and self.time_column.column in { 1374 col for expr in self.partitioned_by_ for col in expr.find_all(exp.Column) 1375 } 1376 1377 @property 1378 def violated_rules_for_query(self) -> t.Dict[type[Rule], t.Any]: 1379 return {} 1380 1381 1382class SqlModel(_Model): 1383 """The model definition which relies on a SQL query to fetch the data. 1384 1385 Args: 1386 query: The main query representing the model. 1387 pre_statements: The list of SQL statements that precede the model's query. 1388 post_statements: The list of SQL statements that follow after the model's query. 1389 on_virtual_update: The list of SQL statements to be executed after the virtual update. 1390 """ 1391 1392 query_: ParsableSql = Field(alias="query") 1393 source_type: t.Literal["sql"] = "sql" 1394 1395 _columns_to_types: t.Optional[t.Dict[str, exp.DataType]] = None 1396 1397 def __getstate__(self) -> t.Dict[t.Any, t.Any]: 1398 state = super().__getstate__() 1399 state["__dict__"] = state["__dict__"].copy() 1400 # query renderer is very expensive to serialize 1401 state["__dict__"].pop("_query_renderer", None) 1402 state["__dict__"].pop("column_descriptions", None) 1403 private = state[PRIVATE_FIELDS] 1404 private["_columns_to_types"] = None 1405 return state 1406 1407 def copy(self, **kwargs: t.Any) -> Self: 1408 model = super().copy(**kwargs) 1409 model.__dict__.pop("_query_renderer", None) 1410 model.__dict__.pop("column_descriptions", None) 1411 model._columns_to_types = None 1412 if kwargs.get("update", {}).keys() & {"depends_on_", "query"}: 1413 model._full_depends_on = None 1414 return model 1415 1416 @property 1417 def query(self) -> t.Union[exp.Query, d.JinjaQuery, d.MacroFunc]: 1418 parsed_query = self.query_.parse(self.dialect) 1419 return t.cast(t.Union[exp.Query, d.JinjaQuery, d.MacroFunc], parsed_query) 1420 1421 def render_query( 1422 self, 1423 *, 1424 start: t.Optional[TimeLike] = None, 1425 end: t.Optional[TimeLike] = None, 1426 execution_time: t.Optional[TimeLike] = None, 1427 snapshots: t.Optional[t.Dict[str, Snapshot]] = None, 1428 table_mapping: t.Optional[t.Dict[str, str]] = None, 1429 expand: t.Iterable[str] = tuple(), 1430 deployability_index: t.Optional[DeployabilityIndex] = None, 1431 engine_adapter: t.Optional[EngineAdapter] = None, 1432 **kwargs: t.Any, 1433 ) -> t.Optional[exp.Query]: 1434 query = self._query_renderer.render( 1435 start=start, 1436 end=end, 1437 execution_time=execution_time, 1438 snapshots=snapshots, 1439 table_mapping=table_mapping, 1440 expand=expand, 1441 deployability_index=deployability_index, 1442 engine_adapter=engine_adapter, 1443 **kwargs, 1444 ) 1445 1446 return query 1447 1448 def render_definition( 1449 self, 1450 include_python: bool = True, 1451 include_defaults: bool = False, 1452 render_query: bool = False, 1453 ) -> t.List[exp.Expr]: 1454 result: t.List[exp.Expr] = super().render_definition( 1455 include_python=include_python, include_defaults=include_defaults 1456 ) 1457 1458 if render_query: 1459 result.extend(self.render_pre_statements()) 1460 result.append(self.render_query() or self.query) 1461 result.extend(self.render_post_statements()) 1462 if virtual_update := self.render_on_virtual_update(): 1463 result.append(d.VirtualUpdateStatement(expressions=virtual_update)) 1464 else: 1465 result.extend(self.pre_statements) 1466 result.append(self.query) 1467 result.extend(self.post_statements) 1468 if self.on_virtual_update: 1469 result.append(d.VirtualUpdateStatement(expressions=self.on_virtual_update)) 1470 1471 return result 1472 1473 @property 1474 def is_sql(self) -> bool: 1475 return True 1476 1477 @property 1478 def columns_to_types(self) -> t.Optional[t.Dict[str, exp.DataType]]: 1479 if self.columns_to_types_ is not None: 1480 self._columns_to_types = self.columns_to_types_ 1481 elif self._columns_to_types is None: 1482 try: 1483 query = self._query_renderer.render() 1484 except Exception: 1485 logger.exception("Failed to render query for model %s", self.fqn) 1486 return None 1487 1488 if query is None: 1489 return None 1490 1491 unknown = exp.DataType.build("unknown") 1492 1493 columns_to_types = {} 1494 for select in query.selects: 1495 output_name = select.output_name 1496 1497 # If model validation is disabled, we cannot assume that projections 1498 # will have inferrable output names or even that they will be unique 1499 if not output_name or output_name in columns_to_types: 1500 return None 1501 1502 # copy data type because it is used in the engine to build CTAS and other queries 1503 # this can change the parent which will mess up the diffing algo 1504 columns_to_types[output_name] = (select.type or unknown).copy() 1505 1506 self._columns_to_types = columns_to_types 1507 1508 if "*" in self._columns_to_types: 1509 return None 1510 1511 return {**self._columns_to_types, **self.managed_columns} 1512 1513 @cached_property 1514 def column_descriptions(self) -> t.Dict[str, str]: 1515 if self.column_descriptions_ is not None: 1516 return self.column_descriptions_ 1517 1518 query = self.render_query() 1519 if query is None: 1520 return {} 1521 1522 return { 1523 select.alias_or_name: select.comments[-1].strip() 1524 for select in query.selects 1525 if select.comments 1526 } 1527 1528 def set_mapping_schema(self, schema: t.Dict) -> None: 1529 super().set_mapping_schema(schema) 1530 self._on_mapping_schema_set() 1531 1532 def update_schema(self, schema: MappingSchema) -> None: 1533 super().update_schema(schema) 1534 self._on_mapping_schema_set() 1535 1536 def _on_mapping_schema_set(self) -> None: 1537 self._columns_to_types = None 1538 self._query_renderer.update_schema(self.mapping_schema) 1539 1540 def validate_definition(self) -> None: 1541 query = self._query_renderer.render() 1542 if query is None: 1543 if self.depends_on_ is None: 1544 raise_config_error( 1545 "Dependencies must be provided explicitly for models that can be rendered only at runtime", 1546 self._path, 1547 ) 1548 return 1549 1550 if not isinstance(query, exp.Query): 1551 raise_config_error("Missing SELECT query in the model definition", self._path) 1552 1553 projection_list = query.selects 1554 if not projection_list: 1555 raise_config_error("Query missing select statements", self._path) 1556 1557 if self.depends_on_self and not self.annotated: 1558 raise_config_error( 1559 "Self-referencing models require inferrable column types. There are three options available to mitigate this issue: add explicit types to all projections in the outermost SELECT statement, leverage external models (https://sqlmesh.readthedocs.io/en/stable/concepts/models/external_models/), or use the `columns` model attribute (https://sqlmesh.readthedocs.io/en/stable/concepts/models/overview/#columns).", 1560 self._path, 1561 ) 1562 1563 super().validate_definition() 1564 1565 def is_breaking_change(self, previous: Model) -> t.Optional[bool]: 1566 if not isinstance(previous, SqlModel): 1567 return None 1568 1569 if self.lookback != previous.lookback: 1570 return None 1571 1572 try: 1573 # the previous model which comes from disk could be unrenderable 1574 previous_query = previous.render_query() 1575 except Exception: 1576 previous_query = None 1577 this_query = self.render_query() 1578 1579 if previous_query is None or this_query is None: 1580 # Can't determine if there's a breaking change if we can't render the query. 1581 return None 1582 1583 if previous_query is this_query: 1584 edits = [] 1585 else: 1586 edits = diff( 1587 previous_query, 1588 this_query, 1589 matchings=[(previous_query, this_query)], 1590 delta_only=True, 1591 dialect=self.dialect if self.dialect == previous.dialect else None, 1592 ) 1593 inserted_expressions = {e.expression for e in edits if isinstance(e, Insert)} 1594 1595 for edit in edits: 1596 if not isinstance(edit, Insert): 1597 return _additive_projection_change(previous_query, this_query, self.dialect) 1598 1599 expr = edit.expression 1600 if isinstance(expr, exp.UDTF): 1601 # projection subqueries do not change cardinality, engines don't allow these to return 1602 # more than one row of data 1603 parent = expr.find_ancestor(exp.Subquery) 1604 1605 if not parent: 1606 return None 1607 1608 expr = parent 1609 1610 if not _is_projection(expr) and expr.parent not in inserted_expressions: 1611 return _additive_projection_change(previous_query, this_query, self.dialect) 1612 1613 return False 1614 1615 def is_metadata_only_change(self, previous: _Node) -> bool: 1616 if self._is_metadata_only_change_cache.get(id(previous), None) is not None: 1617 return self._is_metadata_only_change_cache[id(previous)] 1618 1619 if not super().is_metadata_only_change(previous): 1620 return False 1621 1622 if not isinstance(previous, SqlModel): 1623 self._is_metadata_only_change_cache[id(previous)] = False 1624 return False 1625 1626 this_rendered_query = self.render_query() or self.query 1627 previous_rendered_query = previous.render_query() or previous.query 1628 is_metadata_change = this_rendered_query == previous_rendered_query 1629 1630 self._is_metadata_only_change_cache[id(previous)] = is_metadata_change 1631 return is_metadata_change 1632 1633 @cached_property 1634 def _query_renderer(self) -> QueryRenderer: 1635 no_quote_identifiers = self.kind.is_view and self.dialect in ("trino", "spark") 1636 return QueryRenderer( 1637 self.query, 1638 self.dialect, 1639 self.macro_definitions, 1640 schema=self.mapping_schema, 1641 path=self._path, 1642 jinja_macro_registry=self.jinja_macros, 1643 python_env=self.python_env, 1644 only_execution_time=self.kind.only_execution_time, 1645 default_catalog=self.default_catalog, 1646 quote_identifiers=not no_quote_identifiers, 1647 optimize_query=self.optimize_query, 1648 model=self, 1649 ) 1650 1651 @property 1652 def _data_hash_values_no_sql(self) -> t.List[str]: 1653 return [ 1654 *super()._data_hash_values_no_sql, 1655 *self.jinja_macros.data_hash_values, 1656 ] 1657 1658 @property 1659 def _data_hash_values_sql(self) -> t.List[str]: 1660 return [ 1661 *super()._data_hash_values_sql, 1662 self.query_.sql, 1663 ] 1664 1665 @property 1666 def _additional_metadata(self) -> t.List[str]: 1667 return [*super()._additional_metadata, self.query_.sql] 1668 1669 @property 1670 def violated_rules_for_query(self) -> t.Dict[type[Rule], t.Any]: 1671 self.render_query() 1672 return self._query_renderer._violated_rules 1673 1674 1675class SeedModel(_Model): 1676 """The model definition which uses a pre-built static dataset to source the data from. 1677 1678 Args: 1679 seed: The content of a pre-built static dataset. 1680 """ 1681 1682 kind: SeedKind 1683 seed: Seed 1684 column_hashes_: t.Optional[t.Dict[str, str]] = Field(default=None, alias="column_hashes") 1685 derived_columns_to_types: t.Optional[t.Dict[str, exp.DataType]] = None 1686 is_hydrated: bool = True 1687 source_type: t.Literal["seed"] = "seed" 1688 1689 def __getstate__(self) -> t.Dict[t.Any, t.Any]: 1690 state = super().__getstate__() 1691 state["__dict__"] = state["__dict__"].copy() 1692 state["__dict__"].pop("_reader", None) 1693 return state 1694 1695 def copy(self, **kwargs: t.Any) -> Self: 1696 model = super().copy(**kwargs) 1697 model.__dict__.pop("_reader", None) 1698 return model 1699 1700 def render( 1701 self, 1702 *, 1703 context: ExecutionContext, 1704 start: t.Optional[TimeLike] = None, 1705 end: t.Optional[TimeLike] = None, 1706 execution_time: t.Optional[TimeLike] = None, 1707 **kwargs: t.Any, 1708 ) -> t.Iterator[QueryOrDF]: 1709 if not self.is_hydrated: 1710 return 1711 yield from self.render_seed() 1712 1713 def render_seed(self) -> t.Iterator[QueryOrDF]: 1714 import numpy as np 1715 1716 self._ensure_hydrated() 1717 1718 date_columns = [] 1719 datetime_columns = [] 1720 bool_columns = [] 1721 string_columns = [] 1722 1723 columns_to_types = self.columns_to_types_ or {} 1724 column_names_to_check = set(columns_to_types) 1725 for name, tpe in columns_to_types.items(): 1726 if tpe.this in (exp.DataType.Type.DATE, exp.DataType.Type.DATE32): 1727 date_columns.append(name) 1728 elif tpe.this in exp.DataType.TEMPORAL_TYPES: 1729 datetime_columns.append(name) 1730 elif tpe.is_type("boolean"): 1731 bool_columns.append(name) 1732 elif tpe.this in exp.DataType.TEXT_TYPES: 1733 string_columns.append(name) 1734 1735 for df in self._reader.read(batch_size=self.kind.batch_size): 1736 rename_dict = {} 1737 for column in columns_to_types: 1738 if column not in df: 1739 normalized_name = normalize_identifiers(column, dialect=self.dialect).name 1740 if normalized_name in df: 1741 rename_dict[normalized_name] = column 1742 if rename_dict: 1743 df.rename(columns=rename_dict, inplace=True) 1744 # These names have already been checked 1745 column_names_to_check -= set(rename_dict) 1746 1747 missing_columns = column_names_to_check - set(df.columns) 1748 if missing_columns: 1749 raise_config_error( 1750 f"Seed model '{self.name}' has missing columns: {missing_columns}", self._path 1751 ) 1752 1753 # convert all date/time types to native pandas timestamp 1754 for column in [*date_columns, *datetime_columns]: 1755 import pandas as pd 1756 1757 df[column] = pd.to_datetime(df[column], infer_datetime_format=True, errors="ignore") # type: ignore 1758 1759 # extract datetime.date from pandas timestamp for DATE columns 1760 for column in date_columns: 1761 try: 1762 df[column] = df[column].dt.date 1763 except Exception as ex: 1764 logger.error( 1765 "Failed to convert column '%s' to date in seed model '%s': %s", 1766 column, 1767 self.name, 1768 ex, 1769 ) 1770 1771 for column in bool_columns: 1772 df[column] = df[column].apply(lambda i: str_to_bool(str(i))) 1773 1774 df.loc[:, string_columns] = df[string_columns].mask( 1775 cond=lambda x: x.notna(), # type: ignore 1776 other=df[string_columns].astype(str), # type: ignore 1777 ) 1778 yield df.replace({np.nan: None}) 1779 1780 @property 1781 def columns_to_types(self) -> t.Optional[t.Dict[str, exp.DataType]]: 1782 if self.columns_to_types_ is not None: 1783 return self.columns_to_types_ 1784 if self.derived_columns_to_types is not None: 1785 return self.derived_columns_to_types 1786 if self.is_hydrated: 1787 return self._reader.columns_to_types 1788 return None 1789 1790 @property 1791 def column_hashes(self) -> t.Dict[str, str]: 1792 if self.column_hashes_ is not None: 1793 return self.column_hashes_ 1794 self._ensure_hydrated() 1795 return self._reader.column_hashes 1796 1797 @property 1798 def is_seed(self) -> bool: 1799 return True 1800 1801 @property 1802 def seed_path(self) -> Path: 1803 seed_path = Path(self.kind.path) 1804 if not seed_path.is_absolute(): 1805 if self._path is None: 1806 raise SQLMeshError(f"Seed model '{self.name}' has no path") 1807 return self._path.parent / seed_path 1808 return seed_path 1809 1810 @property 1811 def depends_on(self) -> t.Set[str]: 1812 return (self.depends_on_ or set()) - {self.fqn} 1813 1814 @property 1815 def depends_on_self(self) -> bool: 1816 return False 1817 1818 @property 1819 def batch_size(self) -> t.Optional[int]: 1820 # Unlike other model kinds, the batch size provided in the SEED kind represents the 1821 # maximum number of rows to insert in a single batch. 1822 # We should never batch intervals for seed models. 1823 return None 1824 1825 def to_dehydrated(self) -> SeedModel: 1826 """Creates a dehydrated copy of this model. 1827 1828 The dehydrated seed model will not contain the seed content, but will contain 1829 the column hashes. This is useful for comparing two seed models without 1830 having to read the seed content from disk. 1831 1832 Returns: 1833 A dehydrated copy of this model. 1834 """ 1835 if not self.is_hydrated: 1836 return self 1837 1838 return self.copy( 1839 update={ 1840 "seed": Seed(content=""), 1841 "is_hydrated": False, 1842 "column_hashes_": self.column_hashes, 1843 "derived_columns_to_types": self.columns_to_types 1844 if self.columns_to_types_ is None 1845 else None, 1846 } 1847 ) 1848 1849 def to_hydrated(self, content: str) -> SeedModel: 1850 """Creates a hydrated copy of this model with the given seed content. 1851 1852 Returns: 1853 A hydrated copy of this model. 1854 """ 1855 if self.is_hydrated: 1856 return self 1857 1858 return self.copy( 1859 update={ 1860 "seed": Seed(content=content), 1861 "is_hydrated": True, 1862 "column_hashes_": None, 1863 }, 1864 ) 1865 1866 def is_breaking_change(self, previous: Model) -> t.Optional[bool]: 1867 if not isinstance(previous, SeedModel): 1868 return None 1869 1870 new_columns = set(self.column_hashes) 1871 old_columns = set(previous.column_hashes) 1872 1873 if not new_columns.issuperset(old_columns): 1874 return None 1875 1876 for col in old_columns: 1877 if self.column_hashes[col] != previous.column_hashes[col]: 1878 return None 1879 1880 return False 1881 1882 def _ensure_hydrated(self) -> None: 1883 if not self.is_hydrated: 1884 raise SQLMeshError(f"Seed model '{self.name}' is not hydrated.") 1885 1886 @cached_property 1887 def _reader(self) -> CsvSeedReader: 1888 return self.seed.reader(dialect=self.dialect, settings=self.kind.csv_settings) 1889 1890 @property 1891 def _data_hash_values_no_sql(self) -> t.List[str]: 1892 data = super()._data_hash_values_no_sql 1893 for column_name, column_hash in self.column_hashes.items(): 1894 data.append(column_name) 1895 data.append(column_hash) 1896 1897 # Include grants in data hash for seed models to force recreation on grant changes 1898 # since seed models don't support migration 1899 data.append(json.dumps(self.grants, sort_keys=True) if self.grants else "") 1900 data.append(self.grants_target_layer) 1901 1902 return data 1903 1904 1905class PythonModel(_Model): 1906 """The model definition which relies on a Python script to fetch the data. 1907 1908 Args: 1909 entrypoint: The name of a Python function which contains the data fetching / transformation logic. 1910 """ 1911 1912 kind: ModelKind = FullKind() 1913 entrypoint: str 1914 source_type: t.Literal["python"] = "python" 1915 1916 def validate_definition(self) -> None: 1917 super().validate_definition() 1918 1919 if self.kind and not self.kind.supports_python_models: 1920 raise_config_error( 1921 f"Cannot create Python model '{self.name}' as the '{self.kind.name}' kind doesn't support Python models", 1922 self._path, 1923 ) 1924 1925 def render( 1926 self, 1927 *, 1928 context: ExecutionContext, 1929 start: t.Optional[TimeLike] = None, 1930 end: t.Optional[TimeLike] = None, 1931 execution_time: t.Optional[TimeLike] = None, 1932 **kwargs: t.Any, 1933 ) -> t.Iterator[QueryOrDF]: 1934 env = prepare_env(self.python_env) 1935 start, end = make_inclusive(start or c.EPOCH, end or c.EPOCH, self.dialect) 1936 execution_time = to_datetime(execution_time or c.EPOCH) 1937 1938 variables = { 1939 **env.get(c.SQLMESH_VARS, {}), 1940 **env.get(c.SQLMESH_VARS_METADATA, {}), 1941 **kwargs.pop("variables", {}), 1942 } 1943 blueprint_variables = { 1944 k: d.parse_one(v.sql, dialect=self.dialect) if isinstance(v, SqlValue) else v 1945 for k, v in { 1946 **env.get(c.SQLMESH_BLUEPRINT_VARS, {}), 1947 **env.get(c.SQLMESH_BLUEPRINT_VARS_METADATA, {}), 1948 }.items() 1949 } 1950 try: 1951 kwargs = { 1952 **variables, 1953 **kwargs, 1954 "start": start, 1955 "end": end, 1956 "execution_time": execution_time, 1957 "latest": execution_time, # TODO: Preserved for backward compatibility. Remove in 1.0.0. 1958 } 1959 df_or_iter = env[self.entrypoint]( 1960 context=context.with_variables(variables, blueprint_variables=blueprint_variables), 1961 **kwargs, 1962 ) 1963 1964 if not isinstance(df_or_iter, types.GeneratorType): 1965 df_or_iter = [df_or_iter] 1966 1967 for df in df_or_iter: 1968 yield df 1969 except Exception as e: 1970 raise PythonModelEvalError(format_evaluated_code_exception(e, self.python_env)) 1971 1972 def render_definition( 1973 self, 1974 include_python: bool = True, 1975 include_defaults: bool = False, 1976 render_query: bool = False, 1977 ) -> t.List[exp.Expr]: 1978 # Ignore the provided value for the include_python flag, since the Pyhon model's 1979 # definition without Python code is meaningless. 1980 return super().render_definition( 1981 include_python=True, include_defaults=include_defaults, render_query=render_query 1982 ) 1983 1984 @property 1985 def is_python(self) -> bool: 1986 return True 1987 1988 def is_breaking_change(self, previous: Model) -> t.Optional[bool]: 1989 return None 1990 1991 @property 1992 def _data_hash_values_no_sql(self) -> t.List[str]: 1993 data = super()._data_hash_values_no_sql 1994 data.append(self.entrypoint) 1995 return data 1996 1997 1998class ExternalModel(_Model): 1999 """The model definition which represents an external source/table.""" 2000 2001 kind: ModelKind = ExternalKind() 2002 source_type: t.Literal["external"] = "external" 2003 2004 def is_breaking_change(self, previous: Model) -> t.Optional[bool]: 2005 if not isinstance(previous, ExternalModel): 2006 return None 2007 if not previous.columns_to_types_or_raise.items() - self.columns_to_types_or_raise.items(): 2008 return False 2009 return None 2010 2011 @property 2012 def depends_on(self) -> t.Set[str]: 2013 return set() 2014 2015 @property 2016 def depends_on_self(self) -> bool: 2017 return False 2018 2019 2020Model = t.Union[SqlModel, SeedModel, PythonModel, ExternalModel] 2021 2022 2023class AuditResult(PydanticModel): 2024 audit: Audit 2025 """The audit this result is for.""" 2026 audit_args: t.Dict[t.Any, t.Any] 2027 """Arguments passed to the audit.""" 2028 model: t.Optional[_Model] = None 2029 """The model this audit is for.""" 2030 count: t.Optional[int] = None 2031 """The number of records returned by the audit query. This could be None if the audit was skipped.""" 2032 query: t.Optional[exp.Expr] = None 2033 """The rendered query used by the audit. This could be None if the audit was skipped.""" 2034 skipped: bool = False 2035 """Whether or not the audit was blocking. This can be overriden by the user.""" 2036 blocking: bool = True 2037 2038 2039class EvaluatableSignals(PydanticModel): 2040 signals_to_kwargs: t.Dict[str, t.Dict[str, t.Optional[exp.Expr]]] 2041 """A mapping of signal names to the kwargs passed to the signal.""" 2042 python_env: t.Dict[str, Executable] 2043 """The Python environment that should be used to evaluated the rendered signal calls.""" 2044 prepared_python_env: t.Dict[str, t.Any] 2045 """The prepared Python environment that should be used to evaluated the rendered signal calls.""" 2046 2047 2048def _extract_blueprints(blueprints: t.Any, path: Path) -> t.List[t.Any]: 2049 if not blueprints: 2050 return [None] 2051 if isinstance(blueprints, exp.Paren): 2052 return [blueprints.unnest()] 2053 if isinstance(blueprints, (exp.Tuple, exp.Array)): 2054 return blueprints.expressions 2055 if isinstance(blueprints, list): 2056 return blueprints 2057 2058 raise_config_error( 2059 "Expected a list or tuple consisting of key-value mappings for " 2060 f"the 'blueprints' property, got '{blueprints}' instead", 2061 path, 2062 ) 2063 return [] # This is unreachable, but is done to satisfy mypy 2064 2065 2066def _extract_blueprint_variables(blueprint: t.Any, path: Path) -> t.Dict[str, t.Any]: 2067 if not blueprint: 2068 return {} 2069 if isinstance(blueprint, (exp.Paren, exp.PropertyEQ)): 2070 blueprint = blueprint.unnest() 2071 return {blueprint.left.name.lower(): blueprint.right} 2072 if isinstance(blueprint, (exp.Tuple, exp.Array)): 2073 return {e.left.name.lower(): e.right for e in blueprint.expressions} 2074 if isinstance(blueprint, dict): 2075 return {k.lower(): v for k, v in blueprint.items()} 2076 2077 raise_config_error( 2078 f"Expected a key-value mapping for the blueprint value, got '{blueprint}' instead", 2079 path, 2080 ) 2081 return {} # This is unreachable, but is done to satisfy mypy 2082 2083 2084def create_models_from_blueprints( 2085 gateway: t.Optional[str | exp.Expr], 2086 blueprints: t.Any, 2087 get_variables: t.Callable[[t.Optional[str]], t.Dict[str, str]], 2088 loader: t.Callable[..., Model], 2089 path: Path = Path(), 2090 module_path: Path = Path(), 2091 dialect: DialectType = None, 2092 default_catalog_per_gateway: t.Optional[t.Dict[str, str]] = None, 2093 **loader_kwargs: t.Any, 2094) -> t.List[Model]: 2095 model_blueprints: t.List[Model] = [] 2096 original_default_catalog = loader_kwargs.get("default_catalog") 2097 for blueprint in _extract_blueprints(blueprints, path): 2098 loader_kwargs["default_catalog"] = original_default_catalog 2099 blueprint_variables = _extract_blueprint_variables(blueprint, path) 2100 2101 if gateway: 2102 rendered_gateway = render_expression( 2103 expression=exp.maybe_parse(gateway, dialect=dialect), 2104 module_path=module_path, 2105 macros=loader_kwargs.get("macros"), 2106 jinja_macros=loader_kwargs.get("jinja_macros"), 2107 path=path, 2108 dialect=dialect, 2109 default_catalog=loader_kwargs.get("default_catalog"), 2110 blueprint_variables=blueprint_variables, 2111 ) 2112 gateway_name = rendered_gateway[0].name if rendered_gateway else None 2113 else: 2114 gateway_name = None 2115 2116 if default_catalog_per_gateway and gateway_name: 2117 catalog = default_catalog_per_gateway.get(gateway_name) 2118 if catalog is not None: 2119 loader_kwargs["default_catalog"] = catalog 2120 else: 2121 # Gateway exists but has no entry in the dict (e.g., catalog-unsupported 2122 # engines like ClickHouse). Clear the default catalog so the global 2123 # default from the primary gateway doesn't leak into this model's name. 2124 loader_kwargs["default_catalog"] = None 2125 2126 model_blueprints.append( 2127 loader( 2128 path=path, 2129 module_path=module_path, 2130 dialect=dialect, 2131 variables=get_variables(gateway_name), 2132 blueprint_variables=blueprint_variables, 2133 **loader_kwargs, 2134 ) 2135 ) 2136 2137 return model_blueprints 2138 2139 2140def load_sql_based_models( 2141 expressions: t.List[exp.Expr], 2142 get_variables: t.Callable[[t.Optional[str]], t.Dict[str, str]], 2143 path: Path = Path(), 2144 module_path: Path = Path(), 2145 dialect: DialectType = None, 2146 default_catalog_per_gateway: t.Optional[t.Dict[str, str]] = None, 2147 **loader_kwargs: t.Any, 2148) -> t.List[Model]: 2149 gateway: t.Optional[exp.Expr] = None 2150 blueprints: t.Optional[exp.Expr] = None 2151 2152 model_meta = seq_get(expressions, 0) 2153 for prop in (isinstance(model_meta, d.Model) and model_meta.expressions) or []: 2154 if prop.name == "gateway": 2155 gateway = prop.args["value"] 2156 elif prop.name == "blueprints": 2157 # We pop the `blueprints` here to avoid walking large lists when rendering the meta 2158 blueprints = prop.pop().args["value"] 2159 2160 if isinstance(blueprints, d.MacroFunc): 2161 rendered_blueprints = render_expression( 2162 expression=blueprints, 2163 module_path=module_path, 2164 macros=loader_kwargs.get("macros"), 2165 jinja_macros=loader_kwargs.get("jinja_macros"), 2166 variables=get_variables(None), 2167 path=path, 2168 dialect=dialect, 2169 default_catalog=loader_kwargs.get("default_catalog"), 2170 ) 2171 if not rendered_blueprints: 2172 raise_config_error("Failed to render blueprints property", path) 2173 2174 # Help mypy see that rendered_blueprints can't be None 2175 assert rendered_blueprints 2176 2177 if len(rendered_blueprints) > 1: 2178 rendered_blueprints = [exp.Tuple(expressions=rendered_blueprints)] 2179 2180 blueprints = rendered_blueprints[0] 2181 2182 return create_models_from_blueprints( 2183 gateway=gateway, 2184 blueprints=blueprints, 2185 get_variables=get_variables, 2186 loader=partial(load_sql_based_model, expressions), 2187 path=path, 2188 module_path=module_path, 2189 dialect=dialect, 2190 default_catalog_per_gateway=default_catalog_per_gateway, 2191 **loader_kwargs, 2192 ) 2193 2194 2195def load_sql_based_model( 2196 expressions: t.List[exp.Expr], 2197 *, 2198 defaults: t.Optional[t.Dict[str, t.Any]] = None, 2199 path: t.Optional[Path] = None, 2200 module_path: Path = Path(), 2201 time_column_format: str = c.DEFAULT_TIME_COLUMN_FORMAT, 2202 macros: t.Optional[MacroRegistry] = None, 2203 jinja_macros: t.Optional[JinjaMacroRegistry] = None, 2204 audits: t.Optional[t.Dict[str, ModelAudit]] = None, 2205 python_env: t.Optional[t.Dict[str, Executable]] = None, 2206 dialect: t.Optional[str] = None, 2207 physical_schema_mapping: t.Optional[t.Dict[re.Pattern, str]] = None, 2208 default_catalog: t.Optional[str] = None, 2209 variables: t.Optional[t.Dict[str, t.Any]] = None, 2210 infer_names: t.Optional[bool] = False, 2211 blueprint_variables: t.Optional[t.Dict[str, t.Any]] = None, 2212 **kwargs: t.Any, 2213) -> Model: 2214 """Load a model from a parsed SQLMesh model SQL file. 2215 2216 Args: 2217 expressions: Model, *Statements, Query. 2218 defaults: Definition default values. 2219 path: An optional path to the model definition file. 2220 module_path: The python module path to serialize macros for. 2221 time_column_format: The default time column format to use if no model time column is configured. 2222 macros: The custom registry of macros. If not provided the default registry will be used. 2223 jinja_macros: The registry of Jinja macros. 2224 python_env: The custom Python environment for macros. If not provided the environment will be constructed 2225 from the macro registry. 2226 dialect: The default dialect if no model dialect is configured. 2227 The format must adhere to Python's strftime codes. 2228 physical_schema_mapping: A mapping of regular expressions to match against the model schema to produce the corresponding physical schema 2229 default_catalog: The default catalog if no model catalog is configured. 2230 variables: The variables to pass to the model. 2231 kwargs: Additional kwargs to pass to the loader. 2232 """ 2233 missing_model_msg = f"""Please add a MODEL block at the top of the file. Example: 2234 2235MODEL ( 2236 name sqlmesh_example.full_model, --model name 2237 kind FULL, --materialization 2238 cron '@daily', --schedule 2239); 2240 2241Learn more at https://sqlmesh.readthedocs.io/en/stable/concepts/models/overview 2242""" 2243 2244 if not expressions: 2245 raise_config_error(missing_model_msg) 2246 2247 dialect = dialect or "" 2248 meta = expressions[0] 2249 if not isinstance(meta, d.Model): 2250 if not infer_names: 2251 raise_config_error(missing_model_msg) 2252 meta = d.Model(expressions=[]) # Dummy meta node 2253 expressions.insert(0, meta) 2254 2255 # We deliberately hold off rendering some properties at load time because there is not enough information available 2256 # at load time to render them. They will get rendered later at evaluation time 2257 unrendered_properties = {} 2258 unrendered_merge_filter = None 2259 2260 for prop in meta.expressions: 2261 # Macro functions that programmaticaly generate the key-value pair properties should be rendered 2262 # This is needed in the odd case where a macro shares the name of one of the properties 2263 # eg `@session_properties()` Test: `test_macros_in_model_statement` Reference PR: #2574 2264 if isinstance(prop, d.MacroFunc): 2265 continue 2266 2267 prop_name = prop.name.lower() 2268 if prop_name in {"signals", "audits"} | PROPERTIES: 2269 unrendered_properties[prop_name] = prop.args.get("value") 2270 elif ( 2271 prop.name.lower() == "kind" 2272 and (value := prop.args.get("value")) 2273 and value.name.lower() == "incremental_by_unique_key" 2274 ): 2275 for kind_prop in value.expressions: 2276 if kind_prop.name.lower() == "merge_filter": 2277 unrendered_merge_filter = kind_prop 2278 2279 rendered_meta_exprs = render_expression( 2280 expression=meta, 2281 module_path=module_path, 2282 macros=macros, 2283 jinja_macros=jinja_macros, 2284 variables=variables, 2285 path=path, 2286 dialect=dialect, 2287 default_catalog=default_catalog, 2288 blueprint_variables=blueprint_variables, 2289 ) 2290 2291 if rendered_meta_exprs is None or len(rendered_meta_exprs) != 1: 2292 raise_config_error( 2293 f"Invalid MODEL statement:\n{meta.sql(dialect=dialect, pretty=True)}", 2294 path, 2295 ) 2296 raise 2297 2298 rendered_meta = rendered_meta_exprs[0] 2299 2300 rendered_defaults = ( 2301 render_model_defaults( 2302 defaults=defaults, 2303 module_path=module_path, 2304 macros=macros, 2305 jinja_macros=jinja_macros, 2306 variables=variables, 2307 path=path, 2308 dialect=dialect, 2309 default_catalog=default_catalog, 2310 ) 2311 if defaults 2312 else {} 2313 ) 2314 2315 rendered_defaults = parse_defaults_properties(rendered_defaults, dialect=dialect) 2316 2317 # Extract the query and any pre/post statements 2318 query_or_seed_insert, pre_statements, post_statements, on_virtual_update, inline_audits = ( 2319 _split_sql_model_statements(expressions[1:], path, dialect=dialect) 2320 ) 2321 2322 meta_fields: t.Dict[str, t.Any] = { 2323 "dialect": dialect, 2324 "description": ( 2325 "\n".join(comment.strip() for comment in rendered_meta.comments) 2326 if rendered_meta.comments 2327 else None 2328 ), 2329 **{prop.name.lower(): prop.args.get("value") for prop in rendered_meta.expressions}, 2330 **kwargs, 2331 } 2332 2333 # Discard the potentially half-rendered versions of these properties and replace them with the 2334 # original unrendered versions. They will get rendered properly at evaluation time 2335 meta_fields.update(unrendered_properties) 2336 2337 if unrendered_merge_filter: 2338 for idx, kind_prop in enumerate(meta_fields["kind"].expressions): 2339 if kind_prop.name.lower() == "merge_filter": 2340 meta_fields["kind"].expressions[idx] = unrendered_merge_filter 2341 2342 if isinstance(meta_fields.get("dialect"), exp.Expr): 2343 meta_fields["dialect"] = meta_fields["dialect"].name 2344 2345 # The name of the model will be inferred from its path relative to `models/`, if it's not explicitly specified 2346 name = meta_fields.pop("name", "") 2347 if not name and infer_names: 2348 if path is None: 2349 raise ValueError(f"Model {name} must have a name") 2350 name = get_model_name(path) 2351 2352 if not name: 2353 raise_config_error( 2354 "Please add the required 'name' field to the MODEL block at the top of the file.\n\n" 2355 + "Learn more at https://sqlmesh.readthedocs.io/en/stable/concepts/models/overview" 2356 ) 2357 if "default_catalog" in meta_fields: 2358 raise_config_error( 2359 "`default_catalog` cannot be set on a per-model basis. It must be set at the connection level.", 2360 path, 2361 ) 2362 2363 common_kwargs = dict( 2364 pre_statements=pre_statements, 2365 post_statements=post_statements, 2366 on_virtual_update=on_virtual_update, 2367 defaults=rendered_defaults, 2368 path=path, 2369 module_path=module_path, 2370 macros=macros, 2371 python_env=python_env, 2372 jinja_macros=jinja_macros, 2373 physical_schema_mapping=physical_schema_mapping, 2374 default_catalog=default_catalog, 2375 variables=variables, 2376 inline_audits=inline_audits, 2377 blueprint_variables=blueprint_variables, 2378 use_original_sql=True, 2379 **meta_fields, 2380 ) 2381 2382 kind = common_kwargs.pop("kind", ModelMeta.all_field_infos()["kind"].default) 2383 2384 if kind.name != ModelKindName.SEED: 2385 return create_sql_model( 2386 name, 2387 query_or_seed_insert, 2388 kind=kind, 2389 time_column_format=time_column_format, 2390 **common_kwargs, 2391 ) 2392 2393 seed_properties = {p.name.lower(): p.args.get("value") for p in kind.expressions} 2394 return create_seed_model( 2395 name, 2396 SeedKind(**seed_properties), 2397 **common_kwargs, 2398 ) 2399 2400 2401def create_sql_model( 2402 name: TableName, 2403 query: t.Optional[exp.Expr], 2404 **kwargs: t.Any, 2405) -> Model: 2406 """Creates a SQL model. 2407 2408 Args: 2409 name: The name of the model, which is of the form [catalog].[db].table. 2410 The catalog and db are optional. 2411 query: The model's logic in a form of a SELECT query. 2412 """ 2413 if not isinstance(query, (exp.Query, d.JinjaQuery, d.MacroFunc)): 2414 raise_config_error( 2415 "A query is required and must be a SELECT statement, a UNION statement, or a JINJA_QUERY block", 2416 kwargs.get("path"), 2417 ) 2418 assert isinstance(query, (exp.Query, d.JinjaQuery, d.MacroFunc)) 2419 2420 return _create_model(SqlModel, name, query=query, **kwargs) 2421 2422 2423def create_seed_model( 2424 name: TableName, 2425 seed_kind: SeedKind, 2426 *, 2427 path: t.Optional[Path] = None, 2428 module_path: Path = Path(), 2429 **kwargs: t.Any, 2430) -> Model: 2431 """Creates a Seed model. 2432 2433 Args: 2434 name: The name of the model, which is of the form [catalog].[db].table. 2435 The catalog and db are optional. 2436 seed_kind: The information about the location of a seed and other related configuration. 2437 path: An optional path to the model definition file. 2438 from the macro registry. 2439 """ 2440 seed_path = Path(seed_kind.path) 2441 marker, *subdirs = seed_path.parts 2442 if marker.lower() == "$root": 2443 seed_path = module_path.joinpath(*subdirs) 2444 seed_kind.path = str(seed_path) 2445 elif not seed_path.is_absolute(): 2446 if path is None: 2447 seed_path = seed_path 2448 elif path.is_dir(): 2449 seed_path = path / seed_path 2450 else: 2451 seed_path = path.parent / seed_path 2452 2453 seed = create_seed(seed_path) 2454 2455 return _create_model( 2456 SeedModel, 2457 name, 2458 path=path, 2459 seed=seed, 2460 kind=seed_kind, 2461 depends_on=kwargs.pop("depends_on", None), 2462 module_path=module_path, 2463 **kwargs, 2464 ) 2465 2466 2467def create_python_model( 2468 name: str, 2469 entrypoint: str, 2470 python_env: t.Dict[str, Executable], 2471 *, 2472 macros: t.Optional[MacroRegistry] = None, 2473 jinja_macros: t.Optional[JinjaMacroRegistry] = None, 2474 path: Path = Path(), 2475 module_path: Path = Path(), 2476 depends_on: t.Optional[t.Set[str]] = None, 2477 variables: t.Optional[t.Dict[str, t.Any]] = None, 2478 blueprint_variables: t.Optional[t.Dict[str, t.Any]] = None, 2479 **kwargs: t.Any, 2480) -> Model: 2481 """Creates a Python model. 2482 2483 Args: 2484 name: The name of the model, which is of the form [catalog].[db].table. 2485 The catalog and db are optional. 2486 entrypoint: The name of a Python function which contains the data fetching / transformation logic. 2487 python_env: The Python environment of all objects referenced by the model implementation. 2488 path: An optional path to the model definition file. 2489 depends_on: The custom set of model's upstream dependencies. 2490 variables: The variables to pass to the model. 2491 blueprint_variables: The blueprint's variables to pass to the model. 2492 """ 2493 # Find dependencies for python models by parsing code if they are not explicitly defined 2494 # Also remove self-references that are found 2495 2496 dialect = kwargs.get("dialect") 2497 2498 dependencies_unspecified = depends_on is None 2499 2500 parsed_depends_on, referenced_variables = ( 2501 parse_dependencies( 2502 python_env, 2503 entrypoint, 2504 strict_resolution=dependencies_unspecified, 2505 variables=variables, 2506 blueprint_variables=blueprint_variables, 2507 ) 2508 if python_env is not None 2509 else (set(), set()) 2510 ) 2511 if dependencies_unspecified: 2512 depends_on = parsed_depends_on - {name} 2513 else: 2514 depends_on_rendered = render_expression( 2515 expression=exp.Array( 2516 expressions=[exp.maybe_parse(dep, dialect=dialect) for dep in depends_on or []] 2517 ), 2518 module_path=module_path, 2519 macros=macros, 2520 jinja_macros=jinja_macros, 2521 variables=variables, 2522 path=path, 2523 dialect=dialect, 2524 default_catalog=kwargs.get("default_catalog"), 2525 ) 2526 depends_on = { 2527 dep.sql(dialect=dialect) 2528 for dep in t.cast(t.List[exp.Expr], depends_on_rendered)[0].expressions 2529 } 2530 2531 used_variables = {k: v for k, v in (variables or {}).items() if k in referenced_variables} 2532 if used_variables: 2533 python_env[c.SQLMESH_VARS] = Executable.value(used_variables, sort_root_dict=True) 2534 2535 return _create_model( 2536 PythonModel, 2537 name, 2538 path=path, 2539 depends_on=depends_on, 2540 entrypoint=entrypoint, 2541 python_env=python_env, 2542 macros=macros, 2543 jinja_macros=jinja_macros, 2544 module_path=module_path, 2545 variables=variables, 2546 blueprint_variables=blueprint_variables, 2547 **kwargs, 2548 ) 2549 2550 2551def create_external_model( 2552 name: TableName, 2553 *, 2554 dialect: t.Optional[str] = None, 2555 path: Path = Path(), 2556 defaults: t.Optional[t.Dict[str, t.Any]] = None, 2557 **kwargs: t.Any, 2558) -> ExternalModel: 2559 """Creates an external model. 2560 2561 Args: 2562 name: The name of the model, which is of the form [catalog].[db].table. 2563 The catalog and db are optional. 2564 dialect: The dialect to serialize. 2565 path: An optional path to the model definition file. 2566 """ 2567 return t.cast( 2568 ExternalModel, 2569 _create_model( 2570 ExternalModel, 2571 name, 2572 defaults=defaults, 2573 dialect=dialect, 2574 path=path, 2575 kind=ModelKindName.EXTERNAL.value, 2576 **kwargs, 2577 ), 2578 ) 2579 2580 2581def _create_model( 2582 klass: t.Type[_Model], 2583 name: TableName, 2584 *, 2585 defaults: t.Optional[t.Dict[str, t.Any]] = None, 2586 path: t.Optional[Path] = None, 2587 time_column_format: str = c.DEFAULT_TIME_COLUMN_FORMAT, 2588 jinja_macros: t.Optional[JinjaMacroRegistry] = None, 2589 jinja_macro_references: t.Optional[t.Set[MacroReference]] = None, 2590 depends_on: t.Optional[t.Set[str]] = None, 2591 dialect: t.Optional[str] = None, 2592 physical_schema_mapping: t.Optional[t.Dict[re.Pattern, str]] = None, 2593 python_env: t.Optional[t.Dict[str, Executable]] = None, 2594 audit_definitions: t.Optional[t.Dict[str, ModelAudit]] = None, 2595 inline_audits: t.Optional[t.Dict[str, ModelAudit]] = None, 2596 module_path: Path = Path(), 2597 macros: t.Optional[MacroRegistry] = None, 2598 signal_definitions: t.Optional[SignalRegistry] = None, 2599 variables: t.Optional[t.Dict[str, t.Any]] = None, 2600 blueprint_variables: t.Optional[t.Dict[str, t.Any]] = None, 2601 use_original_sql: bool = False, 2602 **kwargs: t.Any, 2603) -> Model: 2604 validate_extra_and_required_fields( 2605 klass, 2606 {"name", *kwargs} - {"grain", "table_properties"}, 2607 "MODEL block", 2608 path, 2609 ) 2610 2611 for prop in PROPERTIES: 2612 kwargs[prop] = _resolve_properties((defaults or {}).get(prop), kwargs.get(prop)) 2613 2614 dialect = dialect or "" 2615 2616 physical_schema_mapping = physical_schema_mapping or {} 2617 model_schema_name = exp.to_table(name, dialect=dialect).db 2618 physical_schema_override: t.Optional[str] = None 2619 2620 for re_pattern, override_schema in physical_schema_mapping.items(): 2621 if re.match(re_pattern, model_schema_name): 2622 physical_schema_override = override_schema 2623 break 2624 2625 raw_kind = kwargs.pop("kind", None) 2626 if raw_kind: 2627 kwargs["kind"] = create_model_kind(raw_kind, dialect, defaults or {}) 2628 2629 defaults = {k: v for k, v in (defaults or {}).items() if k in klass.all_fields()} 2630 if not issubclass(klass, SqlModel): 2631 defaults.pop("optimize_query", None) 2632 2633 statements: t.List[t.Union[exp.Expr, t.Tuple[exp.Expr, bool]]] = [] 2634 2635 if "query" in kwargs: 2636 statements.append(kwargs["query"]) 2637 kwargs["query"] = ParsableSql.from_parsed_expression( 2638 kwargs["query"], dialect, use_meta_sql=use_original_sql 2639 ) 2640 2641 # Merge default statements with model-specific statements 2642 for statement_field in ["pre_statements", "post_statements", "on_virtual_update"]: 2643 if statement_field in defaults: 2644 kwargs[statement_field] = [ 2645 exp.maybe_parse(stmt, dialect=dialect) for stmt in defaults[statement_field] 2646 ] + kwargs.get(statement_field, []) 2647 if statement_field in kwargs: 2648 # Macros extracted from these statements need to be treated as metadata only 2649 is_metadata = statement_field == "on_virtual_update" 2650 for stmt in kwargs[statement_field]: 2651 # Extract the expression if it's ParsableSql already 2652 expr = stmt.parse(dialect) if isinstance(stmt, ParsableSql) else stmt 2653 statements.append((expr, is_metadata)) 2654 kwargs[statement_field] = [ 2655 # this to retain the transaction information 2656 stmt 2657 if isinstance(stmt, ParsableSql) 2658 else ParsableSql.from_parsed_expression( 2659 stmt, dialect, use_meta_sql=use_original_sql 2660 ) 2661 for stmt in kwargs[statement_field] 2662 ] 2663 2664 # This is done to allow variables like @gateway to be used in these properties 2665 # since rendering shifted from load time to run time. 2666 # Note: we check for Tuple since that's what we expect from _resolve_properties 2667 for property_name in PROPERTIES: 2668 property_values = kwargs.get(property_name) 2669 if isinstance(property_values, exp.Tuple): 2670 statements.extend(property_values.expressions) 2671 2672 if isinstance(getattr(kwargs.get("kind"), "merge_filter", None), exp.Expr): 2673 statements.append(kwargs["kind"].merge_filter) 2674 2675 jinja_macro_references, referenced_variables = extract_macro_references_and_variables( 2676 *(gen(e if isinstance(e, exp.Expr) else e[0]) for e in statements) 2677 ) 2678 2679 if jinja_macros: 2680 jinja_macros = ( 2681 jinja_macros if jinja_macros.trimmed else jinja_macros.trim(jinja_macro_references) 2682 ) 2683 else: 2684 jinja_macros = JinjaMacroRegistry() 2685 2686 for jinja_macro in jinja_macros.root_macros.values(): 2687 referenced_variables.update( 2688 extract_macro_references_and_variables(jinja_macro.definition)[1] 2689 ) 2690 2691 # Merge model-specific audits with default audits 2692 if default_audits := defaults.pop("audits", None): 2693 kwargs["audits"] = default_audits + d.extract_function_calls(kwargs.pop("audits", [])) 2694 2695 model = klass( 2696 name=name, 2697 **{ 2698 **(defaults or {}), 2699 "jinja_macros": jinja_macros or JinjaMacroRegistry(), 2700 "dialect": dialect, 2701 "depends_on": depends_on, 2702 "physical_schema_override": physical_schema_override, 2703 **kwargs, 2704 }, 2705 ) 2706 2707 audit_definitions = { 2708 **(audit_definitions or {}), 2709 **(inline_audits or {}), 2710 } 2711 2712 used_audits: t.Set[str] = {audit_name for audit_name, _ in model.audits} 2713 2714 audit_definitions = { 2715 audit_name: audit_definitions[audit_name] 2716 for audit_name in used_audits 2717 if audit_name in audit_definitions 2718 } 2719 2720 model.audit_definitions.update(audit_definitions) 2721 2722 # Any macro referenced in audits or signals needs to be treated as metadata-only 2723 statements.extend((audit.query, True) for audit in audit_definitions.values()) # type: ignore[misc] 2724 2725 # Ensure that all audits referenced in the model are defined 2726 from sqlmesh.core.audit.builtin import BUILT_IN_AUDITS 2727 2728 available_audits = BUILT_IN_AUDITS.keys() | model.audit_definitions.keys() 2729 for referenced_audit, audit_args in model.audits: 2730 if referenced_audit not in available_audits: 2731 raise_config_error(f"Audit '{referenced_audit}' is undefined", location=path) 2732 2733 statements.extend( 2734 (audit_arg_expression, True) for audit_arg_expression in audit_args.values() 2735 ) 2736 2737 signal_definitions = signal_definitions or UniqueKeyDict("signals") 2738 2739 for referenced_signal, kwargs in model.signals: 2740 if referenced_signal and referenced_signal not in signal_definitions: 2741 raise_config_error(f"Signal '{referenced_signal}' is undefined", location=path) 2742 2743 statements.extend((signal_kwarg, True) for signal_kwarg in kwargs.values()) 2744 2745 python_env = make_python_env( 2746 statements, 2747 jinja_macro_references, 2748 module_path, 2749 macros or macro.get_registry(), 2750 variables=variables, 2751 referenced_variables=referenced_variables, 2752 path=path, 2753 python_env=python_env, 2754 strict_resolution=depends_on is None, 2755 blueprint_variables=blueprint_variables, 2756 dialect=dialect, 2757 ) 2758 2759 env: t.Dict[str, t.Tuple[t.Any, t.Optional[bool]]] = {} 2760 2761 for signal_name, _ in model.signals: 2762 if signal_name and signal_name in signal_definitions: 2763 func = signal_definitions[signal_name].func 2764 setattr(func, c.SQLMESH_METADATA, True) 2765 build_env(func, env=env, name=signal_name, path=module_path) 2766 2767 model.python_env.update(python_env) 2768 model.python_env.update(serialize_env(env, path=module_path)) 2769 model._path = path 2770 model.set_time_format(time_column_format) 2771 2772 return t.cast(Model, model) 2773 2774 2775INSERT_SEED_MACRO_CALL = d.parse_one("@INSERT_SEED()") 2776 2777 2778def _split_sql_model_statements( 2779 expressions: t.List[exp.Expr], 2780 path: t.Optional[Path], 2781 dialect: t.Optional[str] = None, 2782) -> t.Tuple[ 2783 t.Optional[exp.Expr], 2784 t.List[exp.Expr], 2785 t.List[exp.Expr], 2786 t.List[exp.Expr], 2787 UniqueKeyDict[str, ModelAudit], 2788]: 2789 """Extracts the SELECT query from a sequence of expressions. 2790 2791 Args: 2792 expressions: The list of all SQL statements in the model definition. 2793 2794 Returns: 2795 A tuple containing the extracted SELECT query or the `@INSERT_SEED()` call, the statements before the it, 2796 the statements after it, and the inline audit definitions. 2797 2798 Raises: 2799 ConfigError: If the model definition contains more than one SELECT query or `@INSERT_SEED()` call. 2800 """ 2801 from sqlmesh.core.audit import ModelAudit, load_audit 2802 2803 query_positions = [] 2804 sql_statements = [] 2805 on_virtual_update = [] 2806 inline_audits: UniqueKeyDict[str, ModelAudit] = UniqueKeyDict("inline_audits") 2807 2808 idx = 0 2809 length = len(expressions) 2810 while idx < length: 2811 expr = expressions[idx] 2812 2813 if isinstance(expr, d.Audit): 2814 loaded_audit = load_audit([expr, expressions[idx + 1]], dialect=dialect) 2815 assert isinstance(loaded_audit, ModelAudit) 2816 inline_audits[loaded_audit.name] = loaded_audit 2817 idx += 2 2818 elif isinstance(expr, d.VirtualUpdateStatement): 2819 for statement in expr.expressions: 2820 on_virtual_update.append(statement) 2821 idx += 1 2822 else: 2823 if ( 2824 isinstance(expr, (exp.Query, d.JinjaQuery)) 2825 or expr == INSERT_SEED_MACRO_CALL 2826 or ( 2827 isinstance(expr, d.MacroFunc) 2828 and (expr.this.name.lower() == "union" or length == 1) 2829 ) 2830 ): 2831 query_positions.append((expr, idx)) 2832 sql_statements.append(expr) 2833 idx += 1 2834 2835 if not query_positions: 2836 return None, sql_statements, [], on_virtual_update, inline_audits 2837 2838 if len(query_positions) > 1: 2839 raise_config_error("Only one SELECT query is allowed per model", path) 2840 2841 query, pos = query_positions[0] 2842 return query, sql_statements[:pos], sql_statements[pos + 1 :], on_virtual_update, inline_audits 2843 2844 2845def _resolve_model_refs_to_physical_tables( 2846 value: exp.Expr, table_mapping: t.Dict[str, str], dialect: DialectType 2847) -> exp.Literal: 2848 """Resolve managed-model references in a property value to their physical table names. 2849 2850 The value is a single table reference or a comma-separated list of them. Each reference that 2851 matches a managed model (via ``table_mapping``) is swapped for its physical ``db.table`` name; 2852 anything else (e.g. a raw source) is kept as written. Returns a single string literal so the 2853 property renders just like a hand-written value. 2854 """ 2855 if isinstance(value, exp.Literal) and value.is_string: 2856 refs = value.this.split(",") 2857 else: 2858 refs = [value.sql(dialect=dialect)] 2859 2860 def resolve(ref: str) -> str: 2861 table = exp.to_table(ref.strip(), dialect=dialect) 2862 physical = table_mapping.get(exp.table_name(table, identify=True)) 2863 # Managed model -> physical table; otherwise keep the reference (just unquoted/normalized). 2864 return exp.table_name( 2865 exp.to_table(physical, dialect=dialect) if physical else table, identify=False 2866 ) 2867 2868 return exp.Literal.string(",".join(resolve(ref) for ref in refs if ref.strip())) 2869 2870 2871def _resolve_properties( 2872 default: t.Optional[t.Dict[str, t.Any]], 2873 provided: t.Optional[exp.Expr | t.Dict[str, t.Any]], 2874) -> t.Optional[exp.Expr]: 2875 if isinstance(provided, dict): 2876 properties = {k: exp.Literal.string(k).eq(v) for k, v in provided.items()} 2877 elif provided: 2878 if isinstance(provided, exp.Paren): 2879 provided = exp.Tuple(expressions=[provided.this]) 2880 properties = {expr.this.name: expr for expr in provided} 2881 else: 2882 properties = {} 2883 2884 for k, v in (default or {}).items(): 2885 if k not in properties: 2886 properties[k] = exp.Literal.string(k).eq(v) 2887 elif properties[k].expression.sql().lower() in {"none", "null"}: 2888 del properties[k] 2889 2890 if properties: 2891 return exp.Tuple(expressions=list(properties.values())) 2892 2893 return None 2894 2895 2896def _list_of_calls_to_exp(value: t.List[t.Tuple[str, t.Dict[str, t.Any]]]) -> exp.Expr: 2897 return exp.Tuple( 2898 expressions=[ 2899 exp.Anonymous( 2900 this=v[0], 2901 expressions=[ 2902 exp.EQ(this=exp.convert(left), expression=exp.convert(right)) 2903 for left, right in v[1].items() 2904 ], 2905 ) 2906 for v in value 2907 ] 2908 ) 2909 2910 2911def _is_projection(expr: exp.Expr) -> bool: 2912 parent = expr.parent 2913 return isinstance(parent, exp.Select) and expr.arg_key == "expressions" 2914 2915 2916def _has_ordinal_references(query: exp.Select) -> bool: 2917 order = query.args.get("order") 2918 if order and any( 2919 isinstance(ob.this, exp.Literal) and ob.this.is_number for ob in order.expressions 2920 ): 2921 return True 2922 group = query.args.get("group") 2923 return bool( 2924 group and any(isinstance(gb, exp.Literal) and gb.is_number for gb in group.expressions) 2925 ) 2926 2927 2928def _additive_projection_change( 2929 previous_query: exp.Query, 2930 this_query: exp.Query, 2931 dialect: DialectType, 2932) -> t.Optional[bool]: 2933 """Fallback for when SQLGlot's tree diff can't express an additive projection change. 2934 2935 SQLGlot's diff matches nodes by structural similarity, so interchangeable leaves (e.g. two 2936 identical ``CAST(... AS T)`` target types) can be cross-matched. Inserting a same-type cast 2937 above an existing one therefore yields spurious ``Move`` / ``Update`` edits even though a 2938 column was simply added to the SELECT list. In that case the edit-based check above is 2939 inconclusive, so we verify additivity directly against the output projections. 2940 2941 Returns ``False`` (non-breaking) only when the change is provably additive: 2942 * both queries are simple ``SELECT`` statements, 2943 * everything other than the projection list is structurally identical, 2944 * no added projection is a (potentially cardinality-changing) ``UDTF``, 2945 * every previous projection is preserved, in order, within the new projection list, and 2946 * no mid-list insert shifts ordinal ``ORDER BY`` / ``GROUP BY`` references. 2947 2948 Otherwise returns ``None`` (undetermined), preserving the conservative default. 2949 """ 2950 # UNIONs or other query expressions, are left to the caller's conservative diff result. 2951 if not isinstance(previous_query, exp.Select) or not isinstance(this_query, exp.Select): 2952 return None 2953 2954 previous_projections = previous_query.expressions 2955 this_projections = this_query.expressions 2956 # If the new query has not gained any projections, this cannot be an additive projection-only 2957 # change, so there is nothing for this fallback to prove. 2958 if len(this_projections) <= len(previous_projections): 2959 return None 2960 2961 # Adding a UDTF projection (e.g. EXPLODE / UNNEST) can change row cardinality, so such a 2962 # change is not safely non-breaking even when it appears as an extra SELECT item. 2963 for projection in this_projections: 2964 bare = projection.this if isinstance(projection, exp.Alias) else projection 2965 if isinstance(bare, exp.UDTF): 2966 return None 2967 2968 # Everything other than the projection list must be structurally identical. Replacing each 2969 # SELECT list with the same dummy literal lets the expression equality check focus on the 2970 # FROM / WHERE / GROUP BY / ORDER BY / etc. parts of the query. 2971 previous_skeleton = previous_query.copy() 2972 this_skeleton = this_query.copy() 2973 previous_skeleton.set("expressions", [exp.Literal.number(1)]) 2974 this_skeleton.set("expressions", [exp.Literal.number(1)]) 2975 if previous_skeleton != this_skeleton: 2976 return None 2977 2978 # Every previous projection must appear, in order, within the new projection list. Comparing 2979 # dialect-normalized SQL makes semantically equivalent projection nodes match even when the 2980 # parser built distinct object identities. 2981 this_projection_sql = [p.sql(dialect=dialect, comments=False) for p in this_projections] 2982 search_start = 0 2983 matched_at: list[int] = [] 2984 for projection in previous_projections: 2985 target_sql = projection.sql(dialect=dialect, comments=False) 2986 # Continue after the previous match so added columns can appear before, between, or after 2987 # the original projections, but existing projections cannot be reordered or rewritten. 2988 for index in range(search_start, len(this_projection_sql)): 2989 if this_projection_sql[index] == target_sql: 2990 matched_at.append(index) 2991 search_start = index + 1 2992 break 2993 else: 2994 return None 2995 2996 # Mid-list inserts shift ordinal references in ORDER BY / GROUP BY clauses. 2997 if _has_ordinal_references(this_query): 2998 matched_set = set(matched_at) 2999 last_matched = matched_at[-1] 3000 if any(i < last_matched for i in range(len(this_projections)) if i not in matched_set): 3001 return None 3002 3003 # At this point the query shape is unchanged and all prior outputs are preserved, so the only 3004 # remaining difference is one or more additional, non-UDTF projections. 3005 return False 3006 3007 3008def _single_expr_or_tuple(values: t.Sequence[exp.Expr]) -> exp.Expr | exp.Tuple: 3009 return values[0] if len(values) == 1 else exp.Tuple(expressions=values) 3010 3011 3012def _refs_to_sql(values: t.Any) -> exp.Expr: 3013 return exp.Tuple(expressions=values) 3014 3015 3016def render_meta_fields( 3017 fields: t.Dict[str, t.Any], 3018 module_path: Path, 3019 path: t.Optional[Path], 3020 jinja_macros: t.Optional[JinjaMacroRegistry], 3021 macros: t.Optional[MacroRegistry], 3022 dialect: DialectType, 3023 variables: t.Optional[t.Dict[str, t.Any]], 3024 default_catalog: t.Optional[str], 3025 blueprint_variables: t.Optional[t.Dict[str, t.Any]] = None, 3026) -> t.Dict[str, t.Any]: 3027 def render_field_value(value: t.Any) -> t.Any: 3028 if isinstance(value, exp.Expr) or (isinstance(value, str) and "@" in value): 3029 expression = exp.maybe_parse(value, dialect=dialect) 3030 rendered_expr = render_expression( 3031 expression=expression, 3032 module_path=module_path, 3033 macros=macros, 3034 jinja_macros=jinja_macros, 3035 variables=variables, 3036 path=path, 3037 dialect=dialect, 3038 default_catalog=default_catalog, 3039 blueprint_variables=blueprint_variables, 3040 ) 3041 if not rendered_expr: 3042 raise SQLMeshError( 3043 f"Rendering `{expression.sql(dialect=dialect)}` did not return an expression" 3044 ) 3045 3046 if len(rendered_expr) != 1: 3047 raise SQLMeshError( 3048 f"Rendering `{expression.sql(dialect=dialect)}` must return one result, but got {len(rendered_expr)}" 3049 ) 3050 3051 # For cases where a property is conditionally assigned 3052 if rendered_expr[0].sql().lower() in {"none", "null"}: 3053 return None 3054 3055 return rendered_expr[0] 3056 3057 return value 3058 3059 for field_name, field_info in ModelMeta.all_field_infos().items(): 3060 field = field_info.alias or field_name 3061 field_value = fields.get(field) 3062 3063 # We don't want to parse python model cron="@..." kwargs (e.g. @daily) into MacroVar 3064 if ( 3065 field == "cron" 3066 and isinstance(field_value, str) 3067 and field_value.lower() in CRON_SHORTCUTS 3068 ) or field_value is None: 3069 continue 3070 3071 if field in RUNTIME_RENDERED_MODEL_FIELDS: 3072 fields[field] = parse_strings_with_macro_refs(field_value, dialect) 3073 continue 3074 3075 if isinstance(field_value, dict): 3076 rendered_dict = {} 3077 for key, value in field_value.items(): 3078 if field == "columns": 3079 column_name = render_field_value(key) 3080 column_type = render_field_value(value) 3081 # If column_type is an Expr (from rendering macros), convert to string. 3082 # Otherwise, leave it as-is (string) for the validator to parse with the correct dialect. 3083 if isinstance(column_type, exp.Expr): 3084 column_type = column_type.sql(dialect=dialect) 3085 rendered_dict[column_name] = column_type 3086 elif key in RUNTIME_RENDERED_MODEL_FIELDS: 3087 rendered_dict[key] = parse_strings_with_macro_refs(value, dialect) 3088 elif ( 3089 # don't parse kind auto_restatement_cron="@..." kwargs (e.g. @daily) into MacroVar 3090 key == "auto_restatement_cron" 3091 and isinstance(value, str) 3092 and value.lower() in CRON_SHORTCUTS 3093 ): 3094 rendered_dict[key] = value 3095 elif (rendered := render_field_value(value)) is not None: 3096 rendered_dict[key] = rendered 3097 3098 if rendered_dict: 3099 fields[field] = rendered_dict 3100 else: 3101 fields.pop(field) 3102 elif isinstance(field_value, list): 3103 rendered_list = [ 3104 rendered 3105 for value in field_value 3106 if (rendered := render_field_value(value)) is not None 3107 ] 3108 if rendered_list: 3109 fields[field] = rendered_list 3110 else: 3111 fields.pop(field) 3112 else: 3113 rendered_field = render_field_value(field_value) 3114 if rendered_field is not None: 3115 fields[field] = rendered_field 3116 else: 3117 fields.pop(field) 3118 3119 return fields 3120 3121 3122def render_model_defaults( 3123 defaults: t.Dict[str, t.Any], 3124 module_path: Path, 3125 path: t.Optional[Path], 3126 jinja_macros: t.Optional[JinjaMacroRegistry], 3127 macros: t.Optional[MacroRegistry], 3128 dialect: DialectType, 3129 variables: t.Optional[t.Dict[str, t.Any]], 3130 default_catalog: t.Optional[str], 3131) -> t.Dict[str, t.Any]: 3132 rendered_defaults = render_meta_fields( 3133 fields=defaults, 3134 module_path=module_path, 3135 macros=macros, 3136 jinja_macros=jinja_macros, 3137 variables=variables, 3138 path=path, 3139 dialect=dialect, 3140 default_catalog=default_catalog, 3141 ) 3142 3143 # Validate defaults that have macros are rendered to boolean 3144 for boolean in {"optimize_query", "allow_partials", "enabled"}: 3145 var = rendered_defaults.get(boolean) 3146 if var is not None and not isinstance(var, (exp.Boolean, bool)): 3147 raise ConfigError(f"Expected boolean for '{var}', got '{type(var)}' instead") 3148 3149 # Validate the 'interval_unit' if present is an Interval Unit 3150 var = rendered_defaults.get("interval_unit") 3151 if isinstance(var, str): 3152 try: 3153 rendered_defaults["interval_unit"] = IntervalUnit(var) 3154 except ValueError as e: 3155 raise ConfigError(f"Invalid interval unit: {var}") from e 3156 3157 return rendered_defaults 3158 3159 3160def parse_defaults_properties( 3161 defaults: t.Dict[str, t.Any], dialect: DialectType 3162) -> t.Dict[str, t.Any]: 3163 for prop in PROPERTIES: 3164 default_properties = defaults.get(prop) 3165 for key, value in (default_properties or {}).items(): 3166 if isinstance(key, str) and d.SQLMESH_MACRO_PREFIX in str(value): 3167 defaults[prop][key] = exp.maybe_parse(value, dialect=dialect) 3168 3169 return defaults 3170 3171 3172def render_expression( 3173 expression: exp.Expr, 3174 module_path: Path, 3175 path: t.Optional[Path], 3176 jinja_macros: t.Optional[JinjaMacroRegistry] = None, 3177 macros: t.Optional[MacroRegistry] = None, 3178 dialect: DialectType = None, 3179 variables: t.Optional[t.Dict[str, t.Any]] = None, 3180 default_catalog: t.Optional[str] = None, 3181 blueprint_variables: t.Optional[t.Dict[str, t.Any]] = None, 3182) -> t.Optional[t.List[exp.Expr]]: 3183 meta_python_env = make_python_env( 3184 expressions=expression, 3185 jinja_macro_references=None, 3186 module_path=module_path, 3187 macros=macros or macro.get_registry(), 3188 variables=variables, 3189 path=path, 3190 blueprint_variables=blueprint_variables, 3191 ) 3192 return ExpressionRenderer( 3193 expression, 3194 dialect, 3195 [], 3196 path=path, 3197 jinja_macro_registry=jinja_macros, 3198 python_env=meta_python_env, 3199 default_catalog=default_catalog, 3200 quote_identifiers=False, 3201 normalize_identifiers=False, 3202 ).render() 3203 3204 3205META_FIELD_CONVERTER: t.Dict[str, t.Callable] = { 3206 "start": lambda value: exp.Literal.string(value), 3207 "cron": lambda value: exp.Literal.string(value), 3208 "cron_tz": lambda value: exp.Literal.string(value), 3209 "partitioned_by_": _single_expr_or_tuple, 3210 "clustered_by": _single_expr_or_tuple, 3211 "depends_on_": lambda value: exp.Tuple(expressions=sorted(value)), 3212 "pre": _list_of_calls_to_exp, 3213 "post": _list_of_calls_to_exp, 3214 "audits": _list_of_calls_to_exp, 3215 "columns_to_types_": lambda value: exp.Schema( 3216 expressions=[exp.ColumnDef(this=exp.to_column(c), kind=t) for c, t in value.items()] 3217 ), 3218 "column_descriptions_": lambda value: exp.Schema( 3219 expressions=[exp.to_column(c).eq(d) for c, d in value.items()] 3220 ), 3221 "tags": single_value_or_tuple, 3222 "grains": _refs_to_sql, 3223 "references": _refs_to_sql, 3224 "physical_properties_": lambda value: value, 3225 "virtual_properties_": lambda value: value, 3226 "session_properties_": lambda value: value, 3227 "allow_partials": exp.convert, 3228 "signals": lambda values: exp.tuple_( 3229 *( 3230 exp.func( 3231 name, *(exp.PropertyEQ(this=exp.var(k), expression=v) for k, v in args.items()) 3232 ) 3233 if name 3234 else exp.Tuple(expressions=[exp.var(k).eq(v) for k, v in args.items()]) 3235 for name, args in values 3236 ) 3237 ), 3238 "formatting": str, 3239 "optimize_query": str, 3240 "virtual_environment_mode": lambda value: exp.Literal.string(value.value), 3241 "dbt_node_info_": lambda value: value.to_expression(), 3242 "grants_": lambda value: value, 3243 "grants_target_layer": lambda value: exp.Literal.string(value.value), 3244} 3245 3246 3247def get_model_name(path: Path) -> str: 3248 path_parts = list(path.parts[path.parts.index("models") + 1 : -1]) + [path.stem] 3249 return ".".join(path_parts[-3:]) 3250 3251 3252# function applied to time column when automatically used for partitioning in INCREMENTAL_BY_TIME_RANGE models 3253def clickhouse_partition_func( 3254 column: exp.Expr, columns_to_types: t.Optional[t.Dict[str, exp.DataType]] 3255) -> exp.Expr: 3256 # `toMonday()` function accepts a Date or DateTime type column 3257 3258 col_type = (columns_to_types and columns_to_types.get(column.name)) or exp.DataType.build( 3259 "UNKNOWN" 3260 ) 3261 col_type_is_conformable = col_type.is_type( 3262 exp.DataType.Type.DATE, 3263 exp.DataType.Type.DATE32, 3264 exp.DataType.Type.DATETIME, 3265 exp.DataType.Type.DATETIME64, 3266 ) 3267 3268 # if input column is already a conformable type, just pass the column 3269 if col_type_is_conformable: 3270 return exp.func("toMonday", column, dialect="clickhouse") 3271 3272 # if input column type is not known, cast input to DateTime64 3273 if col_type.is_type(exp.DataType.Type.UNKNOWN): 3274 return exp.func( 3275 "toMonday", 3276 exp.cast(column, exp.DataType.build("DateTime64(9, 'UTC')", dialect="clickhouse")), 3277 dialect="clickhouse", 3278 ) 3279 3280 # if input column type is known but not conformable, cast input to DateTime64 and cast output back to original type 3281 return exp.cast( 3282 exp.func( 3283 "toMonday", 3284 exp.cast(column, exp.DataType.build("DateTime64(9, 'UTC')", dialect="clickhouse")), 3285 dialect="clickhouse", 3286 ), 3287 col_type, 3288 ) 3289 3290 3291TIME_COL_PARTITION_FUNC = {"clickhouse": clickhouse_partition_func}
logger =
<Logger sqlmesh.core.model.definition (WARNING)>
PROPERTIES =
{'physical_properties', 'virtual_properties', 'session_properties'}
RUNTIME_RENDERED_MODEL_FIELDS =
{'physical_properties', 'merge_filter', 'virtual_properties', 'session_properties', 'audits', 'signals'}
CRON_SHORTCUTS =
{'@weekly', '@hourly', '@annually', '@daily', '@midnight', '@monthly', '@yearly'}
1383class SqlModel(_Model): 1384 """The model definition which relies on a SQL query to fetch the data. 1385 1386 Args: 1387 query: The main query representing the model. 1388 pre_statements: The list of SQL statements that precede the model's query. 1389 post_statements: The list of SQL statements that follow after the model's query. 1390 on_virtual_update: The list of SQL statements to be executed after the virtual update. 1391 """ 1392 1393 query_: ParsableSql = Field(alias="query") 1394 source_type: t.Literal["sql"] = "sql" 1395 1396 _columns_to_types: t.Optional[t.Dict[str, exp.DataType]] = None 1397 1398 def __getstate__(self) -> t.Dict[t.Any, t.Any]: 1399 state = super().__getstate__() 1400 state["__dict__"] = state["__dict__"].copy() 1401 # query renderer is very expensive to serialize 1402 state["__dict__"].pop("_query_renderer", None) 1403 state["__dict__"].pop("column_descriptions", None) 1404 private = state[PRIVATE_FIELDS] 1405 private["_columns_to_types"] = None 1406 return state 1407 1408 def copy(self, **kwargs: t.Any) -> Self: 1409 model = super().copy(**kwargs) 1410 model.__dict__.pop("_query_renderer", None) 1411 model.__dict__.pop("column_descriptions", None) 1412 model._columns_to_types = None 1413 if kwargs.get("update", {}).keys() & {"depends_on_", "query"}: 1414 model._full_depends_on = None 1415 return model 1416 1417 @property 1418 def query(self) -> t.Union[exp.Query, d.JinjaQuery, d.MacroFunc]: 1419 parsed_query = self.query_.parse(self.dialect) 1420 return t.cast(t.Union[exp.Query, d.JinjaQuery, d.MacroFunc], parsed_query) 1421 1422 def render_query( 1423 self, 1424 *, 1425 start: t.Optional[TimeLike] = None, 1426 end: t.Optional[TimeLike] = None, 1427 execution_time: t.Optional[TimeLike] = None, 1428 snapshots: t.Optional[t.Dict[str, Snapshot]] = None, 1429 table_mapping: t.Optional[t.Dict[str, str]] = None, 1430 expand: t.Iterable[str] = tuple(), 1431 deployability_index: t.Optional[DeployabilityIndex] = None, 1432 engine_adapter: t.Optional[EngineAdapter] = None, 1433 **kwargs: t.Any, 1434 ) -> t.Optional[exp.Query]: 1435 query = self._query_renderer.render( 1436 start=start, 1437 end=end, 1438 execution_time=execution_time, 1439 snapshots=snapshots, 1440 table_mapping=table_mapping, 1441 expand=expand, 1442 deployability_index=deployability_index, 1443 engine_adapter=engine_adapter, 1444 **kwargs, 1445 ) 1446 1447 return query 1448 1449 def render_definition( 1450 self, 1451 include_python: bool = True, 1452 include_defaults: bool = False, 1453 render_query: bool = False, 1454 ) -> t.List[exp.Expr]: 1455 result: t.List[exp.Expr] = super().render_definition( 1456 include_python=include_python, include_defaults=include_defaults 1457 ) 1458 1459 if render_query: 1460 result.extend(self.render_pre_statements()) 1461 result.append(self.render_query() or self.query) 1462 result.extend(self.render_post_statements()) 1463 if virtual_update := self.render_on_virtual_update(): 1464 result.append(d.VirtualUpdateStatement(expressions=virtual_update)) 1465 else: 1466 result.extend(self.pre_statements) 1467 result.append(self.query) 1468 result.extend(self.post_statements) 1469 if self.on_virtual_update: 1470 result.append(d.VirtualUpdateStatement(expressions=self.on_virtual_update)) 1471 1472 return result 1473 1474 @property 1475 def is_sql(self) -> bool: 1476 return True 1477 1478 @property 1479 def columns_to_types(self) -> t.Optional[t.Dict[str, exp.DataType]]: 1480 if self.columns_to_types_ is not None: 1481 self._columns_to_types = self.columns_to_types_ 1482 elif self._columns_to_types is None: 1483 try: 1484 query = self._query_renderer.render() 1485 except Exception: 1486 logger.exception("Failed to render query for model %s", self.fqn) 1487 return None 1488 1489 if query is None: 1490 return None 1491 1492 unknown = exp.DataType.build("unknown") 1493 1494 columns_to_types = {} 1495 for select in query.selects: 1496 output_name = select.output_name 1497 1498 # If model validation is disabled, we cannot assume that projections 1499 # will have inferrable output names or even that they will be unique 1500 if not output_name or output_name in columns_to_types: 1501 return None 1502 1503 # copy data type because it is used in the engine to build CTAS and other queries 1504 # this can change the parent which will mess up the diffing algo 1505 columns_to_types[output_name] = (select.type or unknown).copy() 1506 1507 self._columns_to_types = columns_to_types 1508 1509 if "*" in self._columns_to_types: 1510 return None 1511 1512 return {**self._columns_to_types, **self.managed_columns} 1513 1514 @cached_property 1515 def column_descriptions(self) -> t.Dict[str, str]: 1516 if self.column_descriptions_ is not None: 1517 return self.column_descriptions_ 1518 1519 query = self.render_query() 1520 if query is None: 1521 return {} 1522 1523 return { 1524 select.alias_or_name: select.comments[-1].strip() 1525 for select in query.selects 1526 if select.comments 1527 } 1528 1529 def set_mapping_schema(self, schema: t.Dict) -> None: 1530 super().set_mapping_schema(schema) 1531 self._on_mapping_schema_set() 1532 1533 def update_schema(self, schema: MappingSchema) -> None: 1534 super().update_schema(schema) 1535 self._on_mapping_schema_set() 1536 1537 def _on_mapping_schema_set(self) -> None: 1538 self._columns_to_types = None 1539 self._query_renderer.update_schema(self.mapping_schema) 1540 1541 def validate_definition(self) -> None: 1542 query = self._query_renderer.render() 1543 if query is None: 1544 if self.depends_on_ is None: 1545 raise_config_error( 1546 "Dependencies must be provided explicitly for models that can be rendered only at runtime", 1547 self._path, 1548 ) 1549 return 1550 1551 if not isinstance(query, exp.Query): 1552 raise_config_error("Missing SELECT query in the model definition", self._path) 1553 1554 projection_list = query.selects 1555 if not projection_list: 1556 raise_config_error("Query missing select statements", self._path) 1557 1558 if self.depends_on_self and not self.annotated: 1559 raise_config_error( 1560 "Self-referencing models require inferrable column types. There are three options available to mitigate this issue: add explicit types to all projections in the outermost SELECT statement, leverage external models (https://sqlmesh.readthedocs.io/en/stable/concepts/models/external_models/), or use the `columns` model attribute (https://sqlmesh.readthedocs.io/en/stable/concepts/models/overview/#columns).", 1561 self._path, 1562 ) 1563 1564 super().validate_definition() 1565 1566 def is_breaking_change(self, previous: Model) -> t.Optional[bool]: 1567 if not isinstance(previous, SqlModel): 1568 return None 1569 1570 if self.lookback != previous.lookback: 1571 return None 1572 1573 try: 1574 # the previous model which comes from disk could be unrenderable 1575 previous_query = previous.render_query() 1576 except Exception: 1577 previous_query = None 1578 this_query = self.render_query() 1579 1580 if previous_query is None or this_query is None: 1581 # Can't determine if there's a breaking change if we can't render the query. 1582 return None 1583 1584 if previous_query is this_query: 1585 edits = [] 1586 else: 1587 edits = diff( 1588 previous_query, 1589 this_query, 1590 matchings=[(previous_query, this_query)], 1591 delta_only=True, 1592 dialect=self.dialect if self.dialect == previous.dialect else None, 1593 ) 1594 inserted_expressions = {e.expression for e in edits if isinstance(e, Insert)} 1595 1596 for edit in edits: 1597 if not isinstance(edit, Insert): 1598 return _additive_projection_change(previous_query, this_query, self.dialect) 1599 1600 expr = edit.expression 1601 if isinstance(expr, exp.UDTF): 1602 # projection subqueries do not change cardinality, engines don't allow these to return 1603 # more than one row of data 1604 parent = expr.find_ancestor(exp.Subquery) 1605 1606 if not parent: 1607 return None 1608 1609 expr = parent 1610 1611 if not _is_projection(expr) and expr.parent not in inserted_expressions: 1612 return _additive_projection_change(previous_query, this_query, self.dialect) 1613 1614 return False 1615 1616 def is_metadata_only_change(self, previous: _Node) -> bool: 1617 if self._is_metadata_only_change_cache.get(id(previous), None) is not None: 1618 return self._is_metadata_only_change_cache[id(previous)] 1619 1620 if not super().is_metadata_only_change(previous): 1621 return False 1622 1623 if not isinstance(previous, SqlModel): 1624 self._is_metadata_only_change_cache[id(previous)] = False 1625 return False 1626 1627 this_rendered_query = self.render_query() or self.query 1628 previous_rendered_query = previous.render_query() or previous.query 1629 is_metadata_change = this_rendered_query == previous_rendered_query 1630 1631 self._is_metadata_only_change_cache[id(previous)] = is_metadata_change 1632 return is_metadata_change 1633 1634 @cached_property 1635 def _query_renderer(self) -> QueryRenderer: 1636 no_quote_identifiers = self.kind.is_view and self.dialect in ("trino", "spark") 1637 return QueryRenderer( 1638 self.query, 1639 self.dialect, 1640 self.macro_definitions, 1641 schema=self.mapping_schema, 1642 path=self._path, 1643 jinja_macro_registry=self.jinja_macros, 1644 python_env=self.python_env, 1645 only_execution_time=self.kind.only_execution_time, 1646 default_catalog=self.default_catalog, 1647 quote_identifiers=not no_quote_identifiers, 1648 optimize_query=self.optimize_query, 1649 model=self, 1650 ) 1651 1652 @property 1653 def _data_hash_values_no_sql(self) -> t.List[str]: 1654 return [ 1655 *super()._data_hash_values_no_sql, 1656 *self.jinja_macros.data_hash_values, 1657 ] 1658 1659 @property 1660 def _data_hash_values_sql(self) -> t.List[str]: 1661 return [ 1662 *super()._data_hash_values_sql, 1663 self.query_.sql, 1664 ] 1665 1666 @property 1667 def _additional_metadata(self) -> t.List[str]: 1668 return [*super()._additional_metadata, self.query_.sql] 1669 1670 @property 1671 def violated_rules_for_query(self) -> t.Dict[type[Rule], t.Any]: 1672 self.render_query() 1673 return self._query_renderer._violated_rules
The model definition which relies on a SQL query to fetch the data.
Arguments:
- query: The main query representing the model.
- pre_statements: The list of SQL statements that precede the model's query.
- post_statements: The list of SQL statements that follow after the model's query.
- on_virtual_update: The list of SQL statements to be executed after the virtual update.
def
copy(self, **kwargs: Any) -> typing_extensions.Self:
1408 def copy(self, **kwargs: t.Any) -> Self: 1409 model = super().copy(**kwargs) 1410 model.__dict__.pop("_query_renderer", None) 1411 model.__dict__.pop("column_descriptions", None) 1412 model._columns_to_types = None 1413 if kwargs.get("update", {}).keys() & {"depends_on_", "query"}: 1414 model._full_depends_on = None 1415 return model
Returns a copy of the model.
!!! warning "Deprecated"
This method is now deprecated; use model_copy instead.
If you need include or exclude, use:
python {test="skip" lint="skip"}
data = self.model_dump(include=include, exclude=exclude, round_trip=True)
data = {**data, **(update or {})}
copied = self.model_validate(data)
Arguments:
- include: Optional set or mapping specifying which fields to include in the copied model.
- exclude: Optional set or mapping specifying which fields to exclude in the copied model.
- update: Optional dictionary of field-value pairs to override field values in the copied model.
- deep: If True, the values of fields that are Pydantic models will be deep-copied.
Returns:
A copy of the model with included, excluded and updated fields as specified.
query: Union[sqlglot.expressions.query.Query, sqlmesh.core.dialect.JinjaQuery, sqlmesh.core.dialect.MacroFunc]
def
render_query( self, *, 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, snapshots: Optional[Dict[str, <MagicMock id='130969753143136'>]] = None, table_mapping: Optional[Dict[str, str]] = None, expand: Iterable[str] = (), deployability_index: Optional[<MagicMock id='130969753024064'>] = None, engine_adapter: Optional[sqlmesh.core.engine_adapter.base.EngineAdapter] = None, **kwargs: Any) -> Optional[sqlglot.expressions.query.Query]:
1422 def render_query( 1423 self, 1424 *, 1425 start: t.Optional[TimeLike] = None, 1426 end: t.Optional[TimeLike] = None, 1427 execution_time: t.Optional[TimeLike] = None, 1428 snapshots: t.Optional[t.Dict[str, Snapshot]] = None, 1429 table_mapping: t.Optional[t.Dict[str, str]] = None, 1430 expand: t.Iterable[str] = tuple(), 1431 deployability_index: t.Optional[DeployabilityIndex] = None, 1432 engine_adapter: t.Optional[EngineAdapter] = None, 1433 **kwargs: t.Any, 1434 ) -> t.Optional[exp.Query]: 1435 query = self._query_renderer.render( 1436 start=start, 1437 end=end, 1438 execution_time=execution_time, 1439 snapshots=snapshots, 1440 table_mapping=table_mapping, 1441 expand=expand, 1442 deployability_index=deployability_index, 1443 engine_adapter=engine_adapter, 1444 **kwargs, 1445 ) 1446 1447 return query
Renders a model's query, expanding macros with provided kwargs, and optionally expanding referenced models.
Arguments:
- start: The start datetime to render. Defaults to epoch start.
- end: The end datetime to render. Defaults to epoch start.
- execution_time: The date/time time reference to use for execution time.
- snapshots: All upstream snapshots (by name) to use for expansion and mapping of physical locations.
- table_mapping: Table mapping of physical locations. Takes precedence over snapshot mappings.
- expand: Expand referenced models as subqueries. This is used to bypass backfills when running queries that depend on materialized tables. Model definitions are inlined and can thus be run end to end on the fly.
- deployability_index: Determines snapshots that are deployable in the context of this render.
- kwargs: Additional kwargs to pass to the renderer.
Returns:
The rendered expression.
def
render_definition( self, include_python: bool = True, include_defaults: bool = False, render_query: bool = False) -> List[sqlglot.expressions.core.Expr]:
1449 def render_definition( 1450 self, 1451 include_python: bool = True, 1452 include_defaults: bool = False, 1453 render_query: bool = False, 1454 ) -> t.List[exp.Expr]: 1455 result: t.List[exp.Expr] = super().render_definition( 1456 include_python=include_python, include_defaults=include_defaults 1457 ) 1458 1459 if render_query: 1460 result.extend(self.render_pre_statements()) 1461 result.append(self.render_query() or self.query) 1462 result.extend(self.render_post_statements()) 1463 if virtual_update := self.render_on_virtual_update(): 1464 result.append(d.VirtualUpdateStatement(expressions=virtual_update)) 1465 else: 1466 result.extend(self.pre_statements) 1467 result.append(self.query) 1468 result.extend(self.post_statements) 1469 if self.on_virtual_update: 1470 result.append(d.VirtualUpdateStatement(expressions=self.on_virtual_update)) 1471 1472 return result
Returns the original list of sql expressions comprising the model definition.
Arguments:
- include_python: Whether or not to include Python code in the rendered definition.
columns_to_types: Optional[Dict[str, sqlglot.expressions.datatypes.DataType]]
1478 @property 1479 def columns_to_types(self) -> t.Optional[t.Dict[str, exp.DataType]]: 1480 if self.columns_to_types_ is not None: 1481 self._columns_to_types = self.columns_to_types_ 1482 elif self._columns_to_types is None: 1483 try: 1484 query = self._query_renderer.render() 1485 except Exception: 1486 logger.exception("Failed to render query for model %s", self.fqn) 1487 return None 1488 1489 if query is None: 1490 return None 1491 1492 unknown = exp.DataType.build("unknown") 1493 1494 columns_to_types = {} 1495 for select in query.selects: 1496 output_name = select.output_name 1497 1498 # If model validation is disabled, we cannot assume that projections 1499 # will have inferrable output names or even that they will be unique 1500 if not output_name or output_name in columns_to_types: 1501 return None 1502 1503 # copy data type because it is used in the engine to build CTAS and other queries 1504 # this can change the parent which will mess up the diffing algo 1505 columns_to_types[output_name] = (select.type or unknown).copy() 1506 1507 self._columns_to_types = columns_to_types 1508 1509 if "*" in self._columns_to_types: 1510 return None 1511 1512 return {**self._columns_to_types, **self.managed_columns}
Returns the mapping of column names to types of this model.
column_descriptions: Dict[str, str]
1514 @cached_property 1515 def column_descriptions(self) -> t.Dict[str, str]: 1516 if self.column_descriptions_ is not None: 1517 return self.column_descriptions_ 1518 1519 query = self.render_query() 1520 if query is None: 1521 return {} 1522 1523 return { 1524 select.alias_or_name: select.comments[-1].strip() 1525 for select in query.selects 1526 if select.comments 1527 }
A dictionary of column names to annotation comments.
def
update_schema(self, schema: sqlglot.schema.MappingSchema) -> None:
1533 def update_schema(self, schema: MappingSchema) -> None: 1534 super().update_schema(schema) 1535 self._on_mapping_schema_set()
Updates the schema for this model's dependencies based on the given mapping schema.
def
validate_definition(self) -> None:
1541 def validate_definition(self) -> None: 1542 query = self._query_renderer.render() 1543 if query is None: 1544 if self.depends_on_ is None: 1545 raise_config_error( 1546 "Dependencies must be provided explicitly for models that can be rendered only at runtime", 1547 self._path, 1548 ) 1549 return 1550 1551 if not isinstance(query, exp.Query): 1552 raise_config_error("Missing SELECT query in the model definition", self._path) 1553 1554 projection_list = query.selects 1555 if not projection_list: 1556 raise_config_error("Query missing select statements", self._path) 1557 1558 if self.depends_on_self and not self.annotated: 1559 raise_config_error( 1560 "Self-referencing models require inferrable column types. There are three options available to mitigate this issue: add explicit types to all projections in the outermost SELECT statement, leverage external models (https://sqlmesh.readthedocs.io/en/stable/concepts/models/external_models/), or use the `columns` model attribute (https://sqlmesh.readthedocs.io/en/stable/concepts/models/overview/#columns).", 1561 self._path, 1562 ) 1563 1564 super().validate_definition()
Validates the model's definition.
Raises:
- ConfigError
def
is_breaking_change( self, previous: Union[SqlModel, SeedModel, PythonModel, ExternalModel]) -> Optional[bool]:
1566 def is_breaking_change(self, previous: Model) -> t.Optional[bool]: 1567 if not isinstance(previous, SqlModel): 1568 return None 1569 1570 if self.lookback != previous.lookback: 1571 return None 1572 1573 try: 1574 # the previous model which comes from disk could be unrenderable 1575 previous_query = previous.render_query() 1576 except Exception: 1577 previous_query = None 1578 this_query = self.render_query() 1579 1580 if previous_query is None or this_query is None: 1581 # Can't determine if there's a breaking change if we can't render the query. 1582 return None 1583 1584 if previous_query is this_query: 1585 edits = [] 1586 else: 1587 edits = diff( 1588 previous_query, 1589 this_query, 1590 matchings=[(previous_query, this_query)], 1591 delta_only=True, 1592 dialect=self.dialect if self.dialect == previous.dialect else None, 1593 ) 1594 inserted_expressions = {e.expression for e in edits if isinstance(e, Insert)} 1595 1596 for edit in edits: 1597 if not isinstance(edit, Insert): 1598 return _additive_projection_change(previous_query, this_query, self.dialect) 1599 1600 expr = edit.expression 1601 if isinstance(expr, exp.UDTF): 1602 # projection subqueries do not change cardinality, engines don't allow these to return 1603 # more than one row of data 1604 parent = expr.find_ancestor(exp.Subquery) 1605 1606 if not parent: 1607 return None 1608 1609 expr = parent 1610 1611 if not _is_projection(expr) and expr.parent not in inserted_expressions: 1612 return _additive_projection_change(previous_query, this_query, self.dialect) 1613 1614 return False
Determines whether this model is a breaking change in relation to the previous model.
Arguments:
- previous: The previous model to compare against.
Returns:
True if this model instance represents a breaking change, False if it's a non-breaking change and None if the nature of the change can't be determined.
def
is_metadata_only_change(self, previous: sqlmesh.core.node._Node) -> bool:
1616 def is_metadata_only_change(self, previous: _Node) -> bool: 1617 if self._is_metadata_only_change_cache.get(id(previous), None) is not None: 1618 return self._is_metadata_only_change_cache[id(previous)] 1619 1620 if not super().is_metadata_only_change(previous): 1621 return False 1622 1623 if not isinstance(previous, SqlModel): 1624 self._is_metadata_only_change_cache[id(previous)] = False 1625 return False 1626 1627 this_rendered_query = self.render_query() or self.query 1628 previous_rendered_query = previous.render_query() or previous.query 1629 is_metadata_change = this_rendered_query == previous_rendered_query 1630 1631 self._is_metadata_only_change_cache[id(previous)] = is_metadata_change 1632 return is_metadata_change
Determines if this node is a metadata only change in relation to the previous node.
Arguments:
- previous: The previous node to compare against.
Returns:
True if this node is a metadata only change, False otherwise.
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': (), 'frozen': True}
Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].
def
model_post_init(self: pydantic.main.BaseModel, context: Any, /) -> None:
365def init_private_attributes(self: BaseModel, context: Any, /) -> None: 366 """This function is meant to behave like a BaseModel method to initialize private attributes. 367 368 It takes context as an argument since that's what pydantic-core passes when calling it. 369 370 Args: 371 self: The BaseModel instance. 372 context: The context. 373 """ 374 if getattr(self, '__pydantic_private__', None) is None: 375 pydantic_private = {} 376 for name, private_attr in self.__private_attributes__.items(): 377 # Avoid needlessly creating a new dict for the validated data: 378 if private_attr.default_factory_takes_validated_data: 379 default = private_attr.get_default( 380 call_default_factory=True, validated_data={**self.__dict__, **pydantic_private} 381 ) 382 else: 383 default = private_attr.get_default(call_default_factory=True) 384 if default is not PydanticUndefined: 385 pydantic_private[name] = default 386 object_setattr(self, '__pydantic_private__', pydantic_private)
This function is meant to behave like a BaseModel method to initialize private attributes.
It takes context as an argument since that's what pydantic-core passes when calling it.
Arguments:
- self: The BaseModel instance.
- context: The context.
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_rebuild
- model_validate
- model_validate_json
- model_validate_strings
- parse_file
- from_orm
- construct
- schema
- schema_json
- validate
- update_forward_refs
- _Model
- python_env
- jinja_macros
- audit_definitions
- mapping_schema
- extract_dependencies_from_query
- pre_statements_
- post_statements_
- on_virtual_update_
- render
- render_query_or_raise
- render_pre_statements
- render_post_statements
- render_on_virtual_update
- render_audit_query
- pre_statements
- post_statements
- on_virtual_update
- macro_definitions
- render_signals
- render_signal_calls
- render_merge_filter
- render_physical_properties
- render_virtual_properties
- render_session_properties
- ctas_query
- text_diff
- set_time_format
- convert_to_time_column
- depends_on
- columns_to_types_or_raise
- annotated
- sorted_python_env
- view_name
- schema_name
- physical_schema
- is_python
- is_seed
- depends_on_self
- forward_only
- disable_restatement
- auto_restatement_intervals
- auto_restatement_cron
- auto_restatement_croniter
- wap_supported
- data_hash
- audit_metadata_hash
- metadata_hash
- is_model
- grants_table_type
- full_depends_on
- partitioned_by
- partition_interval_unit
- audits_with_args
- sqlmesh.core.model.meta.ModelMeta
- dialect
- name
- kind
- retention
- table_format
- storage_format
- partitioned_by_
- clustered_by
- default_catalog
- depends_on_
- columns_to_types_
- column_descriptions_
- audits
- grains
- references
- physical_schema_override
- physical_properties_
- virtual_properties_
- session_properties_
- allow_partials
- signals
- enabled
- physical_version
- gateway
- optimize_query
- ignored_rules_
- formatting
- virtual_environment_mode
- grants_
- grants_target_layer
- ignored_rules_validator
- session_properties_validator
- time_column
- unique_key
- lookback
- lookback_start
- batch_size
- batch_concurrency
- physical_properties
- virtual_properties
- session_properties
- custom_materialization_properties
- grants
- all_references
- on
- managed_columns
- when_matched
- merge_filter
- catalog
- fully_qualified_table
- fqn
- on_destructive_change
- on_additive_change
- ignored_rules
1676class SeedModel(_Model): 1677 """The model definition which uses a pre-built static dataset to source the data from. 1678 1679 Args: 1680 seed: The content of a pre-built static dataset. 1681 """ 1682 1683 kind: SeedKind 1684 seed: Seed 1685 column_hashes_: t.Optional[t.Dict[str, str]] = Field(default=None, alias="column_hashes") 1686 derived_columns_to_types: t.Optional[t.Dict[str, exp.DataType]] = None 1687 is_hydrated: bool = True 1688 source_type: t.Literal["seed"] = "seed" 1689 1690 def __getstate__(self) -> t.Dict[t.Any, t.Any]: 1691 state = super().__getstate__() 1692 state["__dict__"] = state["__dict__"].copy() 1693 state["__dict__"].pop("_reader", None) 1694 return state 1695 1696 def copy(self, **kwargs: t.Any) -> Self: 1697 model = super().copy(**kwargs) 1698 model.__dict__.pop("_reader", None) 1699 return model 1700 1701 def render( 1702 self, 1703 *, 1704 context: ExecutionContext, 1705 start: t.Optional[TimeLike] = None, 1706 end: t.Optional[TimeLike] = None, 1707 execution_time: t.Optional[TimeLike] = None, 1708 **kwargs: t.Any, 1709 ) -> t.Iterator[QueryOrDF]: 1710 if not self.is_hydrated: 1711 return 1712 yield from self.render_seed() 1713 1714 def render_seed(self) -> t.Iterator[QueryOrDF]: 1715 import numpy as np 1716 1717 self._ensure_hydrated() 1718 1719 date_columns = [] 1720 datetime_columns = [] 1721 bool_columns = [] 1722 string_columns = [] 1723 1724 columns_to_types = self.columns_to_types_ or {} 1725 column_names_to_check = set(columns_to_types) 1726 for name, tpe in columns_to_types.items(): 1727 if tpe.this in (exp.DataType.Type.DATE, exp.DataType.Type.DATE32): 1728 date_columns.append(name) 1729 elif tpe.this in exp.DataType.TEMPORAL_TYPES: 1730 datetime_columns.append(name) 1731 elif tpe.is_type("boolean"): 1732 bool_columns.append(name) 1733 elif tpe.this in exp.DataType.TEXT_TYPES: 1734 string_columns.append(name) 1735 1736 for df in self._reader.read(batch_size=self.kind.batch_size): 1737 rename_dict = {} 1738 for column in columns_to_types: 1739 if column not in df: 1740 normalized_name = normalize_identifiers(column, dialect=self.dialect).name 1741 if normalized_name in df: 1742 rename_dict[normalized_name] = column 1743 if rename_dict: 1744 df.rename(columns=rename_dict, inplace=True) 1745 # These names have already been checked 1746 column_names_to_check -= set(rename_dict) 1747 1748 missing_columns = column_names_to_check - set(df.columns) 1749 if missing_columns: 1750 raise_config_error( 1751 f"Seed model '{self.name}' has missing columns: {missing_columns}", self._path 1752 ) 1753 1754 # convert all date/time types to native pandas timestamp 1755 for column in [*date_columns, *datetime_columns]: 1756 import pandas as pd 1757 1758 df[column] = pd.to_datetime(df[column], infer_datetime_format=True, errors="ignore") # type: ignore 1759 1760 # extract datetime.date from pandas timestamp for DATE columns 1761 for column in date_columns: 1762 try: 1763 df[column] = df[column].dt.date 1764 except Exception as ex: 1765 logger.error( 1766 "Failed to convert column '%s' to date in seed model '%s': %s", 1767 column, 1768 self.name, 1769 ex, 1770 ) 1771 1772 for column in bool_columns: 1773 df[column] = df[column].apply(lambda i: str_to_bool(str(i))) 1774 1775 df.loc[:, string_columns] = df[string_columns].mask( 1776 cond=lambda x: x.notna(), # type: ignore 1777 other=df[string_columns].astype(str), # type: ignore 1778 ) 1779 yield df.replace({np.nan: None}) 1780 1781 @property 1782 def columns_to_types(self) -> t.Optional[t.Dict[str, exp.DataType]]: 1783 if self.columns_to_types_ is not None: 1784 return self.columns_to_types_ 1785 if self.derived_columns_to_types is not None: 1786 return self.derived_columns_to_types 1787 if self.is_hydrated: 1788 return self._reader.columns_to_types 1789 return None 1790 1791 @property 1792 def column_hashes(self) -> t.Dict[str, str]: 1793 if self.column_hashes_ is not None: 1794 return self.column_hashes_ 1795 self._ensure_hydrated() 1796 return self._reader.column_hashes 1797 1798 @property 1799 def is_seed(self) -> bool: 1800 return True 1801 1802 @property 1803 def seed_path(self) -> Path: 1804 seed_path = Path(self.kind.path) 1805 if not seed_path.is_absolute(): 1806 if self._path is None: 1807 raise SQLMeshError(f"Seed model '{self.name}' has no path") 1808 return self._path.parent / seed_path 1809 return seed_path 1810 1811 @property 1812 def depends_on(self) -> t.Set[str]: 1813 return (self.depends_on_ or set()) - {self.fqn} 1814 1815 @property 1816 def depends_on_self(self) -> bool: 1817 return False 1818 1819 @property 1820 def batch_size(self) -> t.Optional[int]: 1821 # Unlike other model kinds, the batch size provided in the SEED kind represents the 1822 # maximum number of rows to insert in a single batch. 1823 # We should never batch intervals for seed models. 1824 return None 1825 1826 def to_dehydrated(self) -> SeedModel: 1827 """Creates a dehydrated copy of this model. 1828 1829 The dehydrated seed model will not contain the seed content, but will contain 1830 the column hashes. This is useful for comparing two seed models without 1831 having to read the seed content from disk. 1832 1833 Returns: 1834 A dehydrated copy of this model. 1835 """ 1836 if not self.is_hydrated: 1837 return self 1838 1839 return self.copy( 1840 update={ 1841 "seed": Seed(content=""), 1842 "is_hydrated": False, 1843 "column_hashes_": self.column_hashes, 1844 "derived_columns_to_types": self.columns_to_types 1845 if self.columns_to_types_ is None 1846 else None, 1847 } 1848 ) 1849 1850 def to_hydrated(self, content: str) -> SeedModel: 1851 """Creates a hydrated copy of this model with the given seed content. 1852 1853 Returns: 1854 A hydrated copy of this model. 1855 """ 1856 if self.is_hydrated: 1857 return self 1858 1859 return self.copy( 1860 update={ 1861 "seed": Seed(content=content), 1862 "is_hydrated": True, 1863 "column_hashes_": None, 1864 }, 1865 ) 1866 1867 def is_breaking_change(self, previous: Model) -> t.Optional[bool]: 1868 if not isinstance(previous, SeedModel): 1869 return None 1870 1871 new_columns = set(self.column_hashes) 1872 old_columns = set(previous.column_hashes) 1873 1874 if not new_columns.issuperset(old_columns): 1875 return None 1876 1877 for col in old_columns: 1878 if self.column_hashes[col] != previous.column_hashes[col]: 1879 return None 1880 1881 return False 1882 1883 def _ensure_hydrated(self) -> None: 1884 if not self.is_hydrated: 1885 raise SQLMeshError(f"Seed model '{self.name}' is not hydrated.") 1886 1887 @cached_property 1888 def _reader(self) -> CsvSeedReader: 1889 return self.seed.reader(dialect=self.dialect, settings=self.kind.csv_settings) 1890 1891 @property 1892 def _data_hash_values_no_sql(self) -> t.List[str]: 1893 data = super()._data_hash_values_no_sql 1894 for column_name, column_hash in self.column_hashes.items(): 1895 data.append(column_name) 1896 data.append(column_hash) 1897 1898 # Include grants in data hash for seed models to force recreation on grant changes 1899 # since seed models don't support migration 1900 data.append(json.dumps(self.grants, sort_keys=True) if self.grants else "") 1901 data.append(self.grants_target_layer) 1902 1903 return data
The model definition which uses a pre-built static dataset to source the data from.
Arguments:
- seed: The content of a pre-built static dataset.
def
copy(self, **kwargs: Any) -> typing_extensions.Self:
1696 def copy(self, **kwargs: t.Any) -> Self: 1697 model = super().copy(**kwargs) 1698 model.__dict__.pop("_reader", None) 1699 return model
Returns a copy of the model.
!!! warning "Deprecated"
This method is now deprecated; use model_copy instead.
If you need include or exclude, use:
python {test="skip" lint="skip"}
data = self.model_dump(include=include, exclude=exclude, round_trip=True)
data = {**data, **(update or {})}
copied = self.model_validate(data)
Arguments:
- include: Optional set or mapping specifying which fields to include in the copied model.
- exclude: Optional set or mapping specifying which fields to exclude in the copied model.
- update: Optional dictionary of field-value pairs to override field values in the copied model.
- deep: If True, the values of fields that are Pydantic models will be deep-copied.
Returns:
A copy of the model with included, excluded and updated fields as specified.
def
render( self, *, context: sqlmesh.core.context.ExecutionContext, 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, **kwargs: Any) -> Iterator[<MagicMock id='130969752645552'>]:
1701 def render( 1702 self, 1703 *, 1704 context: ExecutionContext, 1705 start: t.Optional[TimeLike] = None, 1706 end: t.Optional[TimeLike] = None, 1707 execution_time: t.Optional[TimeLike] = None, 1708 **kwargs: t.Any, 1709 ) -> t.Iterator[QueryOrDF]: 1710 if not self.is_hydrated: 1711 return 1712 yield from self.render_seed()
Renders the content of this model in a form of either a SELECT query, executing which the data for this model can be fetched, or a dataframe object which contains the data itself.
The type of the returned object (query or dataframe) depends on whether the model was sourced from a SQL query, a Python script or a pre-built dataset (seed).
Arguments:
- context: The execution context used for fetching data.
- start: The start date/time of the run.
- end: The end date/time of the run.
- execution_time: The date/time time reference to use for execution time.
Returns:
A generator which yields either a query object or one of the supported dataframe objects.
def
render_seed(self) -> Iterator[<MagicMock id='130969752645552'>]:
1714 def render_seed(self) -> t.Iterator[QueryOrDF]: 1715 import numpy as np 1716 1717 self._ensure_hydrated() 1718 1719 date_columns = [] 1720 datetime_columns = [] 1721 bool_columns = [] 1722 string_columns = [] 1723 1724 columns_to_types = self.columns_to_types_ or {} 1725 column_names_to_check = set(columns_to_types) 1726 for name, tpe in columns_to_types.items(): 1727 if tpe.this in (exp.DataType.Type.DATE, exp.DataType.Type.DATE32): 1728 date_columns.append(name) 1729 elif tpe.this in exp.DataType.TEMPORAL_TYPES: 1730 datetime_columns.append(name) 1731 elif tpe.is_type("boolean"): 1732 bool_columns.append(name) 1733 elif tpe.this in exp.DataType.TEXT_TYPES: 1734 string_columns.append(name) 1735 1736 for df in self._reader.read(batch_size=self.kind.batch_size): 1737 rename_dict = {} 1738 for column in columns_to_types: 1739 if column not in df: 1740 normalized_name = normalize_identifiers(column, dialect=self.dialect).name 1741 if normalized_name in df: 1742 rename_dict[normalized_name] = column 1743 if rename_dict: 1744 df.rename(columns=rename_dict, inplace=True) 1745 # These names have already been checked 1746 column_names_to_check -= set(rename_dict) 1747 1748 missing_columns = column_names_to_check - set(df.columns) 1749 if missing_columns: 1750 raise_config_error( 1751 f"Seed model '{self.name}' has missing columns: {missing_columns}", self._path 1752 ) 1753 1754 # convert all date/time types to native pandas timestamp 1755 for column in [*date_columns, *datetime_columns]: 1756 import pandas as pd 1757 1758 df[column] = pd.to_datetime(df[column], infer_datetime_format=True, errors="ignore") # type: ignore 1759 1760 # extract datetime.date from pandas timestamp for DATE columns 1761 for column in date_columns: 1762 try: 1763 df[column] = df[column].dt.date 1764 except Exception as ex: 1765 logger.error( 1766 "Failed to convert column '%s' to date in seed model '%s': %s", 1767 column, 1768 self.name, 1769 ex, 1770 ) 1771 1772 for column in bool_columns: 1773 df[column] = df[column].apply(lambda i: str_to_bool(str(i))) 1774 1775 df.loc[:, string_columns] = df[string_columns].mask( 1776 cond=lambda x: x.notna(), # type: ignore 1777 other=df[string_columns].astype(str), # type: ignore 1778 ) 1779 yield df.replace({np.nan: None})
columns_to_types: Optional[Dict[str, sqlglot.expressions.datatypes.DataType]]
1781 @property 1782 def columns_to_types(self) -> t.Optional[t.Dict[str, exp.DataType]]: 1783 if self.columns_to_types_ is not None: 1784 return self.columns_to_types_ 1785 if self.derived_columns_to_types is not None: 1786 return self.derived_columns_to_types 1787 if self.is_hydrated: 1788 return self._reader.columns_to_types 1789 return None
Returns the mapping of column names to types of this model.
depends_on: Set[str]
1811 @property 1812 def depends_on(self) -> t.Set[str]: 1813 return (self.depends_on_ or set()) - {self.fqn}
All of the upstream dependencies referenced in the model's query, excluding self references.
Returns:
A list of all the upstream table names.
batch_size: Optional[int]
1819 @property 1820 def batch_size(self) -> t.Optional[int]: 1821 # Unlike other model kinds, the batch size provided in the SEED kind represents the 1822 # maximum number of rows to insert in a single batch. 1823 # We should never batch intervals for seed models. 1824 return None
The maximal number of units in a single task for a backfill.
1826 def to_dehydrated(self) -> SeedModel: 1827 """Creates a dehydrated copy of this model. 1828 1829 The dehydrated seed model will not contain the seed content, but will contain 1830 the column hashes. This is useful for comparing two seed models without 1831 having to read the seed content from disk. 1832 1833 Returns: 1834 A dehydrated copy of this model. 1835 """ 1836 if not self.is_hydrated: 1837 return self 1838 1839 return self.copy( 1840 update={ 1841 "seed": Seed(content=""), 1842 "is_hydrated": False, 1843 "column_hashes_": self.column_hashes, 1844 "derived_columns_to_types": self.columns_to_types 1845 if self.columns_to_types_ is None 1846 else None, 1847 } 1848 )
Creates a dehydrated copy of this model.
The dehydrated seed model will not contain the seed content, but will contain the column hashes. This is useful for comparing two seed models without having to read the seed content from disk.
Returns:
A dehydrated copy of this model.
1850 def to_hydrated(self, content: str) -> SeedModel: 1851 """Creates a hydrated copy of this model with the given seed content. 1852 1853 Returns: 1854 A hydrated copy of this model. 1855 """ 1856 if self.is_hydrated: 1857 return self 1858 1859 return self.copy( 1860 update={ 1861 "seed": Seed(content=content), 1862 "is_hydrated": True, 1863 "column_hashes_": None, 1864 }, 1865 )
Creates a hydrated copy of this model with the given seed content.
Returns:
A hydrated copy of this model.
def
is_breaking_change( self, previous: Union[SqlModel, SeedModel, PythonModel, ExternalModel]) -> Optional[bool]:
1867 def is_breaking_change(self, previous: Model) -> t.Optional[bool]: 1868 if not isinstance(previous, SeedModel): 1869 return None 1870 1871 new_columns = set(self.column_hashes) 1872 old_columns = set(previous.column_hashes) 1873 1874 if not new_columns.issuperset(old_columns): 1875 return None 1876 1877 for col in old_columns: 1878 if self.column_hashes[col] != previous.column_hashes[col]: 1879 return None 1880 1881 return False
Determines whether this model is a breaking change in relation to the previous model.
Arguments:
- previous: The previous model to compare against.
Returns:
True if this model instance represents a breaking change, False if it's a non-breaking change and None if the nature of the change can't be determined.
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': (), 'frozen': True}
Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].
def
model_post_init(self: pydantic.main.BaseModel, context: Any, /) -> None:
365def init_private_attributes(self: BaseModel, context: Any, /) -> None: 366 """This function is meant to behave like a BaseModel method to initialize private attributes. 367 368 It takes context as an argument since that's what pydantic-core passes when calling it. 369 370 Args: 371 self: The BaseModel instance. 372 context: The context. 373 """ 374 if getattr(self, '__pydantic_private__', None) is None: 375 pydantic_private = {} 376 for name, private_attr in self.__private_attributes__.items(): 377 # Avoid needlessly creating a new dict for the validated data: 378 if private_attr.default_factory_takes_validated_data: 379 default = private_attr.get_default( 380 call_default_factory=True, validated_data={**self.__dict__, **pydantic_private} 381 ) 382 else: 383 default = private_attr.get_default(call_default_factory=True) 384 if default is not PydanticUndefined: 385 pydantic_private[name] = default 386 object_setattr(self, '__pydantic_private__', pydantic_private)
This function is meant to behave like a BaseModel method to initialize private attributes.
It takes context as an argument since that's what pydantic-core passes when calling it.
Arguments:
- self: The BaseModel instance.
- context: The context.
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_rebuild
- model_validate
- model_validate_json
- model_validate_strings
- parse_file
- from_orm
- construct
- schema
- schema_json
- validate
- update_forward_refs
- _Model
- python_env
- jinja_macros
- audit_definitions
- mapping_schema
- extract_dependencies_from_query
- pre_statements_
- post_statements_
- on_virtual_update_
- render_definition
- render_query
- render_query_or_raise
- render_pre_statements
- render_post_statements
- render_on_virtual_update
- render_audit_query
- pre_statements
- post_statements
- on_virtual_update
- macro_definitions
- render_signals
- render_signal_calls
- render_merge_filter
- render_physical_properties
- render_virtual_properties
- render_session_properties
- ctas_query
- text_diff
- set_time_format
- convert_to_time_column
- set_mapping_schema
- update_schema
- columns_to_types_or_raise
- annotated
- sorted_python_env
- view_name
- schema_name
- physical_schema
- is_sql
- is_python
- forward_only
- disable_restatement
- auto_restatement_intervals
- auto_restatement_cron
- auto_restatement_croniter
- wap_supported
- validate_definition
- is_metadata_only_change
- data_hash
- audit_metadata_hash
- metadata_hash
- is_model
- grants_table_type
- full_depends_on
- partitioned_by
- partition_interval_unit
- audits_with_args
- violated_rules_for_query
- sqlmesh.core.model.meta.ModelMeta
- dialect
- name
- retention
- table_format
- storage_format
- partitioned_by_
- clustered_by
- default_catalog
- depends_on_
- columns_to_types_
- column_descriptions_
- audits
- grains
- references
- physical_schema_override
- physical_properties_
- virtual_properties_
- session_properties_
- allow_partials
- signals
- enabled
- physical_version
- gateway
- optimize_query
- ignored_rules_
- formatting
- virtual_environment_mode
- grants_
- grants_target_layer
- ignored_rules_validator
- session_properties_validator
- time_column
- unique_key
- column_descriptions
- lookback
- lookback_start
- batch_concurrency
- physical_properties
- virtual_properties
- session_properties
- custom_materialization_properties
- grants
- all_references
- on
- managed_columns
- when_matched
- merge_filter
- catalog
- fully_qualified_table
- fqn
- on_destructive_change
- on_additive_change
- ignored_rules
1906class PythonModel(_Model): 1907 """The model definition which relies on a Python script to fetch the data. 1908 1909 Args: 1910 entrypoint: The name of a Python function which contains the data fetching / transformation logic. 1911 """ 1912 1913 kind: ModelKind = FullKind() 1914 entrypoint: str 1915 source_type: t.Literal["python"] = "python" 1916 1917 def validate_definition(self) -> None: 1918 super().validate_definition() 1919 1920 if self.kind and not self.kind.supports_python_models: 1921 raise_config_error( 1922 f"Cannot create Python model '{self.name}' as the '{self.kind.name}' kind doesn't support Python models", 1923 self._path, 1924 ) 1925 1926 def render( 1927 self, 1928 *, 1929 context: ExecutionContext, 1930 start: t.Optional[TimeLike] = None, 1931 end: t.Optional[TimeLike] = None, 1932 execution_time: t.Optional[TimeLike] = None, 1933 **kwargs: t.Any, 1934 ) -> t.Iterator[QueryOrDF]: 1935 env = prepare_env(self.python_env) 1936 start, end = make_inclusive(start or c.EPOCH, end or c.EPOCH, self.dialect) 1937 execution_time = to_datetime(execution_time or c.EPOCH) 1938 1939 variables = { 1940 **env.get(c.SQLMESH_VARS, {}), 1941 **env.get(c.SQLMESH_VARS_METADATA, {}), 1942 **kwargs.pop("variables", {}), 1943 } 1944 blueprint_variables = { 1945 k: d.parse_one(v.sql, dialect=self.dialect) if isinstance(v, SqlValue) else v 1946 for k, v in { 1947 **env.get(c.SQLMESH_BLUEPRINT_VARS, {}), 1948 **env.get(c.SQLMESH_BLUEPRINT_VARS_METADATA, {}), 1949 }.items() 1950 } 1951 try: 1952 kwargs = { 1953 **variables, 1954 **kwargs, 1955 "start": start, 1956 "end": end, 1957 "execution_time": execution_time, 1958 "latest": execution_time, # TODO: Preserved for backward compatibility. Remove in 1.0.0. 1959 } 1960 df_or_iter = env[self.entrypoint]( 1961 context=context.with_variables(variables, blueprint_variables=blueprint_variables), 1962 **kwargs, 1963 ) 1964 1965 if not isinstance(df_or_iter, types.GeneratorType): 1966 df_or_iter = [df_or_iter] 1967 1968 for df in df_or_iter: 1969 yield df 1970 except Exception as e: 1971 raise PythonModelEvalError(format_evaluated_code_exception(e, self.python_env)) 1972 1973 def render_definition( 1974 self, 1975 include_python: bool = True, 1976 include_defaults: bool = False, 1977 render_query: bool = False, 1978 ) -> t.List[exp.Expr]: 1979 # Ignore the provided value for the include_python flag, since the Pyhon model's 1980 # definition without Python code is meaningless. 1981 return super().render_definition( 1982 include_python=True, include_defaults=include_defaults, render_query=render_query 1983 ) 1984 1985 @property 1986 def is_python(self) -> bool: 1987 return True 1988 1989 def is_breaking_change(self, previous: Model) -> t.Optional[bool]: 1990 return None 1991 1992 @property 1993 def _data_hash_values_no_sql(self) -> t.List[str]: 1994 data = super()._data_hash_values_no_sql 1995 data.append(self.entrypoint) 1996 return data
The model definition which relies on a Python script to fetch the data.
Arguments:
- entrypoint: The name of a Python function which contains the data fetching / transformation logic.
kind: Annotated[Union[sqlmesh.core.model.kind.EmbeddedKind, sqlmesh.core.model.kind.ExternalKind, sqlmesh.core.model.kind.FullKind, sqlmesh.core.model.kind.IncrementalByTimeRangeKind, sqlmesh.core.model.kind.IncrementalByUniqueKeyKind, sqlmesh.core.model.kind.IncrementalByPartitionKind, sqlmesh.core.model.kind.IncrementalUnmanagedKind, sqlmesh.core.model.kind.SeedKind, sqlmesh.core.model.kind.ViewKind, sqlmesh.core.model.kind.SCDType2ByTimeKind, sqlmesh.core.model.kind.SCDType2ByColumnKind, sqlmesh.core.model.kind.CustomKind, sqlmesh.core.model.kind.ManagedKind, sqlmesh.core.model.kind.DbtCustomKind], FieldInfo(annotation=NoneType, required=True, discriminator='name')]
def
validate_definition(self) -> None:
1917 def validate_definition(self) -> None: 1918 super().validate_definition() 1919 1920 if self.kind and not self.kind.supports_python_models: 1921 raise_config_error( 1922 f"Cannot create Python model '{self.name}' as the '{self.kind.name}' kind doesn't support Python models", 1923 self._path, 1924 )
Validates the model's definition.
Raises:
- ConfigError
def
render( self, *, context: sqlmesh.core.context.ExecutionContext, 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, **kwargs: Any) -> Iterator[<MagicMock id='130969752645552'>]:
1926 def render( 1927 self, 1928 *, 1929 context: ExecutionContext, 1930 start: t.Optional[TimeLike] = None, 1931 end: t.Optional[TimeLike] = None, 1932 execution_time: t.Optional[TimeLike] = None, 1933 **kwargs: t.Any, 1934 ) -> t.Iterator[QueryOrDF]: 1935 env = prepare_env(self.python_env) 1936 start, end = make_inclusive(start or c.EPOCH, end or c.EPOCH, self.dialect) 1937 execution_time = to_datetime(execution_time or c.EPOCH) 1938 1939 variables = { 1940 **env.get(c.SQLMESH_VARS, {}), 1941 **env.get(c.SQLMESH_VARS_METADATA, {}), 1942 **kwargs.pop("variables", {}), 1943 } 1944 blueprint_variables = { 1945 k: d.parse_one(v.sql, dialect=self.dialect) if isinstance(v, SqlValue) else v 1946 for k, v in { 1947 **env.get(c.SQLMESH_BLUEPRINT_VARS, {}), 1948 **env.get(c.SQLMESH_BLUEPRINT_VARS_METADATA, {}), 1949 }.items() 1950 } 1951 try: 1952 kwargs = { 1953 **variables, 1954 **kwargs, 1955 "start": start, 1956 "end": end, 1957 "execution_time": execution_time, 1958 "latest": execution_time, # TODO: Preserved for backward compatibility. Remove in 1.0.0. 1959 } 1960 df_or_iter = env[self.entrypoint]( 1961 context=context.with_variables(variables, blueprint_variables=blueprint_variables), 1962 **kwargs, 1963 ) 1964 1965 if not isinstance(df_or_iter, types.GeneratorType): 1966 df_or_iter = [df_or_iter] 1967 1968 for df in df_or_iter: 1969 yield df 1970 except Exception as e: 1971 raise PythonModelEvalError(format_evaluated_code_exception(e, self.python_env))
Renders the content of this model in a form of either a SELECT query, executing which the data for this model can be fetched, or a dataframe object which contains the data itself.
The type of the returned object (query or dataframe) depends on whether the model was sourced from a SQL query, a Python script or a pre-built dataset (seed).
Arguments:
- context: The execution context used for fetching data.
- start: The start date/time of the run.
- end: The end date/time of the run.
- execution_time: The date/time time reference to use for execution time.
Returns:
A generator which yields either a query object or one of the supported dataframe objects.
def
render_definition( self, include_python: bool = True, include_defaults: bool = False, render_query: bool = False) -> List[sqlglot.expressions.core.Expr]:
1973 def render_definition( 1974 self, 1975 include_python: bool = True, 1976 include_defaults: bool = False, 1977 render_query: bool = False, 1978 ) -> t.List[exp.Expr]: 1979 # Ignore the provided value for the include_python flag, since the Pyhon model's 1980 # definition without Python code is meaningless. 1981 return super().render_definition( 1982 include_python=True, include_defaults=include_defaults, render_query=render_query 1983 )
Returns the original list of sql expressions comprising the model definition.
Arguments:
- include_python: Whether or not to include Python code in the rendered definition.
def
is_breaking_change( self, previous: Union[SqlModel, SeedModel, PythonModel, ExternalModel]) -> Optional[bool]:
Determines whether this model is a breaking change in relation to the previous model.
Arguments:
- previous: The previous model to compare against.
Returns:
True if this model instance represents a breaking change, False if it's a non-breaking change and None if the nature of the change can't be determined.
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': (), 'frozen': True}
Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].
def
model_post_init(self: pydantic.main.BaseModel, context: Any, /) -> None:
365def init_private_attributes(self: BaseModel, context: Any, /) -> None: 366 """This function is meant to behave like a BaseModel method to initialize private attributes. 367 368 It takes context as an argument since that's what pydantic-core passes when calling it. 369 370 Args: 371 self: The BaseModel instance. 372 context: The context. 373 """ 374 if getattr(self, '__pydantic_private__', None) is None: 375 pydantic_private = {} 376 for name, private_attr in self.__private_attributes__.items(): 377 # Avoid needlessly creating a new dict for the validated data: 378 if private_attr.default_factory_takes_validated_data: 379 default = private_attr.get_default( 380 call_default_factory=True, validated_data={**self.__dict__, **pydantic_private} 381 ) 382 else: 383 default = private_attr.get_default(call_default_factory=True) 384 if default is not PydanticUndefined: 385 pydantic_private[name] = default 386 object_setattr(self, '__pydantic_private__', pydantic_private)
This function is meant to behave like a BaseModel method to initialize private attributes.
It takes context as an argument since that's what pydantic-core passes when calling it.
Arguments:
- self: The BaseModel instance.
- context: The context.
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_rebuild
- model_validate
- model_validate_json
- model_validate_strings
- parse_file
- from_orm
- construct
- schema
- schema_json
- validate
- update_forward_refs
- _Model
- python_env
- jinja_macros
- audit_definitions
- mapping_schema
- extract_dependencies_from_query
- pre_statements_
- post_statements_
- on_virtual_update_
- copy
- render_query
- render_query_or_raise
- render_pre_statements
- render_post_statements
- render_on_virtual_update
- render_audit_query
- pre_statements
- post_statements
- on_virtual_update
- macro_definitions
- render_signals
- render_signal_calls
- render_merge_filter
- render_physical_properties
- render_virtual_properties
- render_session_properties
- ctas_query
- text_diff
- set_time_format
- convert_to_time_column
- set_mapping_schema
- update_schema
- depends_on
- columns_to_types
- columns_to_types_or_raise
- annotated
- sorted_python_env
- view_name
- schema_name
- physical_schema
- is_sql
- is_seed
- depends_on_self
- forward_only
- disable_restatement
- auto_restatement_intervals
- auto_restatement_cron
- auto_restatement_croniter
- wap_supported
- is_metadata_only_change
- data_hash
- audit_metadata_hash
- metadata_hash
- is_model
- grants_table_type
- full_depends_on
- partitioned_by
- partition_interval_unit
- audits_with_args
- violated_rules_for_query
- sqlmesh.core.model.meta.ModelMeta
- dialect
- name
- retention
- table_format
- storage_format
- partitioned_by_
- clustered_by
- default_catalog
- depends_on_
- columns_to_types_
- column_descriptions_
- audits
- grains
- references
- physical_schema_override
- physical_properties_
- virtual_properties_
- session_properties_
- allow_partials
- signals
- enabled
- physical_version
- gateway
- optimize_query
- ignored_rules_
- formatting
- virtual_environment_mode
- grants_
- grants_target_layer
- ignored_rules_validator
- session_properties_validator
- time_column
- unique_key
- column_descriptions
- lookback
- lookback_start
- batch_size
- batch_concurrency
- physical_properties
- virtual_properties
- session_properties
- custom_materialization_properties
- grants
- all_references
- on
- managed_columns
- when_matched
- merge_filter
- catalog
- fully_qualified_table
- fqn
- on_destructive_change
- on_additive_change
- ignored_rules
1999class ExternalModel(_Model): 2000 """The model definition which represents an external source/table.""" 2001 2002 kind: ModelKind = ExternalKind() 2003 source_type: t.Literal["external"] = "external" 2004 2005 def is_breaking_change(self, previous: Model) -> t.Optional[bool]: 2006 if not isinstance(previous, ExternalModel): 2007 return None 2008 if not previous.columns_to_types_or_raise.items() - self.columns_to_types_or_raise.items(): 2009 return False 2010 return None 2011 2012 @property 2013 def depends_on(self) -> t.Set[str]: 2014 return set() 2015 2016 @property 2017 def depends_on_self(self) -> bool: 2018 return False
The model definition which represents an external source/table.
kind: Annotated[Union[sqlmesh.core.model.kind.EmbeddedKind, sqlmesh.core.model.kind.ExternalKind, sqlmesh.core.model.kind.FullKind, sqlmesh.core.model.kind.IncrementalByTimeRangeKind, sqlmesh.core.model.kind.IncrementalByUniqueKeyKind, sqlmesh.core.model.kind.IncrementalByPartitionKind, sqlmesh.core.model.kind.IncrementalUnmanagedKind, sqlmesh.core.model.kind.SeedKind, sqlmesh.core.model.kind.ViewKind, sqlmesh.core.model.kind.SCDType2ByTimeKind, sqlmesh.core.model.kind.SCDType2ByColumnKind, sqlmesh.core.model.kind.CustomKind, sqlmesh.core.model.kind.ManagedKind, sqlmesh.core.model.kind.DbtCustomKind], FieldInfo(annotation=NoneType, required=True, discriminator='name')]
def
is_breaking_change( self, previous: Union[SqlModel, SeedModel, PythonModel, ExternalModel]) -> Optional[bool]:
2005 def is_breaking_change(self, previous: Model) -> t.Optional[bool]: 2006 if not isinstance(previous, ExternalModel): 2007 return None 2008 if not previous.columns_to_types_or_raise.items() - self.columns_to_types_or_raise.items(): 2009 return False 2010 return None
Determines whether this model is a breaking change in relation to the previous model.
Arguments:
- previous: The previous model to compare against.
Returns:
True if this model instance represents a breaking change, False if it's a non-breaking change and None if the nature of the change can't be determined.
depends_on: Set[str]
All of the upstream dependencies referenced in the model's query, excluding self references.
Returns:
A list of all the upstream table names.
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': (), 'frozen': True}
Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].
def
model_post_init(self: pydantic.main.BaseModel, context: Any, /) -> None:
365def init_private_attributes(self: BaseModel, context: Any, /) -> None: 366 """This function is meant to behave like a BaseModel method to initialize private attributes. 367 368 It takes context as an argument since that's what pydantic-core passes when calling it. 369 370 Args: 371 self: The BaseModel instance. 372 context: The context. 373 """ 374 if getattr(self, '__pydantic_private__', None) is None: 375 pydantic_private = {} 376 for name, private_attr in self.__private_attributes__.items(): 377 # Avoid needlessly creating a new dict for the validated data: 378 if private_attr.default_factory_takes_validated_data: 379 default = private_attr.get_default( 380 call_default_factory=True, validated_data={**self.__dict__, **pydantic_private} 381 ) 382 else: 383 default = private_attr.get_default(call_default_factory=True) 384 if default is not PydanticUndefined: 385 pydantic_private[name] = default 386 object_setattr(self, '__pydantic_private__', pydantic_private)
This function is meant to behave like a BaseModel method to initialize private attributes.
It takes context as an argument since that's what pydantic-core passes when calling it.
Arguments:
- self: The BaseModel instance.
- context: The context.
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_rebuild
- model_validate
- model_validate_json
- model_validate_strings
- parse_file
- from_orm
- construct
- schema
- schema_json
- validate
- update_forward_refs
- _Model
- python_env
- jinja_macros
- audit_definitions
- mapping_schema
- extract_dependencies_from_query
- pre_statements_
- post_statements_
- on_virtual_update_
- copy
- render
- render_definition
- render_query
- render_query_or_raise
- render_pre_statements
- render_post_statements
- render_on_virtual_update
- render_audit_query
- pre_statements
- post_statements
- on_virtual_update
- macro_definitions
- render_signals
- render_signal_calls
- render_merge_filter
- render_physical_properties
- render_virtual_properties
- render_session_properties
- ctas_query
- text_diff
- set_time_format
- convert_to_time_column
- set_mapping_schema
- update_schema
- columns_to_types
- columns_to_types_or_raise
- annotated
- sorted_python_env
- view_name
- schema_name
- physical_schema
- is_sql
- is_python
- is_seed
- forward_only
- disable_restatement
- auto_restatement_intervals
- auto_restatement_cron
- auto_restatement_croniter
- wap_supported
- validate_definition
- is_metadata_only_change
- data_hash
- audit_metadata_hash
- metadata_hash
- is_model
- grants_table_type
- full_depends_on
- partitioned_by
- partition_interval_unit
- audits_with_args
- violated_rules_for_query
- sqlmesh.core.model.meta.ModelMeta
- dialect
- name
- retention
- table_format
- storage_format
- partitioned_by_
- clustered_by
- default_catalog
- depends_on_
- columns_to_types_
- column_descriptions_
- audits
- grains
- references
- physical_schema_override
- physical_properties_
- virtual_properties_
- session_properties_
- allow_partials
- signals
- enabled
- physical_version
- gateway
- optimize_query
- ignored_rules_
- formatting
- virtual_environment_mode
- grants_
- grants_target_layer
- ignored_rules_validator
- session_properties_validator
- time_column
- unique_key
- column_descriptions
- lookback
- lookback_start
- batch_size
- batch_concurrency
- physical_properties
- virtual_properties
- session_properties
- custom_materialization_properties
- grants
- all_references
- on
- managed_columns
- when_matched
- merge_filter
- catalog
- fully_qualified_table
- fqn
- on_destructive_change
- on_additive_change
- ignored_rules
Model =
typing.Union[SqlModel, SeedModel, PythonModel, ExternalModel]
2024class AuditResult(PydanticModel): 2025 audit: Audit 2026 """The audit this result is for.""" 2027 audit_args: t.Dict[t.Any, t.Any] 2028 """Arguments passed to the audit.""" 2029 model: t.Optional[_Model] = None 2030 """The model this audit is for.""" 2031 count: t.Optional[int] = None 2032 """The number of records returned by the audit query. This could be None if the audit was skipped.""" 2033 query: t.Optional[exp.Expr] = None 2034 """The rendered query used by the audit. This could be None if the audit was skipped.""" 2035 skipped: bool = False 2036 """Whether or not the audit was blocking. This can be overriden by the user.""" 2037 blocking: bool = True
!!! 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__andModel.__root_validators__from Pydantic V1. - __pydantic_generic_metadata__: A dictionary containing metadata about generic Pydantic models.
The
originandargsitems map to the [__origin__][genericalias.__origin__] and [__args__][genericalias.__args__] attributes of [generic aliases][types-genericalias], and theparameteritem 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-coreSchemaSerializerused to dump instances of the model. - __pydantic_validator__: The
pydantic-coreSchemaValidatorused 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.
audit: Union[sqlmesh.core.audit.definition.ModelAudit, sqlmesh.core.audit.definition.StandaloneAudit]
The audit this result is for.
count: Optional[int]
The number of records returned by the audit query. This could be None if the audit was skipped.
query: Optional[sqlglot.expressions.core.Expr]
The rendered query used by the audit. This could be None if the audit was skipped.
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
2040class EvaluatableSignals(PydanticModel): 2041 signals_to_kwargs: t.Dict[str, t.Dict[str, t.Optional[exp.Expr]]] 2042 """A mapping of signal names to the kwargs passed to the signal.""" 2043 python_env: t.Dict[str, Executable] 2044 """The Python environment that should be used to evaluated the rendered signal calls.""" 2045 prepared_python_env: t.Dict[str, t.Any] 2046 """The prepared Python environment that should be used to evaluated the rendered signal calls."""
!!! 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__andModel.__root_validators__from Pydantic V1. - __pydantic_generic_metadata__: A dictionary containing metadata about generic Pydantic models.
The
originandargsitems map to the [__origin__][genericalias.__origin__] and [__args__][genericalias.__args__] attributes of [generic aliases][types-genericalias], and theparameteritem 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-coreSchemaSerializerused to dump instances of the model. - __pydantic_validator__: The
pydantic-coreSchemaValidatorused 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.
signals_to_kwargs: Dict[str, Dict[str, Optional[sqlglot.expressions.core.Expr]]]
A mapping of signal names to the kwargs passed to the signal.
python_env: Dict[str, sqlmesh.utils.metaprogramming.Executable]
The Python environment that should be used to evaluated the rendered signal calls.
prepared_python_env: Dict[str, Any]
The prepared Python environment that should be used to evaluated the rendered signal calls.
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
def
create_models_from_blueprints( gateway: Union[str, sqlglot.expressions.core.Expr, NoneType], blueprints: Any, get_variables: Callable[[Optional[str]], Dict[str, str]], loader: Callable[..., Union[SqlModel, SeedModel, PythonModel, ExternalModel]], path: pathlib.Path = PosixPath('.'), module_path: pathlib.Path = PosixPath('.'), dialect: Union[str, sqlglot.dialects.dialect.Dialect, type[sqlglot.dialects.dialect.Dialect], NoneType] = None, default_catalog_per_gateway: Optional[Dict[str, str]] = None, **loader_kwargs: Any) -> List[Union[SqlModel, SeedModel, PythonModel, ExternalModel]]:
2085def create_models_from_blueprints( 2086 gateway: t.Optional[str | exp.Expr], 2087 blueprints: t.Any, 2088 get_variables: t.Callable[[t.Optional[str]], t.Dict[str, str]], 2089 loader: t.Callable[..., Model], 2090 path: Path = Path(), 2091 module_path: Path = Path(), 2092 dialect: DialectType = None, 2093 default_catalog_per_gateway: t.Optional[t.Dict[str, str]] = None, 2094 **loader_kwargs: t.Any, 2095) -> t.List[Model]: 2096 model_blueprints: t.List[Model] = [] 2097 original_default_catalog = loader_kwargs.get("default_catalog") 2098 for blueprint in _extract_blueprints(blueprints, path): 2099 loader_kwargs["default_catalog"] = original_default_catalog 2100 blueprint_variables = _extract_blueprint_variables(blueprint, path) 2101 2102 if gateway: 2103 rendered_gateway = render_expression( 2104 expression=exp.maybe_parse(gateway, dialect=dialect), 2105 module_path=module_path, 2106 macros=loader_kwargs.get("macros"), 2107 jinja_macros=loader_kwargs.get("jinja_macros"), 2108 path=path, 2109 dialect=dialect, 2110 default_catalog=loader_kwargs.get("default_catalog"), 2111 blueprint_variables=blueprint_variables, 2112 ) 2113 gateway_name = rendered_gateway[0].name if rendered_gateway else None 2114 else: 2115 gateway_name = None 2116 2117 if default_catalog_per_gateway and gateway_name: 2118 catalog = default_catalog_per_gateway.get(gateway_name) 2119 if catalog is not None: 2120 loader_kwargs["default_catalog"] = catalog 2121 else: 2122 # Gateway exists but has no entry in the dict (e.g., catalog-unsupported 2123 # engines like ClickHouse). Clear the default catalog so the global 2124 # default from the primary gateway doesn't leak into this model's name. 2125 loader_kwargs["default_catalog"] = None 2126 2127 model_blueprints.append( 2128 loader( 2129 path=path, 2130 module_path=module_path, 2131 dialect=dialect, 2132 variables=get_variables(gateway_name), 2133 blueprint_variables=blueprint_variables, 2134 **loader_kwargs, 2135 ) 2136 ) 2137 2138 return model_blueprints
def
load_sql_based_models( expressions: List[sqlglot.expressions.core.Expr], get_variables: Callable[[Optional[str]], Dict[str, str]], path: pathlib.Path = PosixPath('.'), module_path: pathlib.Path = PosixPath('.'), dialect: Union[str, sqlglot.dialects.dialect.Dialect, type[sqlglot.dialects.dialect.Dialect], NoneType] = None, default_catalog_per_gateway: Optional[Dict[str, str]] = None, **loader_kwargs: Any) -> List[Union[SqlModel, SeedModel, PythonModel, ExternalModel]]:
2141def load_sql_based_models( 2142 expressions: t.List[exp.Expr], 2143 get_variables: t.Callable[[t.Optional[str]], t.Dict[str, str]], 2144 path: Path = Path(), 2145 module_path: Path = Path(), 2146 dialect: DialectType = None, 2147 default_catalog_per_gateway: t.Optional[t.Dict[str, str]] = None, 2148 **loader_kwargs: t.Any, 2149) -> t.List[Model]: 2150 gateway: t.Optional[exp.Expr] = None 2151 blueprints: t.Optional[exp.Expr] = None 2152 2153 model_meta = seq_get(expressions, 0) 2154 for prop in (isinstance(model_meta, d.Model) and model_meta.expressions) or []: 2155 if prop.name == "gateway": 2156 gateway = prop.args["value"] 2157 elif prop.name == "blueprints": 2158 # We pop the `blueprints` here to avoid walking large lists when rendering the meta 2159 blueprints = prop.pop().args["value"] 2160 2161 if isinstance(blueprints, d.MacroFunc): 2162 rendered_blueprints = render_expression( 2163 expression=blueprints, 2164 module_path=module_path, 2165 macros=loader_kwargs.get("macros"), 2166 jinja_macros=loader_kwargs.get("jinja_macros"), 2167 variables=get_variables(None), 2168 path=path, 2169 dialect=dialect, 2170 default_catalog=loader_kwargs.get("default_catalog"), 2171 ) 2172 if not rendered_blueprints: 2173 raise_config_error("Failed to render blueprints property", path) 2174 2175 # Help mypy see that rendered_blueprints can't be None 2176 assert rendered_blueprints 2177 2178 if len(rendered_blueprints) > 1: 2179 rendered_blueprints = [exp.Tuple(expressions=rendered_blueprints)] 2180 2181 blueprints = rendered_blueprints[0] 2182 2183 return create_models_from_blueprints( 2184 gateway=gateway, 2185 blueprints=blueprints, 2186 get_variables=get_variables, 2187 loader=partial(load_sql_based_model, expressions), 2188 path=path, 2189 module_path=module_path, 2190 dialect=dialect, 2191 default_catalog_per_gateway=default_catalog_per_gateway, 2192 **loader_kwargs, 2193 )
def
load_sql_based_model( expressions: List[sqlglot.expressions.core.Expr], *, defaults: Optional[Dict[str, Any]] = None, path: Optional[pathlib.Path] = None, module_path: pathlib.Path = PosixPath('.'), time_column_format: str = '%Y-%m-%d', macros: Optional[sqlmesh.utils.UniqueKeyDict[str, Union[sqlmesh.utils.metaprogramming.Executable, sqlmesh.core.macros.macro]]] = None, jinja_macros: Optional[sqlmesh.utils.jinja.JinjaMacroRegistry] = None, audits: Optional[Dict[str, sqlmesh.core.audit.definition.ModelAudit]] = None, python_env: Optional[Dict[str, sqlmesh.utils.metaprogramming.Executable]] = None, dialect: Optional[str] = None, physical_schema_mapping: Optional[Dict[re.Pattern, str]] = None, default_catalog: Optional[str] = None, variables: Optional[Dict[str, Any]] = None, infer_names: Optional[bool] = False, blueprint_variables: Optional[Dict[str, Any]] = None, **kwargs: Any) -> Union[SqlModel, SeedModel, PythonModel, ExternalModel]:
2196def load_sql_based_model( 2197 expressions: t.List[exp.Expr], 2198 *, 2199 defaults: t.Optional[t.Dict[str, t.Any]] = None, 2200 path: t.Optional[Path] = None, 2201 module_path: Path = Path(), 2202 time_column_format: str = c.DEFAULT_TIME_COLUMN_FORMAT, 2203 macros: t.Optional[MacroRegistry] = None, 2204 jinja_macros: t.Optional[JinjaMacroRegistry] = None, 2205 audits: t.Optional[t.Dict[str, ModelAudit]] = None, 2206 python_env: t.Optional[t.Dict[str, Executable]] = None, 2207 dialect: t.Optional[str] = None, 2208 physical_schema_mapping: t.Optional[t.Dict[re.Pattern, str]] = None, 2209 default_catalog: t.Optional[str] = None, 2210 variables: t.Optional[t.Dict[str, t.Any]] = None, 2211 infer_names: t.Optional[bool] = False, 2212 blueprint_variables: t.Optional[t.Dict[str, t.Any]] = None, 2213 **kwargs: t.Any, 2214) -> Model: 2215 """Load a model from a parsed SQLMesh model SQL file. 2216 2217 Args: 2218 expressions: Model, *Statements, Query. 2219 defaults: Definition default values. 2220 path: An optional path to the model definition file. 2221 module_path: The python module path to serialize macros for. 2222 time_column_format: The default time column format to use if no model time column is configured. 2223 macros: The custom registry of macros. If not provided the default registry will be used. 2224 jinja_macros: The registry of Jinja macros. 2225 python_env: The custom Python environment for macros. If not provided the environment will be constructed 2226 from the macro registry. 2227 dialect: The default dialect if no model dialect is configured. 2228 The format must adhere to Python's strftime codes. 2229 physical_schema_mapping: A mapping of regular expressions to match against the model schema to produce the corresponding physical schema 2230 default_catalog: The default catalog if no model catalog is configured. 2231 variables: The variables to pass to the model. 2232 kwargs: Additional kwargs to pass to the loader. 2233 """ 2234 missing_model_msg = f"""Please add a MODEL block at the top of the file. Example: 2235 2236MODEL ( 2237 name sqlmesh_example.full_model, --model name 2238 kind FULL, --materialization 2239 cron '@daily', --schedule 2240); 2241 2242Learn more at https://sqlmesh.readthedocs.io/en/stable/concepts/models/overview 2243""" 2244 2245 if not expressions: 2246 raise_config_error(missing_model_msg) 2247 2248 dialect = dialect or "" 2249 meta = expressions[0] 2250 if not isinstance(meta, d.Model): 2251 if not infer_names: 2252 raise_config_error(missing_model_msg) 2253 meta = d.Model(expressions=[]) # Dummy meta node 2254 expressions.insert(0, meta) 2255 2256 # We deliberately hold off rendering some properties at load time because there is not enough information available 2257 # at load time to render them. They will get rendered later at evaluation time 2258 unrendered_properties = {} 2259 unrendered_merge_filter = None 2260 2261 for prop in meta.expressions: 2262 # Macro functions that programmaticaly generate the key-value pair properties should be rendered 2263 # This is needed in the odd case where a macro shares the name of one of the properties 2264 # eg `@session_properties()` Test: `test_macros_in_model_statement` Reference PR: #2574 2265 if isinstance(prop, d.MacroFunc): 2266 continue 2267 2268 prop_name = prop.name.lower() 2269 if prop_name in {"signals", "audits"} | PROPERTIES: 2270 unrendered_properties[prop_name] = prop.args.get("value") 2271 elif ( 2272 prop.name.lower() == "kind" 2273 and (value := prop.args.get("value")) 2274 and value.name.lower() == "incremental_by_unique_key" 2275 ): 2276 for kind_prop in value.expressions: 2277 if kind_prop.name.lower() == "merge_filter": 2278 unrendered_merge_filter = kind_prop 2279 2280 rendered_meta_exprs = render_expression( 2281 expression=meta, 2282 module_path=module_path, 2283 macros=macros, 2284 jinja_macros=jinja_macros, 2285 variables=variables, 2286 path=path, 2287 dialect=dialect, 2288 default_catalog=default_catalog, 2289 blueprint_variables=blueprint_variables, 2290 ) 2291 2292 if rendered_meta_exprs is None or len(rendered_meta_exprs) != 1: 2293 raise_config_error( 2294 f"Invalid MODEL statement:\n{meta.sql(dialect=dialect, pretty=True)}", 2295 path, 2296 ) 2297 raise 2298 2299 rendered_meta = rendered_meta_exprs[0] 2300 2301 rendered_defaults = ( 2302 render_model_defaults( 2303 defaults=defaults, 2304 module_path=module_path, 2305 macros=macros, 2306 jinja_macros=jinja_macros, 2307 variables=variables, 2308 path=path, 2309 dialect=dialect, 2310 default_catalog=default_catalog, 2311 ) 2312 if defaults 2313 else {} 2314 ) 2315 2316 rendered_defaults = parse_defaults_properties(rendered_defaults, dialect=dialect) 2317 2318 # Extract the query and any pre/post statements 2319 query_or_seed_insert, pre_statements, post_statements, on_virtual_update, inline_audits = ( 2320 _split_sql_model_statements(expressions[1:], path, dialect=dialect) 2321 ) 2322 2323 meta_fields: t.Dict[str, t.Any] = { 2324 "dialect": dialect, 2325 "description": ( 2326 "\n".join(comment.strip() for comment in rendered_meta.comments) 2327 if rendered_meta.comments 2328 else None 2329 ), 2330 **{prop.name.lower(): prop.args.get("value") for prop in rendered_meta.expressions}, 2331 **kwargs, 2332 } 2333 2334 # Discard the potentially half-rendered versions of these properties and replace them with the 2335 # original unrendered versions. They will get rendered properly at evaluation time 2336 meta_fields.update(unrendered_properties) 2337 2338 if unrendered_merge_filter: 2339 for idx, kind_prop in enumerate(meta_fields["kind"].expressions): 2340 if kind_prop.name.lower() == "merge_filter": 2341 meta_fields["kind"].expressions[idx] = unrendered_merge_filter 2342 2343 if isinstance(meta_fields.get("dialect"), exp.Expr): 2344 meta_fields["dialect"] = meta_fields["dialect"].name 2345 2346 # The name of the model will be inferred from its path relative to `models/`, if it's not explicitly specified 2347 name = meta_fields.pop("name", "") 2348 if not name and infer_names: 2349 if path is None: 2350 raise ValueError(f"Model {name} must have a name") 2351 name = get_model_name(path) 2352 2353 if not name: 2354 raise_config_error( 2355 "Please add the required 'name' field to the MODEL block at the top of the file.\n\n" 2356 + "Learn more at https://sqlmesh.readthedocs.io/en/stable/concepts/models/overview" 2357 ) 2358 if "default_catalog" in meta_fields: 2359 raise_config_error( 2360 "`default_catalog` cannot be set on a per-model basis. It must be set at the connection level.", 2361 path, 2362 ) 2363 2364 common_kwargs = dict( 2365 pre_statements=pre_statements, 2366 post_statements=post_statements, 2367 on_virtual_update=on_virtual_update, 2368 defaults=rendered_defaults, 2369 path=path, 2370 module_path=module_path, 2371 macros=macros, 2372 python_env=python_env, 2373 jinja_macros=jinja_macros, 2374 physical_schema_mapping=physical_schema_mapping, 2375 default_catalog=default_catalog, 2376 variables=variables, 2377 inline_audits=inline_audits, 2378 blueprint_variables=blueprint_variables, 2379 use_original_sql=True, 2380 **meta_fields, 2381 ) 2382 2383 kind = common_kwargs.pop("kind", ModelMeta.all_field_infos()["kind"].default) 2384 2385 if kind.name != ModelKindName.SEED: 2386 return create_sql_model( 2387 name, 2388 query_or_seed_insert, 2389 kind=kind, 2390 time_column_format=time_column_format, 2391 **common_kwargs, 2392 ) 2393 2394 seed_properties = {p.name.lower(): p.args.get("value") for p in kind.expressions} 2395 return create_seed_model( 2396 name, 2397 SeedKind(**seed_properties), 2398 **common_kwargs, 2399 )
Load a model from a parsed SQLMesh model SQL file.
Arguments:
- expressions: Model, *Statements, Query.
- defaults: Definition default values.
- path: An optional path to the model definition file.
- module_path: The python module path to serialize macros for.
- time_column_format: The default time column format to use if no model time column is configured.
- macros: The custom registry of macros. If not provided the default registry will be used.
- jinja_macros: The registry of Jinja macros.
- python_env: The custom Python environment for macros. If not provided the environment will be constructed from the macro registry.
- dialect: The default dialect if no model dialect is configured. The format must adhere to Python's strftime codes.
- physical_schema_mapping: A mapping of regular expressions to match against the model schema to produce the corresponding physical schema
- default_catalog: The default catalog if no model catalog is configured.
- variables: The variables to pass to the model.
- kwargs: Additional kwargs to pass to the loader.
def
create_sql_model( name: Union[str, sqlglot.expressions.query.Table], query: Optional[sqlglot.expressions.core.Expr], **kwargs: Any) -> Union[SqlModel, SeedModel, PythonModel, ExternalModel]:
2402def create_sql_model( 2403 name: TableName, 2404 query: t.Optional[exp.Expr], 2405 **kwargs: t.Any, 2406) -> Model: 2407 """Creates a SQL model. 2408 2409 Args: 2410 name: The name of the model, which is of the form [catalog].[db].table. 2411 The catalog and db are optional. 2412 query: The model's logic in a form of a SELECT query. 2413 """ 2414 if not isinstance(query, (exp.Query, d.JinjaQuery, d.MacroFunc)): 2415 raise_config_error( 2416 "A query is required and must be a SELECT statement, a UNION statement, or a JINJA_QUERY block", 2417 kwargs.get("path"), 2418 ) 2419 assert isinstance(query, (exp.Query, d.JinjaQuery, d.MacroFunc)) 2420 2421 return _create_model(SqlModel, name, query=query, **kwargs)
Creates a SQL model.
Arguments:
- name: The name of the model, which is of the form [catalog].[db].table. The catalog and db are optional.
- query: The model's logic in a form of a SELECT query.
def
create_seed_model( name: Union[str, sqlglot.expressions.query.Table], seed_kind: sqlmesh.core.model.kind.SeedKind, *, path: Optional[pathlib.Path] = None, module_path: pathlib.Path = PosixPath('.'), **kwargs: Any) -> Union[SqlModel, SeedModel, PythonModel, ExternalModel]:
2424def create_seed_model( 2425 name: TableName, 2426 seed_kind: SeedKind, 2427 *, 2428 path: t.Optional[Path] = None, 2429 module_path: Path = Path(), 2430 **kwargs: t.Any, 2431) -> Model: 2432 """Creates a Seed model. 2433 2434 Args: 2435 name: The name of the model, which is of the form [catalog].[db].table. 2436 The catalog and db are optional. 2437 seed_kind: The information about the location of a seed and other related configuration. 2438 path: An optional path to the model definition file. 2439 from the macro registry. 2440 """ 2441 seed_path = Path(seed_kind.path) 2442 marker, *subdirs = seed_path.parts 2443 if marker.lower() == "$root": 2444 seed_path = module_path.joinpath(*subdirs) 2445 seed_kind.path = str(seed_path) 2446 elif not seed_path.is_absolute(): 2447 if path is None: 2448 seed_path = seed_path 2449 elif path.is_dir(): 2450 seed_path = path / seed_path 2451 else: 2452 seed_path = path.parent / seed_path 2453 2454 seed = create_seed(seed_path) 2455 2456 return _create_model( 2457 SeedModel, 2458 name, 2459 path=path, 2460 seed=seed, 2461 kind=seed_kind, 2462 depends_on=kwargs.pop("depends_on", None), 2463 module_path=module_path, 2464 **kwargs, 2465 )
Creates a Seed model.
Arguments:
- name: The name of the model, which is of the form [catalog].[db].table. The catalog and db are optional.
- seed_kind: The information about the location of a seed and other related configuration.
- path: An optional path to the model definition file. from the macro registry.
def
create_python_model( name: str, entrypoint: str, python_env: Dict[str, sqlmesh.utils.metaprogramming.Executable], *, macros: Optional[sqlmesh.utils.UniqueKeyDict[str, Union[sqlmesh.utils.metaprogramming.Executable, sqlmesh.core.macros.macro]]] = None, jinja_macros: Optional[sqlmesh.utils.jinja.JinjaMacroRegistry] = None, path: pathlib.Path = PosixPath('.'), module_path: pathlib.Path = PosixPath('.'), depends_on: Optional[Set[str]] = None, variables: Optional[Dict[str, Any]] = None, blueprint_variables: Optional[Dict[str, Any]] = None, **kwargs: Any) -> Union[SqlModel, SeedModel, PythonModel, ExternalModel]:
2468def create_python_model( 2469 name: str, 2470 entrypoint: str, 2471 python_env: t.Dict[str, Executable], 2472 *, 2473 macros: t.Optional[MacroRegistry] = None, 2474 jinja_macros: t.Optional[JinjaMacroRegistry] = None, 2475 path: Path = Path(), 2476 module_path: Path = Path(), 2477 depends_on: t.Optional[t.Set[str]] = None, 2478 variables: t.Optional[t.Dict[str, t.Any]] = None, 2479 blueprint_variables: t.Optional[t.Dict[str, t.Any]] = None, 2480 **kwargs: t.Any, 2481) -> Model: 2482 """Creates a Python model. 2483 2484 Args: 2485 name: The name of the model, which is of the form [catalog].[db].table. 2486 The catalog and db are optional. 2487 entrypoint: The name of a Python function which contains the data fetching / transformation logic. 2488 python_env: The Python environment of all objects referenced by the model implementation. 2489 path: An optional path to the model definition file. 2490 depends_on: The custom set of model's upstream dependencies. 2491 variables: The variables to pass to the model. 2492 blueprint_variables: The blueprint's variables to pass to the model. 2493 """ 2494 # Find dependencies for python models by parsing code if they are not explicitly defined 2495 # Also remove self-references that are found 2496 2497 dialect = kwargs.get("dialect") 2498 2499 dependencies_unspecified = depends_on is None 2500 2501 parsed_depends_on, referenced_variables = ( 2502 parse_dependencies( 2503 python_env, 2504 entrypoint, 2505 strict_resolution=dependencies_unspecified, 2506 variables=variables, 2507 blueprint_variables=blueprint_variables, 2508 ) 2509 if python_env is not None 2510 else (set(), set()) 2511 ) 2512 if dependencies_unspecified: 2513 depends_on = parsed_depends_on - {name} 2514 else: 2515 depends_on_rendered = render_expression( 2516 expression=exp.Array( 2517 expressions=[exp.maybe_parse(dep, dialect=dialect) for dep in depends_on or []] 2518 ), 2519 module_path=module_path, 2520 macros=macros, 2521 jinja_macros=jinja_macros, 2522 variables=variables, 2523 path=path, 2524 dialect=dialect, 2525 default_catalog=kwargs.get("default_catalog"), 2526 ) 2527 depends_on = { 2528 dep.sql(dialect=dialect) 2529 for dep in t.cast(t.List[exp.Expr], depends_on_rendered)[0].expressions 2530 } 2531 2532 used_variables = {k: v for k, v in (variables or {}).items() if k in referenced_variables} 2533 if used_variables: 2534 python_env[c.SQLMESH_VARS] = Executable.value(used_variables, sort_root_dict=True) 2535 2536 return _create_model( 2537 PythonModel, 2538 name, 2539 path=path, 2540 depends_on=depends_on, 2541 entrypoint=entrypoint, 2542 python_env=python_env, 2543 macros=macros, 2544 jinja_macros=jinja_macros, 2545 module_path=module_path, 2546 variables=variables, 2547 blueprint_variables=blueprint_variables, 2548 **kwargs, 2549 )
Creates a Python model.
Arguments:
- name: The name of the model, which is of the form [catalog].[db].table. The catalog and db are optional.
- entrypoint: The name of a Python function which contains the data fetching / transformation logic.
- python_env: The Python environment of all objects referenced by the model implementation.
- path: An optional path to the model definition file.
- depends_on: The custom set of model's upstream dependencies.
- variables: The variables to pass to the model.
- blueprint_variables: The blueprint's variables to pass to the model.
def
create_external_model( name: Union[str, sqlglot.expressions.query.Table], *, dialect: Optional[str] = None, path: pathlib.Path = PosixPath('.'), defaults: Optional[Dict[str, Any]] = None, **kwargs: Any) -> ExternalModel:
2552def create_external_model( 2553 name: TableName, 2554 *, 2555 dialect: t.Optional[str] = None, 2556 path: Path = Path(), 2557 defaults: t.Optional[t.Dict[str, t.Any]] = None, 2558 **kwargs: t.Any, 2559) -> ExternalModel: 2560 """Creates an external model. 2561 2562 Args: 2563 name: The name of the model, which is of the form [catalog].[db].table. 2564 The catalog and db are optional. 2565 dialect: The dialect to serialize. 2566 path: An optional path to the model definition file. 2567 """ 2568 return t.cast( 2569 ExternalModel, 2570 _create_model( 2571 ExternalModel, 2572 name, 2573 defaults=defaults, 2574 dialect=dialect, 2575 path=path, 2576 kind=ModelKindName.EXTERNAL.value, 2577 **kwargs, 2578 ), 2579 )
Creates an external model.
Arguments:
- name: The name of the model, which is of the form [catalog].[db].table. The catalog and db are optional.
- dialect: The dialect to serialize.
- path: An optional path to the model definition file.
INSERT_SEED_MACRO_CALL =
MacroFunc(
this=Anonymous(this=INSERT_SEED))
def
render_meta_fields( fields: Dict[str, Any], module_path: pathlib.Path, path: Optional[pathlib.Path], jinja_macros: Optional[sqlmesh.utils.jinja.JinjaMacroRegistry], macros: Optional[sqlmesh.utils.UniqueKeyDict[str, Union[sqlmesh.utils.metaprogramming.Executable, sqlmesh.core.macros.macro]]], dialect: Union[str, sqlglot.dialects.dialect.Dialect, type[sqlglot.dialects.dialect.Dialect], NoneType], variables: Optional[Dict[str, Any]], default_catalog: Optional[str], blueprint_variables: Optional[Dict[str, Any]] = None) -> Dict[str, Any]:
3017def render_meta_fields( 3018 fields: t.Dict[str, t.Any], 3019 module_path: Path, 3020 path: t.Optional[Path], 3021 jinja_macros: t.Optional[JinjaMacroRegistry], 3022 macros: t.Optional[MacroRegistry], 3023 dialect: DialectType, 3024 variables: t.Optional[t.Dict[str, t.Any]], 3025 default_catalog: t.Optional[str], 3026 blueprint_variables: t.Optional[t.Dict[str, t.Any]] = None, 3027) -> t.Dict[str, t.Any]: 3028 def render_field_value(value: t.Any) -> t.Any: 3029 if isinstance(value, exp.Expr) or (isinstance(value, str) and "@" in value): 3030 expression = exp.maybe_parse(value, dialect=dialect) 3031 rendered_expr = render_expression( 3032 expression=expression, 3033 module_path=module_path, 3034 macros=macros, 3035 jinja_macros=jinja_macros, 3036 variables=variables, 3037 path=path, 3038 dialect=dialect, 3039 default_catalog=default_catalog, 3040 blueprint_variables=blueprint_variables, 3041 ) 3042 if not rendered_expr: 3043 raise SQLMeshError( 3044 f"Rendering `{expression.sql(dialect=dialect)}` did not return an expression" 3045 ) 3046 3047 if len(rendered_expr) != 1: 3048 raise SQLMeshError( 3049 f"Rendering `{expression.sql(dialect=dialect)}` must return one result, but got {len(rendered_expr)}" 3050 ) 3051 3052 # For cases where a property is conditionally assigned 3053 if rendered_expr[0].sql().lower() in {"none", "null"}: 3054 return None 3055 3056 return rendered_expr[0] 3057 3058 return value 3059 3060 for field_name, field_info in ModelMeta.all_field_infos().items(): 3061 field = field_info.alias or field_name 3062 field_value = fields.get(field) 3063 3064 # We don't want to parse python model cron="@..." kwargs (e.g. @daily) into MacroVar 3065 if ( 3066 field == "cron" 3067 and isinstance(field_value, str) 3068 and field_value.lower() in CRON_SHORTCUTS 3069 ) or field_value is None: 3070 continue 3071 3072 if field in RUNTIME_RENDERED_MODEL_FIELDS: 3073 fields[field] = parse_strings_with_macro_refs(field_value, dialect) 3074 continue 3075 3076 if isinstance(field_value, dict): 3077 rendered_dict = {} 3078 for key, value in field_value.items(): 3079 if field == "columns": 3080 column_name = render_field_value(key) 3081 column_type = render_field_value(value) 3082 # If column_type is an Expr (from rendering macros), convert to string. 3083 # Otherwise, leave it as-is (string) for the validator to parse with the correct dialect. 3084 if isinstance(column_type, exp.Expr): 3085 column_type = column_type.sql(dialect=dialect) 3086 rendered_dict[column_name] = column_type 3087 elif key in RUNTIME_RENDERED_MODEL_FIELDS: 3088 rendered_dict[key] = parse_strings_with_macro_refs(value, dialect) 3089 elif ( 3090 # don't parse kind auto_restatement_cron="@..." kwargs (e.g. @daily) into MacroVar 3091 key == "auto_restatement_cron" 3092 and isinstance(value, str) 3093 and value.lower() in CRON_SHORTCUTS 3094 ): 3095 rendered_dict[key] = value 3096 elif (rendered := render_field_value(value)) is not None: 3097 rendered_dict[key] = rendered 3098 3099 if rendered_dict: 3100 fields[field] = rendered_dict 3101 else: 3102 fields.pop(field) 3103 elif isinstance(field_value, list): 3104 rendered_list = [ 3105 rendered 3106 for value in field_value 3107 if (rendered := render_field_value(value)) is not None 3108 ] 3109 if rendered_list: 3110 fields[field] = rendered_list 3111 else: 3112 fields.pop(field) 3113 else: 3114 rendered_field = render_field_value(field_value) 3115 if rendered_field is not None: 3116 fields[field] = rendered_field 3117 else: 3118 fields.pop(field) 3119 3120 return fields
def
render_model_defaults( defaults: Dict[str, Any], module_path: pathlib.Path, path: Optional[pathlib.Path], jinja_macros: Optional[sqlmesh.utils.jinja.JinjaMacroRegistry], macros: Optional[sqlmesh.utils.UniqueKeyDict[str, Union[sqlmesh.utils.metaprogramming.Executable, sqlmesh.core.macros.macro]]], dialect: Union[str, sqlglot.dialects.dialect.Dialect, type[sqlglot.dialects.dialect.Dialect], NoneType], variables: Optional[Dict[str, Any]], default_catalog: Optional[str]) -> Dict[str, Any]:
3123def render_model_defaults( 3124 defaults: t.Dict[str, t.Any], 3125 module_path: Path, 3126 path: t.Optional[Path], 3127 jinja_macros: t.Optional[JinjaMacroRegistry], 3128 macros: t.Optional[MacroRegistry], 3129 dialect: DialectType, 3130 variables: t.Optional[t.Dict[str, t.Any]], 3131 default_catalog: t.Optional[str], 3132) -> t.Dict[str, t.Any]: 3133 rendered_defaults = render_meta_fields( 3134 fields=defaults, 3135 module_path=module_path, 3136 macros=macros, 3137 jinja_macros=jinja_macros, 3138 variables=variables, 3139 path=path, 3140 dialect=dialect, 3141 default_catalog=default_catalog, 3142 ) 3143 3144 # Validate defaults that have macros are rendered to boolean 3145 for boolean in {"optimize_query", "allow_partials", "enabled"}: 3146 var = rendered_defaults.get(boolean) 3147 if var is not None and not isinstance(var, (exp.Boolean, bool)): 3148 raise ConfigError(f"Expected boolean for '{var}', got '{type(var)}' instead") 3149 3150 # Validate the 'interval_unit' if present is an Interval Unit 3151 var = rendered_defaults.get("interval_unit") 3152 if isinstance(var, str): 3153 try: 3154 rendered_defaults["interval_unit"] = IntervalUnit(var) 3155 except ValueError as e: 3156 raise ConfigError(f"Invalid interval unit: {var}") from e 3157 3158 return rendered_defaults
def
parse_defaults_properties( defaults: Dict[str, Any], dialect: Union[str, sqlglot.dialects.dialect.Dialect, type[sqlglot.dialects.dialect.Dialect], NoneType]) -> Dict[str, Any]:
3161def parse_defaults_properties( 3162 defaults: t.Dict[str, t.Any], dialect: DialectType 3163) -> t.Dict[str, t.Any]: 3164 for prop in PROPERTIES: 3165 default_properties = defaults.get(prop) 3166 for key, value in (default_properties or {}).items(): 3167 if isinstance(key, str) and d.SQLMESH_MACRO_PREFIX in str(value): 3168 defaults[prop][key] = exp.maybe_parse(value, dialect=dialect) 3169 3170 return defaults
def
render_expression( expression: sqlglot.expressions.core.Expr, module_path: pathlib.Path, path: Optional[pathlib.Path], jinja_macros: Optional[sqlmesh.utils.jinja.JinjaMacroRegistry] = None, macros: Optional[sqlmesh.utils.UniqueKeyDict[str, Union[sqlmesh.utils.metaprogramming.Executable, sqlmesh.core.macros.macro]]] = None, dialect: Union[str, sqlglot.dialects.dialect.Dialect, type[sqlglot.dialects.dialect.Dialect], NoneType] = None, variables: Optional[Dict[str, Any]] = None, default_catalog: Optional[str] = None, blueprint_variables: Optional[Dict[str, Any]] = None) -> Optional[List[sqlglot.expressions.core.Expr]]:
3173def render_expression( 3174 expression: exp.Expr, 3175 module_path: Path, 3176 path: t.Optional[Path], 3177 jinja_macros: t.Optional[JinjaMacroRegistry] = None, 3178 macros: t.Optional[MacroRegistry] = None, 3179 dialect: DialectType = None, 3180 variables: t.Optional[t.Dict[str, t.Any]] = None, 3181 default_catalog: t.Optional[str] = None, 3182 blueprint_variables: t.Optional[t.Dict[str, t.Any]] = None, 3183) -> t.Optional[t.List[exp.Expr]]: 3184 meta_python_env = make_python_env( 3185 expressions=expression, 3186 jinja_macro_references=None, 3187 module_path=module_path, 3188 macros=macros or macro.get_registry(), 3189 variables=variables, 3190 path=path, 3191 blueprint_variables=blueprint_variables, 3192 ) 3193 return ExpressionRenderer( 3194 expression, 3195 dialect, 3196 [], 3197 path=path, 3198 jinja_macro_registry=jinja_macros, 3199 python_env=meta_python_env, 3200 default_catalog=default_catalog, 3201 quote_identifiers=False, 3202 normalize_identifiers=False, 3203 ).render()
META_FIELD_CONVERTER: Dict[str, Callable] =
{'start': <function <lambda>>, 'cron': <function <lambda>>, 'cron_tz': <function <lambda>>, 'partitioned_by_': <function _single_expr_or_tuple>, 'clustered_by': <function _single_expr_or_tuple>, 'depends_on_': <function <lambda>>, 'pre': <function _list_of_calls_to_exp>, 'post': <function _list_of_calls_to_exp>, 'audits': <function _list_of_calls_to_exp>, 'columns_to_types_': <function <lambda>>, 'column_descriptions_': <function <lambda>>, 'tags': <function single_value_or_tuple>, 'grains': <function _refs_to_sql>, 'references': <function _refs_to_sql>, 'physical_properties_': <function <lambda>>, 'virtual_properties_': <function <lambda>>, 'session_properties_': <function <lambda>>, 'allow_partials': <function convert>, 'signals': <function <lambda>>, 'formatting': <class 'str'>, 'optimize_query': <class 'str'>, 'virtual_environment_mode': <function <lambda>>, 'dbt_node_info_': <function <lambda>>, 'grants_': <function <lambda>>, 'grants_target_layer': <function <lambda>>}
def
get_model_name(path: pathlib.Path) -> str:
def
clickhouse_partition_func( column: sqlglot.expressions.core.Expr, columns_to_types: Optional[Dict[str, sqlglot.expressions.datatypes.DataType]]) -> sqlglot.expressions.core.Expr:
3254def clickhouse_partition_func( 3255 column: exp.Expr, columns_to_types: t.Optional[t.Dict[str, exp.DataType]] 3256) -> exp.Expr: 3257 # `toMonday()` function accepts a Date or DateTime type column 3258 3259 col_type = (columns_to_types and columns_to_types.get(column.name)) or exp.DataType.build( 3260 "UNKNOWN" 3261 ) 3262 col_type_is_conformable = col_type.is_type( 3263 exp.DataType.Type.DATE, 3264 exp.DataType.Type.DATE32, 3265 exp.DataType.Type.DATETIME, 3266 exp.DataType.Type.DATETIME64, 3267 ) 3268 3269 # if input column is already a conformable type, just pass the column 3270 if col_type_is_conformable: 3271 return exp.func("toMonday", column, dialect="clickhouse") 3272 3273 # if input column type is not known, cast input to DateTime64 3274 if col_type.is_type(exp.DataType.Type.UNKNOWN): 3275 return exp.func( 3276 "toMonday", 3277 exp.cast(column, exp.DataType.build("DateTime64(9, 'UTC')", dialect="clickhouse")), 3278 dialect="clickhouse", 3279 ) 3280 3281 # if input column type is known but not conformable, cast input to DateTime64 and cast output back to original type 3282 return exp.cast( 3283 exp.func( 3284 "toMonday", 3285 exp.cast(column, exp.DataType.build("DateTime64(9, 'UTC')", dialect="clickhouse")), 3286 dialect="clickhouse", 3287 ), 3288 col_type, 3289 )
TIME_COL_PARTITION_FUNC =
{'clickhouse': <function clickhouse_partition_func>}