Enable audits when evaluating snapshots in Airflow. Fixes #35 (#44)

Author: izeigerman

sqlmesh/core/scheduler.py

@@ -90,7 +90,6 @@ def evaluate(
             mapping=mapping,
             **kwargs,
         )
-        self.state_sync.add_interval(snapshot.snapshot_id, start, end)
         self.snapshot_evaluator.audit(
             snapshot=snapshot,
             start=start,
@@ -99,6 +98,7 @@ def evaluate(
             mapping=mapping,
             **kwargs,
         )
+        self.state_sync.add_interval(snapshot.snapshot_id, start, end)
         self.console.update_snapshot_progress(snapshot.name, 1)
 
     def run(

sqlmesh/core/snapshot_evaluator.py

@@ -76,6 +76,7 @@ def evaluate(
                 as a dataframe.
             kwargs: Additional kwargs to pass to the renderer.
         """
+        logger.info("Evaluating snapshot %s", snapshot.snapshot_id)
         if snapshot.is_embedded_kind:
             return None
 
@@ -220,6 +221,7 @@ def audit(
             collection_exceptions:
             kwargs: Additional kwargs to pass to the renderer.
         """
+        logger.info("Auditing snapshot %s", snapshot.snapshot_id)
         results = []
         for audit, query in snapshot.model.render_audit_queries(
             start=start,

sqlmesh/engines/commands.py

@@ -56,6 +56,13 @@ def evaluate(
         command_payload.latest,
         mapping=command_payload.table_mapping,
     )
+    evaluator.audit(
+        snapshot=command_payload.snapshot,
+        start=command_payload.start,
+        end=command_payload.end,
+        latest=command_payload.latest,
+        mapping=command_payload.table_mapping,
+    )
 
 
 def promote(