fix: inspect contracts through active connection

src/sqlbuild/cli/commands/_helpers/contract/progress.py

@@ -90,6 +90,7 @@ def write_contract_updates_with_progress(
     result: ContractAdoptionResult,
     request: ContractCommandRequest,
     adapter: BaseAdapter,
+    resolved_connection: dict[str, object],
     progress_stream: TextIO,
     use_progress_color: bool,
 ) -> ContractAdoptionResult:
@@ -110,6 +111,7 @@ def write_contract_updates_with_progress(
             result=result,
             overwrite=request.overwrite,
             adapter=adapter,
+            resolved_connection=resolved_connection,
             cli_vars=request.cli_vars,
         )
     except Exception:

src/sqlbuild/cli/commands/main/inspection/_contract.py

@@ -16,12 +16,19 @@
 )
 from sqlbuild.cli.commands.exceptions import CliUserError
 from sqlbuild.cli.commands.models import AdapterConnectionContext, ContractCommandRequest
-from sqlbuild.compiler.compile.models import CompiledModel, CompiledObjectKey, CompiledSource
+from sqlbuild.compiler.compile.models import (
+    CompileAnalysisSelection,
+    CompiledModel,
+    CompiledObjectKey,
+    CompiledProject,
+    CompiledSource,
+)
 from sqlbuild.compiler.contract_adoption.models import ContractAdoptionResult, ContractEvidence
 from sqlbuild.compiler.contract_adoption.types import ContractAction
 from sqlbuild.compiler.discovery.main.discover import discover_project_inputs
 from sqlbuild.compiler.discovery.models import DiscoveredProjectInputs
-from sqlbuild.compiler.pipeline.main.graph import build_project_graph
+from sqlbuild.compiler.pipeline.main.compiled_project import build_compiled_project
+from sqlbuild.compiler.pipeline.main.project_graph import build_project_graph_from_compiled_project
 from sqlbuild.compiler.pipeline.models import ProjectGraph
 from sqlbuild.compiler.planner.main.selection.selection import (
     resolve_project_selectors,
@@ -47,13 +54,15 @@ def run_contract(request: ContractCommandRequest) -> int:
         selected_target=None,
         cli_vars=request.cli_vars,
     )
-    graph: ProjectGraph = build_project_graph(
+    project: CompiledProject = build_compiled_project(
         discovered_inputs=discovered,
         adapter=context.adapter,
         selected_target=request.from_target,
         cli_vars=request.cli_vars,
-        no_cache=True,
+        resolved_connection=context.connection_config,
+        analysis_selection=CompileAnalysisSelection(no_cache=True),
     )
+    graph: ProjectGraph = build_project_graph_from_compiled_project(project=project)
     selected: frozenset[CompiledObjectKey] = resolve_project_selectors(
         select=request.select,
         exclude=request.exclude,
@@ -92,6 +101,7 @@ def run_contract(request: ContractCommandRequest) -> int:
             result=result,
             request=request,
             adapter=context.adapter,
+            resolved_connection=context.connection_config,
             progress_stream=progress_stream,
             use_progress_color=use_progress_color,
         )

src/sqlbuild/compiler/contract_adoption/main/write.py

@@ -6,7 +6,11 @@
 from pathlib import Path
 
 from sqlbuild.adapter.contract.classes.base_adapter import BaseAdapter
-from sqlbuild.compiler.compile.models import CompiledModel, CompiledSource
+from sqlbuild.compiler.compile.models import (
+    CompileAnalysisSelection,
+    CompiledModel,
+    CompiledSource,
+)
 from sqlbuild.compiler.compile.types import CompiledResourceType
 from sqlbuild.compiler.contract_adoption._helpers.repository_edits import (
     edit_model,
@@ -20,7 +24,7 @@
 )
 from sqlbuild.compiler.discovery.main.discover import discover_project_inputs
 from sqlbuild.compiler.discovery.models import DiscoveredProjectInputs
-from sqlbuild.compiler.pipeline.main.graph import build_project_graph
+from sqlbuild.compiler.pipeline.main.compiled_project import build_compiled_project
 from sqlbuild.compiler.pipeline.models import ProjectGraph
 from sqlbuild.lint.main.write_atomically import write_atomically
 
@@ -32,6 +36,7 @@ def write_contracts(
     result: ContractAdoptionResult,
     overwrite: bool,
     adapter: BaseAdapter | None = None,
+    resolved_connection: dict[str, object] | None = None,
     cli_vars: dict[str, object] | None = None,
 ) -> ContractAdoptionResult:
     """Apply safe source-only edits atomically and validate the resulting project."""
@@ -75,12 +80,13 @@ def write_contracts(
             write_atomically(path=path, contents=contents)
         discovered: DiscoveredProjectInputs = discover_project_inputs(project_dir=project_dir)
         if adapter is not None:
-            _ = build_project_graph(
+            _ = build_compiled_project(
                 discovered_inputs=discovered,
                 adapter=adapter,
                 selected_target=result.from_target,
+                resolved_connection=resolved_connection,
                 cli_vars=cli_vars,
-                no_cache=True,
+                analysis_selection=CompileAnalysisSelection(no_cache=True),
             )
     except Exception:
         for path, contents in originals.items():

tests/integration/src/sqlbuild/cli/commands/main/test_contract_command.py

@@ -333,6 +333,105 @@ def test_given_prod_target_credentials_when_contract_diffing_then_uses_active_co
     assert "Inspected 1 contract from target 'prod'" in captured.err
 
 
+@pytest.mark.parametrize(
+    "test_case",
+    [
+        ContractCommandIntegrationTestCase(
+            description="unavailable inspection target credentials do not block diff",
+            expected_exit_code=1,
+        )
+    ],
+    ids=lambda case: case.description,
+)
+def test_given_unavailable_inspection_target_credentials_when_diffing_then_uses_active_connection(
+    test_case: ContractCommandIntegrationTestCase,
+    tmp_path: Path,
+    monkeypatch: pytest.MonkeyPatch,
+    capsys: pytest.CaptureFixture[str],
+) -> None:
+    missing_variable: str = "SQLBUILD_TEST_UNAVAILABLE_PROD_DATABASE"
+    monkeypatch.delenv(missing_variable, raising=False)
+    _ = prepare_contract_project(
+        tmp_path,
+        prod_connection_toml=(
+            'connection = "prod_inspection"\n\n'
+            "[connections.prod_inspection]\n"
+            f'database = "${{ENV:{missing_variable}}}"\n'
+        ),
+    )
+
+    exit_code: int = main(
+        [
+            "--project-dir",
+            str(tmp_path),
+            "contract",
+            "diff",
+            "--from",
+            "prod",
+            "--select",
+            "orders",
+            "--json",
+        ]
+    )
+
+    captured: CaptureResult[str] = capsys.readouterr()
+    payload: dict[str, object] = json.loads(captured.out)
+    assert exit_code == test_case.expected_exit_code
+    assert payload["from_target"] == "prod"
+    assert payload["resources"][0]["relation"].endswith("prod.orders")
+    assert missing_variable not in captured.err
+
+
+@pytest.mark.parametrize(
+    "test_case",
+    [
+        ContractCommandIntegrationTestCase(
+            description="unavailable inspection target credentials do not block write validation",
+            expected_exit_code=0,
+        )
+    ],
+    ids=lambda case: case.description,
+)
+def test_given_unavailable_inspection_target_credentials_when_generating_then_validates_with_active_connection(
+    test_case: ContractCommandIntegrationTestCase,
+    tmp_path: Path,
+    monkeypatch: pytest.MonkeyPatch,
+    capsys: pytest.CaptureFixture[str],
+) -> None:
+    missing_variable: str = "SQLBUILD_TEST_UNAVAILABLE_PROD_DATABASE"
+    monkeypatch.delenv(missing_variable, raising=False)
+    _ = prepare_contract_project(
+        tmp_path,
+        prod_connection_toml=(
+            'connection = "prod_inspection"\n\n'
+            "[connections.prod_inspection]\n"
+            f'database = "${{ENV:{missing_variable}}}"\n'
+        ),
+    )
+    model_path: Path = tmp_path / "models" / "orders.sql"
+
+    exit_code: int = main(
+        [
+            "--no-color",
+            "--project-dir",
+            str(tmp_path),
+            "contract",
+            "generate",
+            "--from",
+            "prod",
+            "--select",
+            "orders",
+            "--write",
+        ]
+    )
+
+    captured: CaptureResult[str] = capsys.readouterr()
+    assert exit_code == test_case.expected_exit_code
+    assert "0 contract difference(s)" in captured.out
+    assert "id (type INTEGER)" in model_path.read_text(encoding="utf-8")
+    assert missing_variable not in captured.err
+
+
 @pytest.mark.parametrize(
     "test_case",
     [