feat: skip state-sync for the format command

sqlmesh/cli/main.py

@@ -41,6 +41,7 @@
     "table_name",
 )
 SKIP_CONTEXT_COMMANDS = ("init", "ui")
+LOCAL_ONLY_COMMANDS = ("format",)
 
 
 def _sqlmesh_version() -> str:
@@ -115,6 +116,8 @@ def cli(
     configure_console(ignore_warnings=ignore_warnings)
 
     load = True
+    # Local-only gating must hold for any number of --paths, so it stays outside the block below.
+    load_state = ctx.invoked_subcommand not in LOCAL_ONLY_COMMANDS
 
     if len(paths) == 1:
         path = os.path.abspath(paths[0])
@@ -135,6 +138,7 @@ def cli(
             config=configs,
             gateway=gateway,
             load=load,
+            load_state=load_state,
         )
     except Exception:
         if debug:

sqlmesh/core/context.py

@@ -363,6 +363,9 @@ class GenericContext(BaseContext, t.Generic[C]):
             connection as it appears in configuration will be used.
         concurrent_tasks: The maximum number of tasks that can use the connection concurrently.
         load: Whether or not to automatically load all models and macros (default True).
+        load_state: Whether to merge remote state into the local project during load (default True).
+            Only intended for local-only operations like format; plan/apply in multi-repo projects
+            require it to see models owned by other projects.
         console: The rich instance used for printing out CLI command results.
         users: A list of users to make known to SQLMesh.
     """
@@ -386,6 +389,7 @@ def __init__(
         users: t.Optional[t.List[User]] = None,
         config_loader_kwargs: t.Optional[t.Dict[str, t.Any]] = None,
         selector: t.Optional[t.Type[Selector]] = None,
+        load_state: bool = True,
     ):
         self.configs = (
             config
@@ -413,6 +417,7 @@ def __init__(
         self._engine_adapter: t.Optional[EngineAdapter] = None
         self._linters: t.Dict[str, Linter] = {}
         self._loaded: bool = False
+        self._load_state: bool = load_state
         self._selector_cls = selector or NativeSelector
 
         self.path, self.config = t.cast(t.Tuple[Path, C], next(iter(self.configs.items())))
@@ -674,7 +679,7 @@ def load(self, update_schemas: bool = True) -> GenericContext[C]:
             )
 
         # Load environment statements from state for projects not in current load
-        if any(self._projects):
+        if self._load_state and any(self._projects):
             prod = self.state_reader.get_environment(c.PROD)
             if prod:
                 existing_statements = self.state_reader.get_environment_statements(c.PROD)
@@ -684,7 +689,7 @@ def load(self, update_schemas: bool = True) -> GenericContext[C]:
 
         uncached = set()
 
-        if any(self._projects):
+        if self._load_state and any(self._projects):
             prod = self.state_reader.get_environment(c.PROD)
 
             if prod:

tests/cli/test_cli.py

@@ -2237,3 +2237,146 @@ def test_format_leading_comma_default(runner: CliRunner, tmp_path: Path):
         assert result.exit_code == 0
     finally:
         del os.environ["SQLMESH__FORMAT__LEADING_COMMA"]
+
+
+def _create_local_only_project(path: Path, project: str) -> None:
+    path.mkdir(parents=True, exist_ok=True)
+    create_example_project(path, template=ProjectTemplate.EMPTY)
+    config_path = path / "config.yaml"
+    existing = config_path.read_text(encoding="utf-8")
+    config_path.write_text(f"project: {project}\n\n" + existing, encoding="utf-8")
+
+    (path / "models" / "example.sql").write_text(
+        f"MODEL(name {project}.example, dialect 'duckdb'); SELECT 1 AS col\n",
+        encoding="utf-8",
+    )
+
+
+def _patch_state_access(mocker):
+    return mocker.patch(
+        "sqlmesh.core.state_sync.db.facade.EngineAdapterStateSync.get_versions",
+        side_effect=RuntimeError("state should not be accessed"),
+    )
+
+
+def _setup_local_only_project(tmp_path, mocker):
+    _create_local_only_project(tmp_path, "cli_test")
+    return _patch_state_access(mocker)
+
+
+def test_format_runs_without_state(runner: CliRunner, tmp_path: Path, mocker):
+    mock = _setup_local_only_project(tmp_path, mocker)
+    result = runner.invoke(cli, ["--paths", str(tmp_path), "format"])
+    assert result.exit_code == 0, f"Format failed: {result.output}\nException: {result.exception}"
+    mock.assert_not_called()
+
+
+def test_format_runs_without_state_multi_repo_partial(runner: CliRunner, copy_to_temp_path, mocker):
+    """Format one repo of a multi-repo project whose upstream models live only in prod state."""
+    repo_2 = copy_to_temp_path("examples/multi")[0] / "repo_2"
+    mock = _patch_state_access(mocker)
+
+    result = runner.invoke(cli, ["--gateway", "memory", "--paths", str(repo_2), "format"])
+    assert result.exit_code == 0, f"Format failed: {result.output}\nException: {result.exception}"
+    mock.assert_not_called()
+
+
+def test_lint_still_loads_state(runner: CliRunner, tmp_path: Path, mocker):
+    """Guard that `lint` explicitly passes `load_state=True` and still reaches state sync."""
+    mock = _setup_local_only_project(tmp_path, mocker)
+    init_spy = mocker.spy(Context, "__init__")
+
+    runner.invoke(cli, ["--paths", str(tmp_path), "lint"])
+
+    assert init_spy.called, "Context was never constructed"
+    for call in init_spy.call_args_list:
+        assert "load_state" in call.kwargs, (
+            "CLI didn't pass load_state= explicitly; missing kwarg defaults to True silently"
+        )
+        assert call.kwargs["load_state"] is True, (
+            f"Context was constructed with load_state={call.kwargs['load_state']} for `lint`"
+        )
+    assert mock.called, "state-sync was never accessed during `lint`"
+
+
+@pytest.mark.parametrize("command", ["format"])
+def test_local_only_commands_skip_state_multiple_paths(
+    runner: CliRunner, tmp_path: Path, mocker, command: str
+):
+    project_a = tmp_path / "a"
+    project_b = tmp_path / "b"
+    _create_local_only_project(project_a, "proj_a")
+    _create_local_only_project(project_b, "proj_b")
+    mock = _patch_state_access(mocker)
+
+    result = runner.invoke(cli, ["--paths", str(project_a), "--paths", str(project_b), command])
+    assert result.exit_code == 0, (
+        f"{command} failed: {result.output}\nException: {result.exception}"
+    )
+    mock.assert_not_called()
+
+
+def test_plan_still_loads_state(runner: CliRunner, tmp_path: Path, mocker):
+    """Guard that `plan` explicitly passes `load_state=True` and still reaches state sync."""
+    mock = _setup_local_only_project(tmp_path, mocker)
+    init_spy = mocker.spy(Context, "__init__")
+
+    runner.invoke(cli, ["--paths", str(tmp_path), "plan"], input="n\n")
+
+    assert init_spy.called, "Context was never constructed"
+    for call in init_spy.call_args_list:
+        assert "load_state" in call.kwargs, (
+            "CLI didn't pass load_state= explicitly; missing kwarg defaults to True silently"
+        )
+        assert call.kwargs["load_state"] is True, (
+            f"Context was constructed with load_state={call.kwargs['load_state']} for `plan`"
+        )
+    assert mock.called, "state-sync was never accessed during `plan`"
+
+
+def test_format_does_not_open_state_connection(
+    runner: CliRunner, tmp_path: Path, mocker, monkeypatch
+):
+    """Format must not open a configured remote Postgres state connection when CI secrets are unset."""
+    pytest.importorskip("psycopg2")
+
+    for var in ("PG_HOST", "PG_USER", "PG_PASSWORD", "PG_DATABASE"):
+        monkeypatch.delenv(var, raising=False)
+
+    create_example_project(tmp_path, template=ProjectTemplate.EMPTY)
+    (tmp_path / "config.yaml").write_text(
+        """project: cli_test
+
+gateways:
+  prod:
+    state_connection:
+      type: postgres
+      host: "{{ env_var('PG_HOST', 'postgres.internal.example.com') }}"
+      port: 5432
+      user: "{{ env_var('PG_USER') }}"
+      password: "{{ env_var('PG_PASSWORD') }}"
+      database: "{{ env_var('PG_DATABASE', 'sqlmesh_state') }}"
+    connection:
+      type: duckdb
+      database: "warehouse.db"
+
+default_gateway: prod
+
+model_defaults:
+  dialect: duckdb
+""",
+        encoding="utf-8",
+    )
+    (tmp_path / "models" / "example.sql").write_text(
+        "MODEL(name local.example, dialect 'duckdb'); SELECT 1 AS col\n",
+        encoding="utf-8",
+    )
+
+    mock = mocker.patch(
+        "sqlmesh.core.state_sync.db.facade.EngineAdapterStateSync.get_versions",
+        side_effect=RuntimeError("state should not be accessed"),
+    )
+
+    result = runner.invoke(cli, ["--paths", str(tmp_path), "format"])
+    assert result.exit_code == 0, f"Format failed: {result.output}\nException: {result.exception}"
+    mock.assert_not_called()

tests/core/test_format.py

@@ -144,3 +144,20 @@ def test_ignore_formating_files(tmp_path: pathlib.Path):
         model3.read_text(encoding="utf-8")
         == "MODEL (\n  name this.model3,\n  dialect 'duckdb',\n  formatting TRUE\n);\n\nSELECT\n  1 AS col"
     )
+
+
+def test_format_without_state_load(tmp_path: pathlib.Path, mocker: MockerFixture):
+    mock = mocker.patch(
+        "sqlmesh.core.state_sync.db.facade.EngineAdapterStateSync.get_versions",
+        side_effect=RuntimeError("state should not be accessed"),
+    )
+
+    create_temp_file(
+        tmp_path,
+        pathlib.Path("models/example.sql"),
+        "MODEL(name local.example, dialect 'duckdb'); SELECT 1 AS col",
+    )
+
+    context = Context(paths=tmp_path, config=Config(project="local_only"), load_state=False)
+    context.format(check=True)
+    mock.assert_not_called()