Respect audit skip flag (#700)

Author: vchan

sqlmesh/core/audit/definition.py

@@ -222,10 +222,12 @@ def _create_query_renderer(self, model: Model) -> QueryRenderer:
 class AuditResult(PydanticModel):
     audit: Audit
     """The audit this result is for."""
-    count: int
-    """The number of records returned by the audit query."""
-    query: exp.Expression
-    """The rendered query used by the audit."""
+    count: t.Optional[int]
+    """The number of records returned by the audit query. This could be None if the audit was skipped."""
+    query: t.Optional[exp.Expression]
+    """The rendered query used by the audit. This could be None if the audit was skipped."""
+    skipped: bool = False
+    """Whether this audit was skipped or not."""
 
 
 def _raise_config_error(msg: str, path: pathlib.Path) -> None:

sqlmesh/core/context.py

@@ -786,6 +786,7 @@ def audit(
         num_audits = sum(len(snapshot.model.audits) for snapshot in snapshots)
         self.console.log_status_update(f"Found {num_audits} audit(s).")
         errors = []
+        skipped_count = 0
         for snapshot in snapshots:
             for audit_result in self.snapshot_evaluator.audit(
                 snapshot=snapshot,
@@ -794,13 +795,19 @@ def audit(
                 snapshots=self.snapshots,
                 raise_exception=False,
             ):
-                if audit_result.count:
+                if audit_result.skipped:
+                    self.console.log_status_update(f"{audit_result.audit.name} SKIPPED.")
+                    skipped_count += 1
+                elif audit_result.count:
                     errors.append(audit_result)
                     self.console.log_status_update(f"{audit_result.audit.name} FAIL.")
                 else:
                     self.console.log_status_update(f"{audit_result.audit.name} PASS.")
 
-        self.console.log_status_update(f"\nFinished with {len(errors)} audit error(s).")
+        self.console.log_status_update(
+            f"\nFinished with {len(errors)} audit error{'' if len(errors) == 1 else 's'} "
+            f"and {skipped_count} audit{'' if skipped_count == 1 else 's'} skipped."
+        )
         for error in errors:
             self.console.log_status_update(
                 f"\nFailure in audit {error.audit.name} ({error.audit._path})."

sqlmesh/core/snapshot/evaluator.py

@@ -314,6 +314,9 @@ def audit(
         results = []
         for audit_name, audit_args in snapshot.model.audits:
             audit = audits_by_name[audit_name]
+            if audit.skip:
+                results.append(AuditResult(audit=audit, skipped=True))
+                continue
             query = audit.render_query(
                 snapshot,
                 start=start,