feat: skip state-sync for format and lint commands

sqlmesh/cli/main.py

@@ -41,6 +41,7 @@
     "table_name",
 )
 SKIP_CONTEXT_COMMANDS = ("init", "ui")
+LOCAL_ONLY_COMMANDS = ("format", "lint")
 
 
 def _sqlmesh_version() -> str:
@@ -115,6 +116,8 @@ def cli(
     configure_console(ignore_warnings=ignore_warnings)
 
     load = True
+    # Outside the single-path block: applies regardless of --paths count.
+    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,7 @@ 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).
         console: The rich instance used for printing out CLI command results.
         users: A list of users to make known to SQLMesh.
     """
@@ -386,6 +387,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 +415,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 +677,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 +687,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,100 @@ 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 _setup_local_only_project(tmp_path, mocker):
+    create_example_project(tmp_path, template=ProjectTemplate.EMPTY)
+    config_path = tmp_path / "config.yaml"
+    existing = config_path.read_text(encoding="utf-8")
+    config_path.write_text("project: cli_test\n\n" + existing, encoding="utf-8")
+
+    (tmp_path / "models" / "example.sql").write_text(
+        "MODEL(name local.example, dialect 'duckdb'); SELECT 1 AS col\n",
+        encoding="utf-8",
+    )
+
+    return mocker.patch(
+        "sqlmesh.core.state_sync.db.facade.EngineAdapterStateSync.get_versions",
+        side_effect=RuntimeError("state should not be accessed"),
+    )
+
+
+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_lint_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), "lint"])
+    assert result.exit_code == 0, f"Lint 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/linter/test_builtin.py

@@ -232,3 +232,51 @@ def test_no_missing_unit_tests(tmp_path, copy_to_temp_path):
         assert len(model_violations) == 0, (
             f"Model {model_name} should not have a violation since it has a test"
         )
+
+
+def test_lint_without_state_load(tmp_path, copy_to_temp_path, mocker) -> None:
+    sushi_paths = copy_to_temp_path("examples/sushi")
+    sushi_path = sushi_paths[0]
+
+    with open(sushi_path / "config.py", "r") as f:
+        read_file = f.read()
+
+    # Set a project name so state-merge code reaches the `self._load_state` guard.
+    project_anchor = "config = Config(\n    gateways="
+    assert project_anchor in read_file, (
+        "sushi config.py shape drifted; update project_anchor in test"
+    )
+    read_file = read_file.replace(
+        project_anchor,
+        'config = Config(\n    project="sushi",\n    gateways=',
+    )
+
+    # Enable one built-in rule so `lint_models` doesn't take the empty-rule-set path.
+    before = """    linter=LinterConfig(
+        enabled=False,
+        rules=[
+            "ambiguousorinvalidcolumn",
+            "invalidselectstarexpansion",
+            "noselectstar",
+            "nomissingaudits",
+            "nomissingowner",
+            "nomissingexternalmodels",
+        ],
+    ),"""
+    after = 'linter=LinterConfig(enabled=True, rules=["nomissingexternalmodels"]),'
+    assert before in read_file, (
+        "sushi config.py LinterConfig block shape drifted; update `before` in test"
+    )
+    read_file = read_file.replace(before, after)
+
+    with open(sushi_path / "config.py", "w") as f:
+        f.write(read_file)
+
+    mock = mocker.patch(
+        "sqlmesh.core.state_sync.db.facade.EngineAdapterStateSync.get_versions",
+        side_effect=RuntimeError("state should not be accessed"),
+    )
+
+    context = Context(paths=[sushi_path], load_state=False)
+    context.lint_models(raise_on_error=False)
+    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()