Fix style check and mypy errors from CI

- Apply ruff formatting to new/modified lines
- Fix mypy error in test_audit_only_no_nested_concurrency: use
  fully mocked evaluator instead of real evaluator with replaced
  method, avoiding type mismatch on call_count/call_args_list

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

Author: airhorns

sqlmesh/core/scheduler.py

@@ -38,7 +38,11 @@
 )
 from sqlmesh.core.state_sync import StateSync
 from sqlmesh.utils import CompletionStatus
-from sqlmesh.utils.concurrency import concurrent_apply_to_dag, concurrent_apply_to_values, NodeExecutionFailedError
+from sqlmesh.utils.concurrency import (
+    concurrent_apply_to_dag,
+    concurrent_apply_to_values,
+    NodeExecutionFailedError,
+)
 from sqlmesh.utils.dag import DAG
 from sqlmesh.utils.date import (
     TimeLike,
@@ -527,7 +531,9 @@ def run_merged_intervals(
                     }
 
                     dag = self._dag(
-                        batched_intervals, snapshot_dag=snapshot_dag, snapshots_to_create=snapshots_to_create
+                        batched_intervals,
+                        snapshot_dag=snapshot_dag,
+                        snapshots_to_create=snapshots_to_create,
                     )
 
                     def run_node(node: SchedulingUnit) -> None:
@@ -545,7 +551,8 @@ def run_node(node: SchedulingUnit) -> None:
 
                             # If batch_index > 0, then the target table must exist since the first batch would have created it
                             target_table_exists = (
-                                snapshot.snapshot_id not in snapshots_to_create or node.batch_index > 0
+                                snapshot.snapshot_id not in snapshots_to_create
+                                or node.batch_index > 0
                             )
 
                             def _do_evaluate() -> t.List[AuditResult]:
@@ -972,9 +979,7 @@ def _run_node_with_progress(
                 num_audits - num_audits_failed,
                 num_audits_failed,
                 execution_stats=execution_stats,
-                auto_restatement_triggers=auto_restatement_triggers.get(
-                    snapshot.snapshot_id
-                ),
+                auto_restatement_triggers=auto_restatement_triggers.get(snapshot.snapshot_id),
             )
 
     def _run_audits_concurrently(

sqlmesh/core/snapshot/evaluator.py

@@ -633,7 +633,9 @@ def _run_audit(
                 **kwargs,
             )
 
-        tasks_num = audit_concurrent_tasks if audit_concurrent_tasks is not None else self.concurrent_tasks
+        tasks_num = (
+            audit_concurrent_tasks if audit_concurrent_tasks is not None else self.concurrent_tasks
+        )
         results = concurrent_apply_to_values(
             prepared_audits,
             _run_audit,

tests/core/test_scheduler.py

@@ -1619,10 +1619,6 @@ def test_audit_only_no_nested_concurrency(mocker: MockerFixture, make_snapshot):
     This prevents nested thread pool multiplication: max_workers * concurrent_tasks threads hitting
     the DB at the same time.
     """
-    import sqlmesh.core.snapshot.evaluator as evaluator_module
-
-    spy = mocker.spy(evaluator_module, "concurrent_apply_to_values")
-
     snapshot_a = make_snapshot(SqlModel(name="a", query=parse_one("SELECT 1 as id")))
     snapshot_b = make_snapshot(SqlModel(name="b", query=parse_one("SELECT 2 as id")))
     snapshot_a.categorize_as(SnapshotChangeCategory.BREAKING)
@@ -1634,13 +1630,9 @@ def test_audit_only_no_nested_concurrency(mocker: MockerFixture, make_snapshot):
     mock_evaluator.concurrent_context.return_value.__enter__ = mocker.Mock(return_value=None)
     mock_evaluator.concurrent_context.return_value.__exit__ = mocker.Mock(return_value=False)
 
-    # Use the real SnapshotEvaluator to test the audit_concurrent_tasks parameter flows through
-    real_evaluator = SnapshotEvaluator(adapters=mocker.MagicMock(), concurrent_tasks=4)
-    real_evaluator.audit = mocker.MagicMock(return_value=[])  # type: ignore
-
     scheduler = Scheduler(
         snapshots=[snapshot_a, snapshot_b],
-        snapshot_evaluator=real_evaluator,
+        snapshot_evaluator=mock_evaluator,
         state_sync=mocker.MagicMock(),
         default_catalog=None,
         max_workers=2,
@@ -1661,10 +1653,10 @@ def test_audit_only_no_nested_concurrency(mocker: MockerFixture, make_snapshot):
 
     assert errors == []
     assert skipped == []
-    assert real_evaluator.audit.call_count == 2
+    assert mock_evaluator.audit.call_count == 2
 
     # Verify that audit_concurrent_tasks=1 was passed to each audit call to prevent nested pools
-    for call in real_evaluator.audit.call_args_list:
+    for call in mock_evaluator.audit.call_args_list:
         assert call.kwargs.get("audit_concurrent_tasks") == 1, (
             "audit_concurrent_tasks=1 must be passed to prevent nested thread pool multiplication"
         )