style: pre-commit auto-fixes

[pre-commit.ci] auto-applied fixes from configured hooks

Author: pre-commit-ci[bot]

CHANGELOG.md

@@ -219,7 +219,7 @@ Headline release: schema-aware linting via the new **Contracts pack**, three com
 - `test_duration_tracked` no longer fails on fast hardware where
   `time.perf_counter` resolution is coarser than the scan duration.
   Assertion relaxed from `> 0` to `>= 0` with an explicit float type check.
-- Added W014: warn on OVER() without ORDER BY / PARTITION BY to flag non-deterministic window 
+- Added W014: warn on OVER() without ORDER BY / PARTITION BY to flag non-deterministic window
   functions.
 
 ## [0.4.1] - 2026-04-19

scripts/scaffold_rule.py

@@ -76,7 +76,7 @@ def check_line(self, line: str, line_number: int, file: str):
 '''
 
 
-TEST_TEMPLATE = '''\
+TEST_TEMPLATE = """\
 def test_{code_lower}_fires_on_bad_sql():
     rule = {klass}()
     finding = _line(rule, {bad!r})
@@ -87,7 +87,7 @@ def test_{code_lower}_fires_on_bad_sql():
 def test_{code_lower}_passes_on_safe_sql():
     rule = {klass}()
     assert _line(rule, {good!r}) is None
-'''
+"""
 
 
 def main() -> int:
@@ -96,8 +96,12 @@ def main() -> int:
         formatter_class=argparse.RawDescriptionHelpFormatter,
         epilog=__doc__.split("Example:")[1] if "Example:" in __doc__ else "",
     )
-    parser.add_argument("--code", required=True, help="Rule code, e.g. W024 (must match [EWTSP]NNN)")
-    parser.add_argument("--name", required=True, help="Rule name in kebab-case, e.g. negate-of-equality")
+    parser.add_argument(
+        "--code", required=True, help="Rule code, e.g. W024 (must match [EWTSP]NNN)"
+    )
+    parser.add_argument(
+        "--name", required=True, help="Rule name in kebab-case, e.g. negate-of-equality"
+    )
     parser.add_argument("--description", required=True, help="One-line description for the rule")
     parser.add_argument("--regex", default="YOUR_PATTERN_HERE", help="Regex pattern to match")
     parser.add_argument("--message", help="Finding message (defaults to description)")
@@ -106,8 +110,12 @@ def main() -> int:
         default="Refactor for clarity and performance",
         help="Suggested fix shown alongside the finding",
     )
-    parser.add_argument("--bad", default="-- TODO: bad SQL example", help="SQL example that should fire")
-    parser.add_argument("--good", default="-- TODO: safe SQL example", help="SQL example that should not fire")
+    parser.add_argument(
+        "--bad", default="-- TODO: bad SQL example", help="SQL example that should fire"
+    )
+    parser.add_argument(
+        "--good", default="-- TODO: safe SQL example", help="SQL example that should not fire"
+    )
     args = parser.parse_args()
 
     code = args.code.upper()

sql_guard/checker.py

@@ -127,29 +127,35 @@ def check_file(
         try:
             content = path.read_text(encoding="latin-1")
         except Exception as e:
-            return [Finding(
+            return [
+                Finding(
+                    rule_id="SYS",
+                    severity="error",
+                    file=file_str,
+                    line=0,
+                    message=f"Cannot read file: {e}",
+                )
+            ]
+    except PermissionError:
+        return [
+            Finding(
                 rule_id="SYS",
                 severity="error",
                 file=file_str,
                 line=0,
-                message=f"Cannot read file: {e}",
-            )]
-    except PermissionError:
-        return [Finding(
-            rule_id="SYS",
-            severity="error",
-            file=file_str,
-            line=0,
-            message="Permission denied",
-        )]
+                message="Permission denied",
+            )
+        ]
     except OSError as e:
-        return [Finding(
-            rule_id="SYS",
-            severity="error",
-            file=file_str,
-            line=0,
-            message=f"Cannot read file: {e}",
-        )]
+        return [
+            Finding(
+                rule_id="SYS",
+                severity="error",
+                file=file_str,
+                line=0,
+                message=f"Cannot read file: {e}",
+            )
+        ]
 
     single_pass_rules = [r for r in rules if not r.multiline]
     multi_line_rules = [r for r in rules if r.multiline]

sql_guard/cli.py

@@ -30,9 +30,13 @@
 @app.command("check")
 def check_cmd(
     paths: list[str] = typer.Argument(default=None, help="Files or directories to check."),
-    severity: str = typer.Option("warning", "--severity", "-s", help="Minimum severity: error or warning."),
+    severity: str = typer.Option(
+        "warning", "--severity", "-s", help="Minimum severity: error or warning."
+    ),
     fail_fast: bool = typer.Option(False, "--fail-fast", help="Stop after first error."),
-    disable: Optional[list[str]] = typer.Option(None, "--disable", "-d", help="Rule IDs to disable."),
+    disable: Optional[list[str]] = typer.Option(
+        None, "--disable", "-d", help="Rule IDs to disable."
+    ),
     include_python: bool = typer.Option(
         False,
         "--include-python",
@@ -96,14 +100,10 @@ def check_cmd(
         try:
             contract = Contract.from_file(effective_contract_path)
         except FileNotFoundError:
-            console.print(
-                f"[red]Contract file not found:[/red] {effective_contract_path}"
-            )
+            console.print(f"[red]Contract file not found:[/red] {effective_contract_path}")
             raise typer.Exit(code=2)
         except Exception as exc:
-            console.print(
-                f"[red]Failed to load contract {effective_contract_path}:[/red] {exc}"
-            )
+            console.print(f"[red]Failed to load contract {effective_contract_path}:[/red] {exc}")
             raise typer.Exit(code=2)
 
     if changed_only:
@@ -200,12 +200,7 @@ def validate_contract_cmd(
 
     column_count = sum(len(t.columns) for t in contract.tables.values())
     pk_count = sum(len(t.primary_keys) for t in contract.tables.values())
-    fk_count = sum(
-        1
-        for t in contract.tables.values()
-        for c in t.columns.values()
-        if c.foreign_key
-    )
+    fk_count = sum(1 for t in contract.tables.values() for c in t.columns.values() if c.foreign_key)
 
     console.print(
         f"[green]OK[/green] {contract_path}: {table_count} tables, "
@@ -251,9 +246,7 @@ def schema_snapshot_cmd(
     from sql_guard import snapshot as snapshot_mod
 
     try:
-        data = snapshot_mod.introspect(
-            dsn=dsn, schema=schema, include_tables=include_table
-        )
+        data = snapshot_mod.introspect(dsn=dsn, schema=schema, include_tables=include_table)
     except snapshot_mod.SnapshotError as exc:
         console.print(f"[red]{exc}[/red]")
         raise typer.Exit(code=2)
@@ -264,9 +257,7 @@ def schema_snapshot_cmd(
     snapshot_mod.write_snapshot(data, output_path)
 
     table_count = len(data.get("tables") or {})
-    console.print(
-        f"[green]OK[/green] wrote {output_path} with {table_count} tables."
-    )
+    console.print(f"[green]OK[/green] wrote {output_path} with {table_count} tables.")
 
 
 @app.command()

sql_guard/fluent.py

@@ -67,10 +67,7 @@ def summary(self) -> str:
         if not parts:
             return f"no issues in {self.files_checked} file{'s' if self.files_checked != 1 else ''}"
         count_str = ", ".join(parts)
-        return (
-            f"{count_str} in "
-            f"{self.files_checked} file{'s' if self.files_checked != 1 else ''}"
-        )
+        return f"{count_str} in {self.files_checked} file{'s' if self.files_checked != 1 else ''}"
 
     def __len__(self) -> int:
         return len(self.findings)
@@ -123,7 +120,10 @@ def scan(self, sql_string: str) -> ScanResult:
         rules = get_rules(enabled_ids=self._enabled, disabled_ids=self._disabled)
 
         with tempfile.NamedTemporaryFile(
-            mode="w", suffix=".sql", delete=False, encoding="utf-8",
+            mode="w",
+            suffix=".sql",
+            delete=False,
+            encoding="utf-8",
         ) as tmp:
             tmp.write(sql_string)
             tmp_path = Path(tmp.name)

sql_guard/git_filter.py

@@ -54,9 +54,7 @@ def changed_files(base: str | None = None) -> set[Path] | None:
     return out
 
 
-def filter_to_changed(
-    discovered: list[Path], base: str | None = None
-) -> tuple[list[Path], bool]:
+def filter_to_changed(discovered: list[Path], base: str | None = None) -> tuple[list[Path], bool]:
     """Filter ``discovered`` to only the files git reports as changed.
 
     Returns ``(filtered_paths, used_git)``. When git isn't available the

sql_guard/python_scanner.py

@@ -175,7 +175,9 @@ def visit_Call(self, node: "cst.Call") -> None:  # noqa: N802 - libcst API
         if target is None:
             return
         if not (
-            isinstance(target, (cst.SimpleString, cst.ConcatenatedString, cst.FormattedString, cst.Name))
+            isinstance(
+                target, (cst.SimpleString, cst.ConcatenatedString, cst.FormattedString, cst.Name)
+            )
             or isinstance(target, cst.BinaryOperation)
             or _is_dot_format(target)
         ):

sql_guard/reporters/sarif.py

@@ -17,8 +17,7 @@
 
 INFO_URI = "https://github.com/Pawansingh3889/sql-guard"
 SARIF_SCHEMA = (
-    "https://docs.oasis-open.org/sarif/sarif/v2.1.0/cos02/schemas/"
-    "sarif-schema-2.1.0.json"
+    "https://docs.oasis-open.org/sarif/sarif/v2.1.0/cos02/schemas/sarif-schema-2.1.0.json"
 )
 
 
@@ -31,14 +30,16 @@ def _all_rule_descriptors() -> list[dict]:
     """Build the SARIF rule catalogue."""
     descriptors: list[dict] = []
     for rule in [*ALL_RULES, *PYTHON_RULES]:
-        descriptors.append({
-            "id": rule.id,
-            "name": rule.name,
-            "shortDescription": {"text": rule.description},
-            "fullDescription": {"text": rule.description},
-            "defaultConfiguration": {"level": _level(rule.severity)},
-            "helpUri": INFO_URI,
-        })
+        descriptors.append(
+            {
+                "id": rule.id,
+                "name": rule.name,
+                "shortDescription": {"text": rule.description},
+                "fullDescription": {"text": rule.description},
+                "defaultConfiguration": {"level": _level(rule.severity)},
+                "helpUri": INFO_URI,
+            }
+        )
     return descriptors
 
 
@@ -50,14 +51,16 @@ def build(result: CheckResult) -> dict:
             "ruleId": finding.rule_id,
             "level": _level(finding.severity),
             "message": {"text": finding.message},
-            "locations": [{
-                "physicalLocation": {
-                    "artifactLocation": {
-                        "uri": str(Path(finding.file).as_posix()),
+            "locations": [
+                {
+                    "physicalLocation": {
+                        "artifactLocation": {
+                            "uri": str(Path(finding.file).as_posix()),
+                        },
+                        "region": {"startLine": max(1, finding.line)},
                     },
-                    "region": {"startLine": max(1, finding.line)},
-                },
-            }],
+                }
+            ],
         }
         if finding.suggestion:
             sarif_result["message"]["text"] += f" -- {finding.suggestion}"
@@ -66,17 +69,19 @@ def build(result: CheckResult) -> dict:
     return {
         "$schema": SARIF_SCHEMA,
         "version": "2.1.0",
-        "runs": [{
-            "tool": {
-                "driver": {
-                    "name": "sql-guard",
-                    "version": __version__,
-                    "informationUri": INFO_URI,
-                    "rules": _all_rule_descriptors(),
+        "runs": [
+            {
+                "tool": {
+                    "driver": {
+                        "name": "sql-guard",
+                        "version": __version__,
+                        "informationUri": INFO_URI,
+                        "rules": _all_rule_descriptors(),
+                    },
                 },
-            },
-            "results": results,
-        }],
+                "results": results,
+            }
+        ],
     }
 
 

sql_guard/rules/contracts.py

@@ -84,8 +84,7 @@ def check_statement(self, statement: str, start_line: int, file: str) -> Finding
                         f"for table '{table_name}'"
                     ),
                     suggestion=(
-                        f"Either add '{ref_col}' to the contract or correct "
-                        f"the column name"
+                        f"Either add '{ref_col}' to the contract or correct the column name"
                     ),
                 )
         return None
@@ -162,9 +161,7 @@ def check_statement(self, statement: str, start_line: int, file: str) -> Finding
         if table is None:
             return None
 
-        listed_cols = {
-            c.strip().strip("[]`\"").lower() for c in m.group(2).split(",")
-        }
+        listed_cols = {c.strip().strip('[]`"').lower() for c in m.group(2).split(",")}
         missing = [c for c in table.required_columns if c not in listed_cols]
 
         if missing:
@@ -216,9 +213,7 @@ def check_statement(self, statement: str, start_line: int, file: str) -> Finding
         if table is None or not table.primary_keys:
             return None
 
-        listed_cols = {
-            c.strip().strip("[]`\"").lower() for c in m.group(2).split(",")
-        }
+        listed_cols = {c.strip().strip('[]`"').lower() for c in m.group(2).split(",")}
         missing_pks = [
             pk
             for pk in table.primary_keys
@@ -274,8 +269,11 @@ class UnmappedForeignKey(ContractRule):
     )
 
     def _fk_resolves(
-        self, source_table_name: str, source_col: str,
-        target_table_name: str, target_col: str,
+        self,
+        source_table_name: str,
+        source_col: str,
+        target_table_name: str,
+        target_col: str,
     ) -> bool:
         if self.contract is None:
             return False
@@ -287,8 +285,7 @@ def _fk_resolves(
             return False
         ref_table, _, ref_col = col.foreign_key.partition(".")
         return (
-            ref_table.lower() == target_table_name.lower()
-            and ref_col.lower() == target_col.lower()
+            ref_table.lower() == target_table_name.lower() and ref_col.lower() == target_col.lower()
         )
 
     def check_statement(self, statement: str, start_line: int, file: str) -> Finding | None:

sql_guard/rules/python_rules.py

@@ -148,9 +148,7 @@ def check(self, hit: ExtractedSql, file: str) -> Finding | None:
                 file=file,
                 line=hit.line,
                 message="f-string passed to sqlalchemy.text() -- SQL injection risk",
-                suggestion=(
-                    "Use bound parameters: text(\"... WHERE id = :id\"), {\"id\": user_id}"
-                ),
+                suggestion=('Use bound parameters: text("... WHERE id = :id"), {"id": user_id}'),
             )
         return None