Coverage for lintro / tools / definitions / sqlfluff.py: 96%
106 statements
« prev ^ index » next coverage.py v7.13.0, created at 2026-04-03 18:53 +0000
1"""SQLFluff tool definition.
3SQLFluff is a SQL linter and formatter with support for many SQL dialects.
4It parses SQL into an AST and performs linting rules on top of it.
5"""
7from __future__ import annotations
9import subprocess # nosec B404 - used safely with shell disabled
10from dataclasses import dataclass
11from typing import Any
13from lintro._tool_versions import get_min_version
14from lintro.enums.doc_url_template import DocUrlTemplate
15from lintro.enums.tool_name import ToolName
16from lintro.enums.tool_type import ToolType
17from lintro.models.core.tool_result import ToolResult
18from lintro.parsers.sqlfluff.sqlfluff_parser import parse_sqlfluff_output
19from lintro.plugins.base import BaseToolPlugin
20from lintro.plugins.file_processor import FileProcessingResult
21from lintro.plugins.protocol import ToolDefinition
22from lintro.plugins.registry import register_tool
23from lintro.tools.core.option_validators import (
24 filter_none_options,
25 validate_list,
26 validate_str,
27)
29# Constants for SQLFluff configuration
30SQLFLUFF_DEFAULT_TIMEOUT: int = 60
31SQLFLUFF_DEFAULT_PRIORITY: int = 50
32SQLFLUFF_FILE_PATTERNS: list[str] = ["*.sql"]
33SQLFLUFF_DEFAULT_FORMAT: str = "json"
36@register_tool
37@dataclass
38class SqlfluffPlugin(BaseToolPlugin):
39 """SQLFluff SQL linter and formatter plugin.
41 This plugin integrates SQLFluff with Lintro for linting and formatting
42 SQL files.
43 """
45 @property
46 def definition(self) -> ToolDefinition:
47 """Return the tool definition.
49 Returns:
50 ToolDefinition containing tool metadata.
51 """
52 return ToolDefinition(
53 name="sqlfluff",
54 description="SQL linter and formatter with dialect support",
55 can_fix=True,
56 tool_type=ToolType.LINTER | ToolType.FORMATTER,
57 file_patterns=SQLFLUFF_FILE_PATTERNS,
58 priority=SQLFLUFF_DEFAULT_PRIORITY,
59 conflicts_with=[],
60 native_configs=[".sqlfluff", "pyproject.toml"],
61 version_command=["sqlfluff", "--version"],
62 min_version=get_min_version(ToolName.SQLFLUFF),
63 default_options={
64 "timeout": SQLFLUFF_DEFAULT_TIMEOUT,
65 "dialect": None,
66 "exclude_rules": None,
67 "rules": None,
68 "templater": None,
69 },
70 default_timeout=SQLFLUFF_DEFAULT_TIMEOUT,
71 )
73 def set_options(
74 self,
75 dialect: str | None = None,
76 exclude_rules: list[str] | None = None,
77 rules: list[str] | None = None,
78 templater: str | None = None,
79 **kwargs: Any,
80 ) -> None:
81 """Set SQLFluff-specific options.
83 Args:
84 dialect: SQL dialect (ansi, bigquery, postgres, mysql, snowflake,
85 sqlite, etc.).
86 exclude_rules: List of rules to exclude.
87 rules: List of rules to include.
88 templater: Templater to use (raw, jinja, python, placeholder).
89 **kwargs: Other tool options.
90 """
91 validate_str(dialect, "dialect")
92 validate_list(exclude_rules, "exclude_rules")
93 validate_list(rules, "rules")
94 validate_str(templater, "templater")
96 options = filter_none_options(
97 dialect=dialect,
98 exclude_rules=exclude_rules,
99 rules=rules,
100 templater=templater,
101 )
102 super().set_options(**options, **kwargs)
104 def _build_lint_command(self, files: list[str]) -> list[str]:
105 """Build the sqlfluff lint command.
107 Args:
108 files: List of files to lint.
110 Returns:
111 List of command arguments.
112 """
113 cmd: list[str] = ["sqlfluff", "lint", "--format", SQLFLUFF_DEFAULT_FORMAT]
115 # Add dialect option
116 dialect_opt = self.options.get("dialect")
117 if dialect_opt is not None:
118 cmd.extend(["--dialect", str(dialect_opt)])
120 # Add exclude rules (comma-separated per SQLFluff CLI docs)
121 exclude_rules_opt = self.options.get("exclude_rules")
122 if isinstance(exclude_rules_opt, list) and exclude_rules_opt:
123 cmd.extend(["--exclude-rules", ",".join(map(str, exclude_rules_opt))])
125 # Add rules (comma-separated per SQLFluff CLI docs)
126 rules_opt = self.options.get("rules")
127 if isinstance(rules_opt, list) and rules_opt:
128 cmd.extend(["--rules", ",".join(map(str, rules_opt))])
130 # Add templater
131 templater_opt = self.options.get("templater")
132 if templater_opt is not None:
133 cmd.extend(["--templater", str(templater_opt)])
135 # Add end-of-options separator to handle filenames starting with '-'
136 cmd.append("--")
138 # Add files
139 cmd.extend(files)
141 return cmd
143 def _build_fix_command(self, files: list[str]) -> list[str]:
144 """Build the sqlfluff fix command.
146 Args:
147 files: List of files to fix.
149 Returns:
150 List of command arguments.
151 """
152 cmd: list[str] = ["sqlfluff", "fix", "--force"]
154 # Add dialect option
155 dialect_opt = self.options.get("dialect")
156 if dialect_opt is not None:
157 cmd.extend(["--dialect", str(dialect_opt)])
159 # Add exclude rules (comma-separated per SQLFluff CLI docs)
160 exclude_rules_opt = self.options.get("exclude_rules")
161 if isinstance(exclude_rules_opt, list) and exclude_rules_opt:
162 cmd.extend(["--exclude-rules", ",".join(map(str, exclude_rules_opt))])
164 # Add rules (comma-separated per SQLFluff CLI docs)
165 rules_opt = self.options.get("rules")
166 if isinstance(rules_opt, list) and rules_opt:
167 cmd.extend(["--rules", ",".join(map(str, rules_opt))])
169 # Add templater
170 templater_opt = self.options.get("templater")
171 if templater_opt is not None:
172 cmd.extend(["--templater", str(templater_opt)])
174 # Add end-of-options separator to handle filenames starting with '-'
175 cmd.append("--")
177 # Add files
178 cmd.extend(files)
180 return cmd
182 def _process_single_file_check(
183 self,
184 file_path: str,
185 timeout: int,
186 ) -> FileProcessingResult:
187 """Process a single SQL file with sqlfluff lint.
189 Args:
190 file_path: Path to the SQL file to process.
191 timeout: Timeout in seconds for the sqlfluff command.
193 Returns:
194 FileProcessingResult with check results for this file.
195 """
196 cmd = self._build_lint_command(files=[str(file_path)])
197 try:
198 success, output = self._run_subprocess(cmd=cmd, timeout=timeout)
199 issues = parse_sqlfluff_output(output=output)
200 # success is False if issues exist or tool failed
201 final_success = success and len(issues) == 0
202 return FileProcessingResult(
203 success=final_success,
204 output=output,
205 issues=issues,
206 )
207 except subprocess.TimeoutExpired:
208 return FileProcessingResult(
209 success=False,
210 output="",
211 issues=[],
212 skipped=True,
213 )
214 except (OSError, ValueError, RuntimeError) as e:
215 return FileProcessingResult(
216 success=False,
217 output="",
218 issues=[],
219 error=str(e),
220 )
222 def _process_single_file_fix(
223 self,
224 file_path: str,
225 timeout: int,
226 ) -> FileProcessingResult:
227 """Process a single SQL file with sqlfluff fix.
229 Args:
230 file_path: Path to the SQL file to fix.
231 timeout: Timeout in seconds for the sqlfluff command.
233 Returns:
234 FileProcessingResult with fix results for this file.
235 """
236 cmd = self._build_fix_command(files=[str(file_path)])
237 try:
238 success, output = self._run_subprocess(cmd=cmd, timeout=timeout)
239 return FileProcessingResult(
240 success=success,
241 output=output,
242 issues=[],
243 )
244 except subprocess.TimeoutExpired:
245 return FileProcessingResult(
246 success=False,
247 output="",
248 issues=[],
249 skipped=True,
250 )
251 except (OSError, ValueError, RuntimeError) as e:
252 return FileProcessingResult(
253 success=False,
254 output="",
255 issues=[],
256 error=str(e),
257 )
259 def doc_url(self, code: str) -> str | None:
260 """Return SQLFluff documentation URL for the given rule code.
262 Args:
263 code: SQLFluff rule code (e.g., "LT01").
265 Returns:
266 URL to the SQLFluff rule documentation.
267 """
268 if code:
269 return DocUrlTemplate.SQLFLUFF.format(code=code)
270 return None
272 def check(self, paths: list[str], options: dict[str, object]) -> ToolResult:
273 """Check files with SQLFluff.
275 Args:
276 paths: List of file or directory paths to check.
277 options: Runtime options that override defaults.
279 Returns:
280 ToolResult with check results.
281 """
282 # Use shared preparation for version check, path validation, file discovery
283 ctx = self._prepare_execution(paths, options)
284 if ctx.should_skip:
285 return ctx.early_result # type: ignore[return-value]
287 # Process files with progress bar support
288 def processor(file_path: str) -> FileProcessingResult:
289 return self._process_single_file_check(file_path, ctx.timeout)
291 result = self._process_files_with_progress(
292 files=ctx.files,
293 processor=processor,
294 timeout=ctx.timeout,
295 label="Processing files",
296 )
298 return ToolResult(
299 name=self.definition.name,
300 success=result.all_success,
301 output=result.build_output(timeout=ctx.timeout),
302 issues_count=result.total_issues,
303 issues=result.all_issues,
304 )
306 def fix(self, paths: list[str], options: dict[str, object]) -> ToolResult:
307 """Fix issues in files with SQLFluff.
309 Args:
310 paths: List of file or directory paths to fix.
311 options: Runtime options that override defaults.
313 Returns:
314 ToolResult with fix results.
315 """
316 # Use shared preparation for version check, path validation, file discovery
317 ctx = self._prepare_execution(paths, options)
318 if ctx.should_skip:
319 return ctx.early_result # type: ignore[return-value]
321 # Process files with progress bar support
322 def processor(file_path: str) -> FileProcessingResult:
323 return self._process_single_file_fix(file_path, ctx.timeout)
325 result = self._process_files_with_progress(
326 files=ctx.files,
327 processor=processor,
328 timeout=ctx.timeout,
329 label="Fixing files",
330 )
332 return ToolResult(
333 name=self.definition.name,
334 success=result.all_success,
335 output=result.build_output(timeout=ctx.timeout),
336 issues_count=0,
337 issues=[],
338 )