Chore: Move inline audits ouside the model meta object (#2834)

Author: izeigerman

sqlmesh/core/model/definition.py

@@ -36,6 +36,7 @@
     JinjaMacroRegistry,
     extract_macro_references_and_variables,
 )
+from sqlmesh.utils.pydantic import field_validator, field_validator_v1_args
 from sqlmesh.utils.metaprogramming import (
     Executable,
     build_env,
@@ -637,6 +638,10 @@ def disable_restatement(self) -> bool:
     def wap_supported(self) -> bool:
         return self.kind.is_materialized and (self.storage_format or "").lower() == "iceberg"
 
+    @property
+    def inline_audits(self) -> t.Dict[str, ModelAudit]:
+        return {}
+
     def validate_definition(self) -> None:
         """Validates the model's definition.
 
@@ -834,11 +839,30 @@ class _SqlBasedModel(_Model):
     post_statements_: t.Optional[t.List[exp.Expression]] = Field(
         default=None, alias="post_statements"
     )
+    inline_audits_: t.Dict[str, t.Any] = Field(default={}, alias="inline_audits")
 
     __statement_renderers: t.Dict[int, ExpressionRenderer] = {}
 
     _expression_validator = expression_validator
 
+    @field_validator("inline_audits_", mode="before")
+    @field_validator_v1_args
+    def _inline_audits_validator(cls, v: t.Any, values: t.Dict[str, t.Any]) -> t.Any:
+        if not isinstance(v, dict):
+            return {}
+
+        from sqlmesh.core.audit import ModelAudit
+
+        inline_audits = {}
+
+        for name, audit in v.items():
+            if isinstance(audit, ModelAudit):
+                inline_audits[name] = audit
+            elif isinstance(audit, dict):
+                inline_audits[name] = ModelAudit.parse_obj(audit)
+
+        return inline_audits
+
     def render_pre_statements(
         self,
         *,
@@ -900,6 +924,10 @@ def macro_definitions(self) -> t.List[d.MacroDef]:
         """All macro definitions from the list of expressions."""
         return [s for s in self.pre_statements + self.post_statements if isinstance(s, d.MacroDef)]
 
+    @property
+    def inline_audits(self) -> t.Dict[str, ModelAudit]:
+        return self.inline_audits_
+
     def _render_statements(
         self,
         statements: t.Iterable[exp.Expression],
@@ -1859,7 +1887,6 @@ def _create_model(
     depends_on: t.Optional[t.Set[str]] = None,
     dialect: t.Optional[str] = None,
     physical_schema_override: t.Optional[t.Dict[str, str]] = None,
-    inline_audits: t.Optional[t.Dict[str, ModelAudit]] = None,
     **kwargs: t.Any,
 ) -> Model:
     _validate_model_fields(klass, {"name", *kwargs} - {"grain", "table_properties"}, path)
@@ -1880,7 +1907,6 @@ def _create_model(
     try:
         model = klass(
             name=name,
-            inline_audits=inline_audits,
             **{
                 **(defaults or {}),
                 "jinja_macros": jinja_macros or JinjaMacroRegistry(),

sqlmesh/core/model/meta.py

@@ -67,7 +67,6 @@ class ModelMeta(_Node):
         default=None, alias="column_descriptions"
     )
     audits: t.List[AuditReference] = []
-    inline_audits: t.Dict[str, t.Any] = {}
     grains: t.List[exp.Expression] = []
     references: t.List[exp.Expression] = []
     physical_schema_override: t.Optional[str] = None
@@ -84,24 +83,6 @@ class ModelMeta(_Node):
     _default_catalog_validator = default_catalog_validator
     _depends_on_validator = depends_on_validator
 
-    @field_validator("inline_audits", mode="before")
-    @field_validator_v1_args
-    def _inline_audits_validator(cls, v: t.Any, values: t.Dict[str, t.Any]) -> t.Any:
-        if not isinstance(v, dict):
-            return {}
-
-        from sqlmesh.core.audit import ModelAudit
-
-        inline_audits = {}
-
-        for name, audit in v.items():
-            if isinstance(audit, ModelAudit):
-                inline_audits[name] = audit
-            elif isinstance(audit, dict):
-                inline_audits[name] = ModelAudit.parse_obj(audit)
-
-        return inline_audits
-
     @field_validator("audits", mode="before")
     def _audits_validator(cls, v: t.Any) -> t.Any:
         def extract(v: exp.Expression) -> t.Tuple[str, t.Dict[str, exp.Expression]]: