diff --git a/.github/workflows/main-ci.yml b/.github/workflows/main-ci.yml index 0dad5915d00..8fc0d6f3f63 100644 --- a/.github/workflows/main-ci.yml +++ b/.github/workflows/main-ci.yml @@ -53,8 +53,12 @@ jobs: filters: | api: - 'api/**' + - 'scripts/ast_grep_guard.py' - 'scripts/check_no_new_getattr.py' + - 'scripts/check_no_new_controller_sqlalchemy.py' + - 'scripts/lint_controller_sqlalchemy.py' - 'scripts/ast_grep_rules/no_new_getattr.yml' + - 'scripts/ast_grep_rules/no_new_controller_sqlalchemy.yml' - '.github/workflows/style.yml' - '.github/workflows/main-ci.yml' - '.github/workflows/api-tests.yml' diff --git a/.github/workflows/style.yml b/.github/workflows/style.yml index eecd7f83e5e..1b3334dd274 100644 --- a/.github/workflows/style.yml +++ b/.github/workflows/style.yml @@ -34,8 +34,12 @@ jobs: with: files: | api/** + scripts/ast_grep_guard.py scripts/check_no_new_getattr.py + scripts/check_no_new_controller_sqlalchemy.py + scripts/lint_controller_sqlalchemy.py scripts/ast_grep_rules/no_new_getattr.yml + scripts/ast_grep_rules/no_new_controller_sqlalchemy.yml .github/workflows/style.yml .github/workflows/main-ci.yml @@ -63,6 +67,10 @@ jobs: if: steps.changed-files.outputs.any_changed == 'true' run: uv run --project api python scripts/check_no_new_getattr.py --base-rev "${{ inputs.base-rev }}" + - name: Run No New Controller SQLAlchemy Guard + if: steps.changed-files.outputs.any_changed == 'true' + run: uv run --project api python scripts/check_no_new_controller_sqlalchemy.py --base-rev "${{ inputs.base-rev }}" + - name: Run Type Checks if: steps.changed-files.outputs.any_changed == 'true' env: diff --git a/api/tests/unit_tests/commands/test_check_no_new_getattr.py b/api/tests/unit_tests/commands/test_check_no_new_getattr.py index 72e72a631aa..4efcb37da92 100644 --- a/api/tests/unit_tests/commands/test_check_no_new_getattr.py +++ b/api/tests/unit_tests/commands/test_check_no_new_getattr.py @@ -13,11 +13,15 @@ from pathlib import Path import pytest REPO_ROOT = Path(__file__).resolve().parents[4] -SCRIPT_PATH = REPO_ROOT / "scripts" / "check_no_new_getattr.py" +SCRIPTS_DIR = REPO_ROOT / "scripts" +SCRIPT_PATH = SCRIPTS_DIR / "check_no_new_getattr.py" +GUARD_HELPER_PATH = SCRIPTS_DIR / "ast_grep_guard.py" def load_guard_module() -> types.ModuleType: - spec = importlib.util.spec_from_file_location("check_no_new_getattr_under_test", SCRIPT_PATH) + if str(SCRIPTS_DIR) not in sys.path: + sys.path.insert(0, str(SCRIPTS_DIR)) + spec = importlib.util.spec_from_file_location("ast_grep_guard_under_test", GUARD_HELPER_PATH) assert spec is not None assert spec.loader is not None module = importlib.util.module_from_spec(spec) diff --git a/scripts/ast_grep_guard.py b/scripts/ast_grep_guard.py new file mode 100644 index 00000000000..3e46486f24e --- /dev/null +++ b/scripts/ast_grep_guard.py @@ -0,0 +1,343 @@ +#!/usr/bin/env python3 +"""Shared helpers for baseline-aware ast-grep CI guards.""" + +from __future__ import annotations + +import argparse +import json +import re +import shutil +import subprocess +import sys +import tempfile +from collections.abc import Callable +from dataclasses import dataclass +from pathlib import Path +from typing import Any + + +REPO_ROOT = Path(__file__).resolve().parents[1] +SCAN_ROOT = Path.cwd() +RULES_ROOT = REPO_ROOT / "scripts" / "ast_grep_rules" +HUNK_PATTERN = re.compile( + r"^@@ -(?P\d+)(?:,(?P\d+))? \+(?P\d+)(?:,(?P\d+))? @@" +) + +PathPredicate = Callable[[str], bool] +MatchPredicate = Callable[["Match"], bool] + + +@dataclass(frozen=True) +class Hunk: + old_start: int + old_count: int + new_start: int + new_count: int + + def contains_old(self, line_number: int) -> bool: + return self._contains(line_number, self.old_start, self.old_count) + + def contains_new(self, line_number: int) -> bool: + return self._contains(line_number, self.new_start, self.new_count) + + @staticmethod + def _contains(line_number: int, start: int, count: int) -> bool: + if count == 0: + return False + return start <= line_number < start + count + + +@dataclass(frozen=True) +class Match: + line_number: int + source_line: str + text: str + meta_variables: dict[str, str] + + +@dataclass(frozen=True) +class Violation: + path: str + line_number: int + message: str + + +def rule_path(filename: str) -> Path: + return RULES_ROOT / filename + + +def parse_args(description: str | None) -> argparse.Namespace: + parser = argparse.ArgumentParser(description=description) + diff_source_group = parser.add_mutually_exclusive_group(required=True) + diff_source_group.add_argument( + "--staged", + action="store_true", + help="Inspect staged changes against HEAD, for pre-commit style usage.", + ) + diff_source_group.add_argument( + "--base-rev", + help="Inspect changes between the provided git revision and HEAD.", + ) + return parser.parse_args() + + +def resolve_ast_grep_command() -> list[str]: + if shutil.which("ast-grep"): + return ["ast-grep"] + if shutil.which("uvx"): + return ["uvx", "--from", "ast-grep-cli", "ast-grep"] + raise RuntimeError("ast-grep executable not found") + + +def git_output(*args: str, allow_missing: bool = False) -> str: + completed = subprocess.run( + ["git", *args], + cwd=SCAN_ROOT, + text=True, + capture_output=True, + check=False, + ) + if completed.returncode == 0: + return completed.stdout + if allow_missing: + return "" + raise RuntimeError(completed.stderr.strip() or completed.stdout.strip() or "git command failed") + + +def collect_diff_text(args: argparse.Namespace) -> str: + if args.staged: + return git_output("diff", "--cached", "--unified=0", "--diff-filter=AM", "--no-ext-diff") + return git_output( + "diff", + "--unified=0", + "--diff-filter=AM", + "--no-ext-diff", + f"{args.base_rev}..HEAD", + ) + + +def parse_changed_hunks(diff_text: str) -> dict[str, list[Hunk]]: + changed_hunks: dict[str, list[Hunk]] = {} + current_path: str | None = None + + for line in diff_text.splitlines(): + if line.startswith("diff --git "): + parts = line.split() + current_path = parts[3][2:] + changed_hunks.setdefault(current_path, []) + continue + + if current_path is None: + continue + + hunk_match = HUNK_PATTERN.match(line) + if not hunk_match: + continue + + changed_hunks[current_path].append( + Hunk( + old_start=int(hunk_match.group("old_start")), + old_count=parse_count(hunk_match.group("old_count")), + new_start=int(hunk_match.group("new_start")), + new_count=parse_count(hunk_match.group("new_count")), + ) + ) + + return {path: hunks for path, hunks in changed_hunks.items() if hunks} + + +def parse_count(value: str | None) -> int: + if value is None: + return 1 + return int(value) + + +def is_python_source_path(path: str) -> bool: + return Path(path).suffix in {".py", ".pyi"} + + +def load_file_versions(path: str, args: argparse.Namespace) -> tuple[str, str]: + if args.staged: + return ( + git_output("show", f"HEAD:{path}", allow_missing=True), + git_output("show", f":{path}"), + ) + + return ( + git_output("show", f"{args.base_rev}:{path}", allow_missing=True), + git_output("show", f"HEAD:{path}"), + ) + + +def run_ast_grep(source: str, *, rule: Path) -> list[Match]: + if not source.strip(): + return [] + + with tempfile.NamedTemporaryFile(mode="w", suffix=".py", encoding="utf-8") as temp_file: + temp_file.write(source) + temp_file.flush() + completed = subprocess.run( + [ + *resolve_ast_grep_command(), + "scan", + "--rule", + str(rule), + "--json=compact", + temp_file.name, + ], + cwd=REPO_ROOT, + text=True, + capture_output=True, + check=False, + ) + + if completed.returncode not in (0, 1): + raise RuntimeError(completed.stderr.strip() or completed.stdout.strip() or "ast-grep command failed") + + if not completed.stdout.strip(): + return [] + + lines = source.splitlines() + raw_matches = json.loads(completed.stdout) + matches: list[Match] = [] + for raw_match in raw_matches: + line_number = int(raw_match["range"]["start"]["line"]) + 1 + source_line = lines[line_number - 1] if line_number - 1 < len(lines) else "" + matches.append( + Match( + line_number=line_number, + source_line=source_line, + text=extract_match_text(raw_match, fallback=source_line), + meta_variables=extract_meta_variables(raw_match), + ) + ) + return matches + + +def extract_match_text(raw_match: dict[str, Any], *, fallback: str) -> str: + text = raw_match.get("text") + if isinstance(text, str) and text: + return text + + lines = raw_match.get("lines") + if isinstance(lines, str) and lines: + return lines + + return fallback + + +def extract_meta_variables(raw_match: dict[str, Any]) -> dict[str, str]: + meta_variables = raw_match.get("metaVariables") + if not isinstance(meta_variables, dict): + return {} + + single_variables = meta_variables.get("single") + if not isinstance(single_variables, dict): + return {} + + result: dict[str, str] = {} + for name, value in single_variables.items(): + if not isinstance(name, str) or not isinstance(value, dict): + continue + text = value.get("text") + if isinstance(text, str): + result[name] = text + return result + + +def has_reasoned_noqa(source_line: str, rule_id: str) -> bool: + pattern = re.compile(rf"# noqa: {re.escape(rule_id)}(?:\s+(?P\S.*))?\s*$") + match = pattern.search(source_line) + if not match: + return False + reason = match.group("reason") + return reason is not None and bool(reason.strip()) + + +def collect_hunk_violations( + path: str, + old_matches: list[Match], + new_matches: list[Match], + hunk: Hunk, + *, + is_reportable_match: MatchPredicate, + violation_message: str, +) -> list[Violation]: + old_in_hunk = [ + match for match in old_matches if hunk.contains_old(match.line_number) and is_reportable_match(match) + ] + new_in_hunk = [ + match for match in new_matches if hunk.contains_new(match.line_number) and is_reportable_match(match) + ] + surplus = len(new_in_hunk) - len(old_in_hunk) + if surplus <= 0: + return [] + + return [ + Violation(path=path, line_number=match.line_number, message=violation_message) + for match in new_in_hunk[-surplus:] + ] + + +def find_net_new_violations( + changed_hunks: dict[str, list[Hunk]], + args: argparse.Namespace, + *, + rule: Path, + is_scanned_path: PathPredicate, + is_reportable_match: MatchPredicate, + violation_message: str, +) -> list[Violation]: + violations: list[Violation] = [] + for path, hunks in changed_hunks.items(): + if not is_scanned_path(path): + continue + old_source, new_source = load_file_versions(path, args) + old_matches = run_ast_grep(old_source, rule=rule) + new_matches = run_ast_grep(new_source, rule=rule) + for hunk in hunks: + violations.extend( + collect_hunk_violations( + path, + old_matches, + new_matches, + hunk, + is_reportable_match=is_reportable_match, + violation_message=violation_message, + ) + ) + return sorted(violations, key=lambda item: (item.path, item.line_number)) + + +def print_violations(violations: list[Violation]) -> None: + for violation in violations: + print(f"{violation.path}:{violation.line_number}: {violation.message}", file=sys.stderr) + + +def run_guard( + *, + description: str | None, + rule: Path, + is_scanned_path: PathPredicate, + is_reportable_match: MatchPredicate, + violation_message: str, +) -> int: + try: + args = parse_args(description) + changed_hunks = parse_changed_hunks(collect_diff_text(args)) + violations = find_net_new_violations( + changed_hunks, + args, + rule=rule, + is_scanned_path=is_scanned_path, + is_reportable_match=is_reportable_match, + violation_message=violation_message, + ) + except RuntimeError as exc: + print(str(exc), file=sys.stderr) + return 2 + + if violations: + print_violations(violations) + return 1 + return 0 diff --git a/scripts/ast_grep_rules/no_new_controller_sqlalchemy.yml b/scripts/ast_grep_rules/no_new_controller_sqlalchemy.yml new file mode 100644 index 00000000000..742914c6db1 --- /dev/null +++ b/scripts/ast_grep_rules/no_new_controller_sqlalchemy.yml @@ -0,0 +1,54 @@ +id: no-new-controller-sqlalchemy +language: python +rule: + any: + - pattern: db.session.add($$$ARGS) + - pattern: db.session.add_all($$$ARGS) + - pattern: db.session.begin($$$ARGS) + - pattern: db.session.commit($$$ARGS) + - pattern: db.session.delete($$$ARGS) + - pattern: db.session.execute($$$ARGS) + - pattern: db.session.flush($$$ARGS) + - pattern: db.session.get($MODEL, $$$ARGS) + - pattern: db.session.merge($$$ARGS) + - pattern: db.session.refresh($$$ARGS) + - pattern: db.session.rollback($$$ARGS) + - pattern: db.session.scalar($$$ARGS) + - pattern: db.session.scalars($$$ARGS) + - pattern: session.add($$$ARGS) + - pattern: session.add_all($$$ARGS) + - pattern: session.begin($$$ARGS) + - pattern: session.commit($$$ARGS) + - pattern: session.delete($$$ARGS) + - pattern: session.execute($$$ARGS) + - pattern: session.flush($$$ARGS) + - pattern: session.get($MODEL, $$$ARGS) + - pattern: session.merge($$$ARGS) + - pattern: session.refresh($$$ARGS) + - pattern: session.rollback($$$ARGS) + - pattern: session.scalar($$$ARGS) + - pattern: session.scalars($$$ARGS) + - pattern: Session($$$ARGS) + - pattern: sessionmaker($$$ARGS) + - pattern: select($$$ARGS) + - pattern: insert($$$ARGS) + - pattern: update($$$ARGS) + - pattern: delete($$$ARGS) + - pattern: text($$$ARGS) + - pattern: sa.select($$$ARGS) + - pattern: sa.insert($$$ARGS) + - pattern: sa.update($$$ARGS) + - pattern: sa.delete($$$ARGS) + - pattern: sa.text($$$ARGS) + - pattern: sqlalchemy.select($$$ARGS) + - pattern: sqlalchemy.insert($$$ARGS) + - pattern: sqlalchemy.update($$$ARGS) + - pattern: sqlalchemy.delete($$$ARGS) + - pattern: sqlalchemy.text($$$ARGS) + - pattern: db.select($$$ARGS) + - pattern: db.insert($$$ARGS) + - pattern: db.update($$$ARGS) + - pattern: db.delete($$$ARGS) + - pattern: db.text($$$ARGS) +message: net-new direct SQLAlchemy call in controller code +severity: error diff --git a/scripts/check_no_new_controller_sqlalchemy.py b/scripts/check_no_new_controller_sqlalchemy.py new file mode 100644 index 00000000000..bf1bea336b7 --- /dev/null +++ b/scripts/check_no_new_controller_sqlalchemy.py @@ -0,0 +1,61 @@ +#!/usr/bin/env python3 +"""Block net-new direct SQLAlchemy calls in controller code. + +Allowing flush() and commit() to allow controllers to control +the transaction border. +""" + +from __future__ import annotations + +import re +from pathlib import Path + +from ast_grep_guard import Match, has_reasoned_noqa, rule_path, run_guard + + +RULE_ID = "no-new-controller-sqlalchemy" +RULE_PATH = rule_path("no_new_controller_sqlalchemy.yml") +CONTROLLER_ROOT = Path("api/controllers") +FLASK_SESSION_GET_PATTERN = re.compile(r"^session\.get\(\s*['\"]") +ALLOWED_SESSION_METHODS = frozenset({"commit", "flush"}) +VIOLATION_MESSAGE = "no-new-controller-sqlalchemy net-new direct SQLAlchemy call in controller code" + + +def is_controller_python_source_path(path: str) -> bool: + source_path = Path(path) + return source_path.suffix in {".py", ".pyi"} and source_path.is_relative_to(CONTROLLER_ROOT) + + +def is_allowed_session_boundary(match: Match) -> bool: + method = match.meta_variables.get("METHOD") or match.meta_variables.get("SESSION_METHOD") + if method in ALLOWED_SESSION_METHODS: + return True + + stripped_text = match.text.strip() + return stripped_text.startswith(("db.session.commit(", "db.session.flush(", "session.commit(", "session.flush(")) + + +def is_flask_session_get(match: Match) -> bool: + return bool(FLASK_SESSION_GET_PATTERN.match(match.text.strip())) + + +def is_suppressed(match: Match) -> bool: + return has_reasoned_noqa(match.source_line, RULE_ID) + + +def is_reportable_match(match: Match) -> bool: + return not is_allowed_session_boundary(match) and not is_flask_session_get(match) and not is_suppressed(match) + + +def main() -> int: + return run_guard( + description=__doc__, + rule=RULE_PATH, + is_scanned_path=is_controller_python_source_path, + is_reportable_match=is_reportable_match, + violation_message=VIOLATION_MESSAGE, + ) + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/scripts/check_no_new_getattr.py b/scripts/check_no_new_getattr.py index 8bf437c5367..e788395c62c 100644 --- a/scripts/check_no_new_getattr.py +++ b/scripts/check_no_new_getattr.py @@ -3,258 +3,26 @@ from __future__ import annotations -import argparse -import json -import re -import shutil -import subprocess -import sys -import tempfile -from dataclasses import dataclass -from pathlib import Path +from ast_grep_guard import Match, has_reasoned_noqa, is_python_source_path, rule_path, run_guard -REPO_ROOT = Path(__file__).resolve().parents[1] -SCAN_ROOT = Path.cwd() -RULE_PATH = REPO_ROOT / "scripts" / "ast_grep_rules" / "no_new_getattr.yml" -HUNK_PATTERN = re.compile( - r"^@@ -(?P\d+)(?:,(?P\d+))? \+(?P\d+)(?:,(?P\d+))? @@" -) -SUPPRESSION_PATTERN = re.compile(r"# noqa: no-new-getattr(?:\s+(?P\S.*))?\s*$") +RULE_ID = "no-new-getattr" +RULE_PATH = rule_path("no_new_getattr.yml") +VIOLATION_MESSAGE = "no-new-getattr net-new getattr() in added code" -@dataclass(frozen=True) -class Hunk: - old_start: int - old_count: int - new_start: int - new_count: int - - def contains_old(self, line_number: int) -> bool: - return self._contains(line_number, self.old_start, self.old_count) - - def contains_new(self, line_number: int) -> bool: - return self._contains(line_number, self.new_start, self.new_count) - - @staticmethod - def _contains(line_number: int, start: int, count: int) -> bool: - if count == 0: - return False - return start <= line_number < start + count - - -@dataclass(frozen=True) -class Match: - line_number: int - source_line: str - - -@dataclass(frozen=True) -class Violation: - path: str - line_number: int - message: str - - -def parse_args() -> argparse.Namespace: - parser = argparse.ArgumentParser(description=__doc__) - diff_source_group = parser.add_mutually_exclusive_group(required=True) - diff_source_group.add_argument( - "--staged", - action="store_true", - help="Inspect staged changes against HEAD, for pre-commit style usage.", - ) - diff_source_group.add_argument( - "--base-rev", - help="Inspect changes between the provided git revision and HEAD.", - ) - return parser.parse_args() - - -def resolve_ast_grep_command() -> list[str]: - if shutil.which("ast-grep"): - return ["ast-grep"] - if shutil.which("uvx"): - return ["uvx", "--from", "ast-grep-cli", "ast-grep"] - raise RuntimeError("ast-grep executable not found") - - -def git_output(*args: str, allow_missing: bool = False) -> str: - completed = subprocess.run( - ["git", *args], - cwd=SCAN_ROOT, - text=True, - capture_output=True, - check=False, - ) - if completed.returncode == 0: - return completed.stdout - if allow_missing: - return "" - raise RuntimeError(completed.stderr.strip() or completed.stdout.strip() or "git command failed") - - -def collect_diff_text(args: argparse.Namespace) -> str: - if args.staged: - return git_output("diff", "--cached", "--unified=0", "--diff-filter=AM", "--no-ext-diff") - return git_output( - "diff", - "--unified=0", - "--diff-filter=AM", - "--no-ext-diff", - f"{args.base_rev}..HEAD", - ) - - -def parse_changed_hunks(diff_text: str) -> dict[str, list[Hunk]]: - changed_hunks: dict[str, list[Hunk]] = {} - current_path: str | None = None - - for line in diff_text.splitlines(): - if line.startswith("diff --git "): - parts = line.split() - current_path = parts[3][2:] - changed_hunks.setdefault(current_path, []) - continue - - if current_path is None: - continue - - hunk_match = HUNK_PATTERN.match(line) - if not hunk_match: - continue - - changed_hunks[current_path].append( - Hunk( - old_start=int(hunk_match.group("old_start")), - old_count=parse_count(hunk_match.group("old_count")), - new_start=int(hunk_match.group("new_start")), - new_count=parse_count(hunk_match.group("new_count")), - ) - ) - - return {path: hunks for path, hunks in changed_hunks.items() if hunks} - - -def parse_count(value: str | None) -> int: - if value is None: - return 1 - return int(value) - - -def is_python_source_path(path: str) -> bool: - return Path(path).suffix in {".py", ".pyi"} - - -def load_file_versions(path: str, args: argparse.Namespace) -> tuple[str, str]: - if args.staged: - return ( - git_output("show", f"HEAD:{path}", allow_missing=True), - git_output("show", f":{path}"), - ) - - return ( - git_output("show", f"{args.base_rev}:{path}", allow_missing=True), - git_output("show", f"HEAD:{path}"), - ) - - -def run_ast_grep(source: str) -> list[Match]: - if not source.strip(): - return [] - - with tempfile.NamedTemporaryFile(mode="w", suffix=".py", encoding="utf-8") as temp_file: - temp_file.write(source) - temp_file.flush() - completed = subprocess.run( - [ - *resolve_ast_grep_command(), - "scan", - "--rule", - str(RULE_PATH), - "--json=compact", - temp_file.name, - ], - cwd=REPO_ROOT, - text=True, - capture_output=True, - check=False, - ) - - if completed.returncode not in (0, 1): - raise RuntimeError(completed.stderr.strip() or completed.stdout.strip() or "ast-grep command failed") - - if not completed.stdout.strip(): - return [] - - lines = source.splitlines() - raw_matches = json.loads(completed.stdout) - matches: list[Match] = [] - for raw_match in raw_matches: - line_number = int(raw_match["range"]["start"]["line"]) + 1 - source_line = lines[line_number - 1] if line_number - 1 < len(lines) else "" - matches.append(Match(line_number=line_number, source_line=source_line)) - return matches - - -def is_suppressed(source_line: str) -> bool: - match = SUPPRESSION_PATTERN.search(source_line) - if not match: - return False - reason = match.group("reason") - return reason is not None and bool(reason.strip()) - - -def collect_hunk_violations( - path: str, old_matches: list[Match], new_matches: list[Match], hunk: Hunk -) -> list[Violation]: - old_in_hunk = [ - match for match in old_matches if hunk.contains_old(match.line_number) and not is_suppressed(match.source_line) - ] - new_in_hunk = [ - match for match in new_matches if hunk.contains_new(match.line_number) and not is_suppressed(match.source_line) - ] - surplus = len(new_in_hunk) - len(old_in_hunk) - if surplus <= 0: - return [] - - return [ - Violation(path=path, line_number=match.line_number, message="no-new-getattr net-new getattr() in added code") - for match in new_in_hunk[-surplus:] - ] - - -def find_net_new_getattr_violations(changed_hunks: dict[str, list[Hunk]], args: argparse.Namespace) -> list[Violation]: - violations: list[Violation] = [] - for path, hunks in changed_hunks.items(): - if not is_python_source_path(path): - continue - old_source, new_source = load_file_versions(path, args) - old_matches = run_ast_grep(old_source) - new_matches = run_ast_grep(new_source) - for hunk in hunks: - violations.extend(collect_hunk_violations(path, old_matches, new_matches, hunk)) - return sorted(violations, key=lambda item: (item.path, item.line_number)) - - -def print_violations(violations: list[Violation]) -> None: - for violation in violations: - print(f"{violation.path}:{violation.line_number}: {violation.message}", file=sys.stderr) +def is_reportable_match(match: Match) -> bool: + return not has_reasoned_noqa(match.source_line, RULE_ID) def main() -> int: - try: - args = parse_args() - changed_hunks = parse_changed_hunks(collect_diff_text(args)) - violations = find_net_new_getattr_violations(changed_hunks, args) - except RuntimeError as exc: - print(str(exc), file=sys.stderr) - return 2 - - if violations: - print_violations(violations) - return 1 - return 0 + return run_guard( + description=__doc__, + rule=RULE_PATH, + is_scanned_path=is_python_source_path, + is_reportable_match=is_reportable_match, + violation_message=VIOLATION_MESSAGE, + ) if __name__ == "__main__": diff --git a/scripts/lint_controller_sqlalchemy.py b/scripts/lint_controller_sqlalchemy.py new file mode 100644 index 00000000000..87c9582b17f --- /dev/null +++ b/scripts/lint_controller_sqlalchemy.py @@ -0,0 +1,223 @@ +#!/usr/bin/env python3 +"""Report direct SQLAlchemy usage in controller code. + +This is the full-inventory companion to ``check_no_new_controller_sqlalchemy.py``. +It scans the current tree instead of a git diff, so migration work can inspect +existing controller usage without turning historical debt into a CI failure. +""" + +from __future__ import annotations + +import argparse +import json +import re +import sys +from collections import Counter +from collections.abc import Iterable, Sequence +from dataclasses import asdict, dataclass +from pathlib import Path + +import check_no_new_controller_sqlalchemy as controller_sqlalchemy_guard +from ast_grep_guard import Match, REPO_ROOT, run_ast_grep + + +DEFAULT_CONTROLLER_DIR = REPO_ROOT / "api" / "controllers" +MAX_TEXT_LENGTH = 180 +SESSION_METHOD_PATTERN = re.compile(r"^(?:db\.session|session)\.(?P[A-Za-z_][A-Za-z0-9_]*)\(") + + +@dataclass(frozen=True) +class Finding: + classification: str + category: str + file: str + line: int + text: str + source_line: str + + +def parse_args() -> argparse.Namespace: + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument( + "paths", + nargs="*", + help="Files or directories to scan. Defaults to api/controllers.", + ) + parser.add_argument("--include-allowed", action="store_true", help="Print allowed flush()/commit() findings.") + parser.add_argument("--include-suppressed", action="store_true", help="Print reasoned noqa suppressions.") + parser.add_argument("--summary-only", action="store_true", help="Print counts without per-finding details.") + parser.add_argument("--json", action="store_true", help="Emit machine-readable JSON.") + parser.add_argument( + "--fail-on-findings", + action="store_true", + help="Treat reportable direct SQLAlchemy usage as a failure. By default this scanner is report-only.", + ) + return parser.parse_args() + + +def iter_python_files(paths: Iterable[Path]) -> Iterable[Path]: + for path in paths: + if path.is_file() and path.suffix in {".py", ".pyi"}: + yield path + elif path.is_dir(): + yield from sorted(child for child in path.rglob("*.py") if child.is_file()) + + +def display_path(path: Path) -> str: + try: + return str(path.relative_to(REPO_ROOT)) + except ValueError: + return str(path) + + +def compact_text(text: str) -> str: + compact = " ".join(text.split()) + if len(compact) <= MAX_TEXT_LENGTH: + return compact + return f"{compact[: MAX_TEXT_LENGTH - 3]}..." + + +def category_for_match(match: Match) -> str: + text = match.text.strip() + method_match = SESSION_METHOD_PATTERN.match(text) + method = method_match.group("method") if method_match else None + + if text.startswith("db.session."): + return f"scoped-session.{method or 'call'}" + if text.startswith("session."): + return f"explicit-session.{method or 'call'}" + if text.startswith(("Session(", "sessionmaker(")): + return "session-factory" + if text.startswith(("sa.", "sqlalchemy.", "db.")): + return "sql-expression-qualified" + return "sql-expression" + + +def classification_for_match(match: Match) -> str: + if controller_sqlalchemy_guard.is_allowed_session_boundary(match): + return "allowed" + if controller_sqlalchemy_guard.is_suppressed(match): + return "suppressed" + if controller_sqlalchemy_guard.is_flask_session_get(match): + return "ignored" + return "reportable" + + +def findings_for_file(path: Path) -> list[Finding]: + source = path.read_text(encoding="utf-8") + findings: list[Finding] = [] + for match in run_ast_grep(source, rule=controller_sqlalchemy_guard.RULE_PATH): + findings.append( + Finding( + classification=classification_for_match(match), + category=category_for_match(match), + file=display_path(path), + line=match.line_number, + text=compact_text(match.text), + source_line=match.source_line.strip(), + ) + ) + return findings + + +def print_text_report( + findings: Sequence[Finding], + *, + include_allowed: bool, + include_suppressed: bool, + summary_only: bool, +) -> None: + counts = Counter(finding.classification for finding in findings) + reportable = [finding for finding in findings if finding.classification == "reportable"] + category_counts = Counter(finding.category for finding in reportable) + file_counts = Counter(finding.file for finding in reportable) + sys.stdout.write( + "Controller SQLAlchemy scan: " + f"{counts['reportable']} reportable, " + f"{counts['allowed']} allowed, " + f"{counts['suppressed']} suppressed, " + f"{counts['ignored']} ignored\n" + ) + + if category_counts: + sys.stdout.write("\nREPORTABLE BY CATEGORY:\n") + for category, count in sorted(category_counts.items()): + sys.stdout.write(f"- {category}: {count}\n") + + if file_counts: + sys.stdout.write("\nREPORTABLE BY FILE:\n") + for file, count in sorted(file_counts.items()): + sys.stdout.write(f"- {file}: {count}\n") + + if summary_only: + return + + print_findings("REPORTABLE", reportable) + if include_allowed: + print_findings("ALLOWED", [finding for finding in findings if finding.classification == "allowed"]) + if include_suppressed: + print_findings("SUPPRESSED", [finding for finding in findings if finding.classification == "suppressed"]) + + +def print_findings(title: str, findings: Sequence[Finding]) -> None: + if not findings: + return + + sys.stdout.write(f"\n{title}:\n") + for finding in findings: + sys.stdout.write(f"- {finding.file}:{finding.line} [{finding.category}] {finding.text}\n") + + +def jsonable_findings( + findings: Sequence[Finding], + *, + include_allowed: bool, + include_suppressed: bool, + summary_only: bool, +) -> dict[str, object]: + visible = [ + finding + for finding in findings + if finding.classification == "reportable" + or (include_allowed and finding.classification == "allowed") + or (include_suppressed and finding.classification == "suppressed") + ] + counts = Counter(finding.classification for finding in findings) + return { + "summary": dict(sorted(counts.items())), + "findings": [] if summary_only else [asdict(finding) for finding in visible], + } + + +def main() -> int: + args = parse_args() + raw_paths = args.paths or [str(DEFAULT_CONTROLLER_DIR)] + paths = [path if path.is_absolute() else REPO_ROOT / path for path in map(Path, raw_paths)] + + findings: list[Finding] = [] + for path in iter_python_files(paths): + findings.extend(findings_for_file(path.resolve())) + findings.sort(key=lambda finding: (finding.file, finding.line, finding.category, finding.text)) + + if args.json: + payload = jsonable_findings( + findings, + include_allowed=bool(args.include_allowed), + include_suppressed=bool(args.include_suppressed), + summary_only=bool(args.summary_only), + ) + sys.stdout.write(f"{json.dumps(payload, indent=2, sort_keys=True)}\n") + else: + print_text_report( + findings, + include_allowed=bool(args.include_allowed), + include_suppressed=bool(args.include_suppressed), + summary_only=bool(args.summary_only), + ) + + has_reportable = any(finding.classification == "reportable" for finding in findings) + return int(bool(args.fail_on_findings) and has_reportable) + + +if __name__ == "__main__": + raise SystemExit(main())