feat: support linting or formatting whole project with multithreading and manage dbt deferral from server

Author: z3z1ma

src/dbt_core_interface/client.py

@@ -217,7 +217,13 @@ def lint_sql(
         if raw_sql is not None and sql_path is None:
             data = raw_sql
             headers = {"Content-Type": "text/plain"}
-        resp = self._request("POST", "/api/v1/lint", params=params, data=data, headers=headers)
+        resp = self._request(
+            "POST" if data is not None else "GET",
+            "/api/v1/lint",
+            params=params,
+            data=data,
+            headers=headers,
+        )
         return ServerLintResult.model_validate(resp.json())
 
     def format_sql(
@@ -237,7 +243,13 @@ def format_sql(
         if raw_sql is not None and sql_path is None:
             data = raw_sql
             headers = {"Content-Type": "text/plain"}
-        resp = self._request("POST", "/api/v1/format", params=params, data=data, headers=headers)
+        resp = self._request(
+            "POST" if data is not None else "GET",
+            "/api/v1/format",
+            params=params,
+            data=data,
+            headers=headers,
+        )
         return ServerFormatResult.model_validate(resp.json())
 
     def parse_project(
@@ -290,6 +302,24 @@ def command(
     source_freshness = functools.partialmethod(command, "source freshness")
     test = functools.partialmethod(command, "test")
 
+    def inject_state(self, directory: Path | str) -> dict[str, t.Any]:
+        """Inject manifest state for dbt deferral into the server."""
+        resp = self._request(
+            "GET",
+            "/api/v1/state",
+            json_payload={"directory": str(directory)},
+        )
+        if resp.ok:
+            return {}
+        return resp.json()
+
+    def clear_state(self) -> dict[str, t.Any]:
+        """Clear the deferral manifest state on the server."""
+        resp = self._request("DELETE", "/api/v1/state")
+        if resp.ok:
+            return {}
+        return resp.json()
+
     def status(self) -> dict[str, t.Any]:
         """Check server diagnostic status."""
         resp = self._request("GET", "/api/v1/status")

src/dbt_core_interface/project.py

@@ -801,7 +801,7 @@ def lint(
         extra_config_path: Path | str | None = None,
         ignore_local_config: bool = False,
         fluff_conf: FluffConfig | None = None,
-    ) -> LintingRecord | None:
+    ) -> list[LintingRecord]:
         """Lint specified file or SQL string."""
         from sqlfluff.cli.commands import get_linter_and_formatter
 
@@ -815,15 +815,34 @@ def lint(
         lint, _ = get_linter_and_formatter(fluff_conf)
 
         if sql is None:
-            # TODO: lint whole project
-            return
+
+            def _lint(node: ManifestNode) -> list[LintingRecord]:
+                records: list[LintingRecord] = []
+                try:
+                    if node.resource_type == "model":
+                        records = self.lint(
+                            Path(node.original_file_path),
+                            extra_config_path=extra_config_path,
+                            ignore_local_config=ignore_local_config,
+                            fluff_conf=fluff_conf,
+                        )
+                except Exception as e:
+                    logger.error(f"Error formatting node {node.name}: {e}")
+                return records
+
+            all_records: list[LintingRecord] = []
+            for records in self.pool.map(_lint, self.manifest.nodes.values()):
+                all_records.extend(records)
+
+            return all_records
         elif isinstance(sql, str):
             result = lint.lint_string_wrapped(sql)
         else:
+            if not sql.is_absolute() and not sql.exists():
+                sql = self.project_root / sql
             result = lint.lint_paths((str(sql),), ignore_files=False)
 
-        records = result.as_records()
-        return records[0] if records else None
+        return result.as_records()
 
     def format(
         self,
@@ -836,13 +855,6 @@ def format(
         from sqlfluff.cli.commands import get_linter_and_formatter
         from sqlfluff.core import SQLLintError
 
-        logger.info(f"""format_command(
-        {self.project_root},
-        {str(sql)[:100]},
-        {extra_config_path},
-        {ignore_local_config})
-        """)
-
         fluff_conf = fluff_conf or self.get_sqlfluff_configuration(
             sql if isinstance(sql, Path) else None,
             extra_config_path,
@@ -868,11 +880,27 @@ def format(
 
         result_sql = None
         if sql is None:
-            # TODO: format whole project
-            return True, result_sql
+
+            def _format(node: ManifestNode) -> bool:
+                success = True
+                try:
+                    if node.resource_type == "model":
+                        success, _ = self.format(
+                            Path(node.original_file_path),
+                            extra_config_path=extra_config_path,
+                            ignore_local_config=ignore_local_config,
+                            fluff_conf=fluff_conf,
+                        )
+                except Exception as e:
+                    logger.error(f"Error formatting node {node.name}: {e}")
+                    success = False
+                return success
+
+            return all(res for res in self.pool.map(_format, self.manifest.nodes.values())), None
         if isinstance(sql, str):
             logger.info(f"Formatting SQL string: {sql[:100]}")
             result = lint.lint_string_wrapped(sql, fname="stdin", fix=True)
+
             _, num_filtered_errors = result.count_tmp_prs_errors()
             result.discard_fixes_for_lint_errors_in_files_with_tmp_or_prs_errors()
             success = not num_filtered_errors
@@ -886,6 +914,8 @@ def format(
                     logger.info("No fixable errors in SQL string")
                     result_sql = sql
         else:
+            if not sql.is_absolute() and not sql.exists():
+                sql = self.project_root / sql
             logger.info(f"Formatting SQL file: {sql}")
             before_modified = datetime.fromtimestamp(sql.stat().st_mtime).strftime(
                 "%Y-%m-%d %H:%M:%S"
@@ -898,6 +928,7 @@ def format(
                 apply_fixes=True,
                 fix_even_unparsable=False,
             )
+
             _, num_filtered_errors = lint_result.count_tmp_prs_errors()
             lint_result.discard_fixes_for_lint_errors_in_files_with_tmp_or_prs_errors()
             success = not num_filtered_errors