From 954d02288040b83d1ea952aad70b857005bcfb47 Mon Sep 17 00:00:00 2001 From: Adrian Ehrsam Date: Wed, 30 Sep 2026 14:39:43 +0200 Subject: [PATCH 1/3] wip: allow simple inline insert/update/delete in bdt lint From b058e2422ad1f90adec6ced37fc3bb0a2811ccc1 Mon Sep 17 00:00:00 2001 From: Adrian Ehrsam Date: Wed, 30 Sep 2026 14:40:35 +0200 Subject: [PATCH 2/3] feat(lint): allow simple inline insert/update/delete Co-Authored-By: Claude Sonnet 5.5 Claude-Session: https://claude.ai/code/session_017Ug7e9Gfy5oanqX1WWUkmt --- README.md | 2 ++ bmsdna/devtools/lint_sql.py | 16 +++++++++--- tests/test_lint_sql.py | 49 +++++++++++++++++++++++++++++++++++++ uv.lock | 2 +- 4 files changed, 65 insertions(+), 4 deletions(-) diff --git a/README.md b/README.md index 8f86b67..345f694 100644 --- a/README.md +++ b/README.md @@ -535,6 +535,8 @@ it can't see through): - **`sql-inline-too-complex`** — more than a trivial (≤4 line) query, or a JOIN/CTE/subquery/aggregation, inline instead of `load_sql()`/a `.sql` file. + Simple `INSERT`/`UPDATE`/`DELETE` are exempt from the line limit; `INSERT ... SELECT`, + `UPDATE ... FROM` and `DELETE ... USING` still count as complex. - **`sql-fstring-injection`** / **`sql-concat-injection`** / **`sql-percent-format-injection`** / **`sql-format-injection`** — SQL built with an f-string, `+` concatenation, the `%` operator, or `str.format()` diff --git a/bmsdna/devtools/lint_sql.py b/bmsdna/devtools/lint_sql.py index 986a4ed..a06054d 100644 --- a/bmsdna/devtools/lint_sql.py +++ b/bmsdna/devtools/lint_sql.py @@ -287,13 +287,23 @@ def _literal_findings(text: str, path: Path, lineno: int) -> list[Finding]: findings: list[Finding] = [] line_count = len([line for line in text.splitlines() if line.strip()]) - is_complex = ( - line_count > 4 - or parsed.find(exp.Join) is not None + has_complex_construct = ( + parsed.find(exp.Join) is not None or parsed.find(exp.With) is not None or parsed.find(exp.Subquery) is not None or parsed.find(exp.AggFunc) is not None ) + if isinstance(parsed, (exp.Insert, exp.Update, exp.Delete)): + # A simple write may span many lines (long column lists) -- only its shape counts: + # INSERT ... SELECT, UPDATE ... FROM and DELETE ... USING are query logic, so they're complex. + is_complex = ( + has_complex_construct + or parsed.find(exp.Select) is not None + or parsed.find(exp.From) is not None + or bool(parsed.args.get("using")) + ) + else: + is_complex = line_count > 4 or has_complex_construct if is_complex: findings.append( Finding( diff --git a/tests/test_lint_sql.py b/tests/test_lint_sql.py index f817a20..e547e1f 100644 --- a/tests/test_lint_sql.py +++ b/tests/test_lint_sql.py @@ -294,3 +294,52 @@ def c(cur): path.write_text(source) assert sorted(f.rule for f in check_sql_file(path)) == ["sql-fstring-injection"] assert sorted(f.rule for f in check_sql_file(path, review=True)) == ["sql-fstring-injection", "sql-unverified-cast"] + + +def test_long_simple_insert_update_delete_allowed_inline(tmp_path: Path) -> None: + source = ''' +async def f(cur): + await cur.execute( + """ + insert into dim.customer ( + id, + name, + email, + active + ) values (%(id)s, %(name)s, %(email)s, true) + on conflict (id) do update set name = excluded.name + """, + {}, + ) + await cur.execute( + """ + update dim.customer + set name = %(name)s, + email = %(email)s, + active = false + where id = %(id)s + """, + {}, + ) + await cur.execute( + """ + delete from dim.customer + where id = %(id)s + and active = false + and email is null + """, + {}, + ) +''' + assert "sql-inline-too-complex" not in _rules(_findings(source, tmp_path / "a.py")) + + +def test_insert_select_and_update_from_still_too_complex(tmp_path: Path) -> None: + source = ''' +async def f(cur): + await cur.execute("insert into a (id) select id from b") + await cur.execute("update a set x = b.x from b where a.id = b.id") + await cur.execute("delete from a using b where a.id = b.id") +''' + findings = _findings(source, tmp_path / "a.py") + assert [f.rule for f in findings].count("sql-inline-too-complex") == 3 diff --git a/uv.lock b/uv.lock index c4c35b9..6c8ae7c 100644 --- a/uv.lock +++ b/uv.lock @@ -22,7 +22,7 @@ wheels = [ [[package]] name = "bmsdna-devtools" -version = "0.28.0" +version = "0.28.1" source = { editable = "." } dependencies = [ { name = "pgdevkit", extra = ["db"] }, From ec83b68138920dc17e40eeac355673c2a1430784 Mon Sep 17 00:00:00 2001 From: Adrian Ehrsam Date: Wed, 30 Sep 2026 14:48:09 +0200 Subject: [PATCH 3/3] feat(lint): flag complex literals in *_sql variables; bump to 0.29.0 Co-Authored-By: Claude Sonnet 5.5 Claude-Session: https://claude.ai/code/session_017Ug7e9Gfy5oanqX1WWUkmt --- README.md | 2 ++ bmsdna/devtools/lint_sql.py | 41 +++++++++++++++++++++++++++++++------ pyproject.toml | 2 +- tests/test_lint_sql.py | 24 ++++++++++++++++++++++ 4 files changed, 62 insertions(+), 7 deletions(-) diff --git a/README.md b/README.md index 345f694..9929e43 100644 --- a/README.md +++ b/README.md @@ -537,6 +537,8 @@ it can't see through): JOIN/CTE/subquery/aggregation, inline instead of `load_sql()`/a `.sql` file. Simple `INSERT`/`UPDATE`/`DELETE` are exempt from the line limit; `INSERT ... SELECT`, `UPDATE ... FROM` and `DELETE ... USING` still count as complex. + Also applied to any `*_sql`/`*_SQL` variable assigned a literal, even if it never reaches an + `.execute()` call in the same file (e.g. it's handed to a helper). - **`sql-fstring-injection`** / **`sql-concat-injection`** / **`sql-percent-format-injection`** / **`sql-format-injection`** — SQL built with an f-string, `+` concatenation, the `%` operator, or `str.format()` diff --git a/bmsdna/devtools/lint_sql.py b/bmsdna/devtools/lint_sql.py index a06054d..acf49f0 100644 --- a/bmsdna/devtools/lint_sql.py +++ b/bmsdna/devtools/lint_sql.py @@ -280,12 +280,8 @@ def _injection_finding(text: str, path: Path, lineno: int, rule: str, how: str) return [Finding(path, lineno, rule, f"SQL built with {how} -- injection risk. {fix}")] -def _literal_findings(text: str, path: Path, lineno: int) -> list[Finding]: - parsed = parse_sql_text(text) - if parsed is None: - return [] +def _complexity_findings(parsed: "exp.Expression", text: str, path: Path, lineno: int) -> list[Finding]: findings: list[Finding] = [] - line_count = len([line for line in text.splitlines() if line.strip()]) has_complex_construct = ( parsed.find(exp.Join) is not None @@ -314,6 +310,14 @@ def _literal_findings(text: str, path: Path, lineno: int) -> list[Finding]: "-- move it to its own .sql file and load it with load_sql()/SqlLoader.", ) ) + return findings + + +def _literal_findings(text: str, path: Path, lineno: int) -> list[Finding]: + parsed = parse_sql_text(text) + if parsed is None: + return [] + findings = _complexity_findings(parsed, text, path, lineno) if _POSITIONAL_PARAM_RE.search(text): findings.append( @@ -440,6 +444,12 @@ def _check_query_arg( return findings +def _is_sql_named(name: str) -> bool: + """`FOO_SQL` / `foo_sql` (or a bare `sql`/`SQL`): a variable whose name says it holds SQL text.""" + lowered = name.lower() + return lowered == "sql" or lowered.endswith("_sql") + + def _execute_query_arg(call: ast.Call) -> ast.expr | None: if not (isinstance(call.func, ast.Attribute) and call.func.attr in ("execute", "executemany")): return None @@ -480,12 +490,31 @@ def _visit_scoped(self, node: ast.AST) -> None: def visit_Assign(self, node: ast.Assign) -> None: if len(node.targets) == 1 and isinstance(node.targets[0], ast.Name): self._scopes[-1].setdefault(node.targets[0].id, []).append(node.value) + self._check_sql_named_literal(node.targets[0].id, node.value, node.lineno) self.generic_visit(node) + def visit_AnnAssign(self, node: ast.AnnAssign) -> None: + if isinstance(node.target, ast.Name) and node.value is not None: + self._scopes[-1].setdefault(node.target.id, []).append(node.value) + self._check_sql_named_literal(node.target.id, node.value, node.lineno) + self.generic_visit(node) + + def _check_sql_named_literal(self, name: str, value: ast.expr, lineno: int) -> None: + """A `*_sql`/`*_SQL` variable bound to a literal is inline SQL even if it never reaches an + `.execute()` call in this file (e.g. it's passed to a helper) -- apply the complexity rule.""" + if not (_is_sql_named(name) and isinstance(value, ast.Constant) and isinstance(value.value, str)): + return + parsed = parse_sql_text(value.value) + if parsed is not None: + self.findings.extend(_complexity_findings(parsed, value.value, self.path, lineno)) + def visit_Call(self, node: ast.Call) -> None: query_arg = _execute_query_arg(node) if query_arg is not None: - self.findings.extend(_check_query_arg(query_arg, self._lookup, self.path, node.lineno, self.trust, self.review)) + found = _check_query_arg(query_arg, self._lookup, self.path, node.lineno, self.trust, self.review) + if isinstance(query_arg, ast.Name) and _is_sql_named(query_arg.id): + found = [f for f in found if f.rule != "sql-inline-too-complex"] # already reported at the assignment + self.findings.extend(found) self.generic_visit(node) diff --git a/pyproject.toml b/pyproject.toml index 696eded..a54bacc 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -10,7 +10,7 @@ packages = ["bmsdna"] [project] name = "bmsdna-devtools" -version = "0.28.1" +version = "0.29.0" description = "Shared Azure DevOps / GitHub / git / Azure Monitor developer tooling for BMS projects" readme = "README.md" requires-python = ">=3.14" # pgdevkit>=0.7.1 requires 3.14; was >=3.11 before adding it as a dependency diff --git a/tests/test_lint_sql.py b/tests/test_lint_sql.py index e547e1f..f545076 100644 --- a/tests/test_lint_sql.py +++ b/tests/test_lint_sql.py @@ -343,3 +343,27 @@ async def f(cur): ''' findings = _findings(source, tmp_path / "a.py") assert [f.rule for f in findings].count("sql-inline-too-complex") == 3 + + +_COMPLEX_SELECT = '''""" +select a.id, p.price +from core.dim_article a +left join core.dim_price p on p.article_id = a.id +where a.active +"""''' + + +def test_sql_named_variable_with_complex_literal_flagged_without_execute(tmp_path: Path) -> None: + source = f"ARTICLES_SQL = {_COMPLEX_SELECT}\nother_sql: str = {_COMPLEX_SELECT}\nnot_a_query = {_COMPLEX_SELECT}\n" + findings = _findings(source, tmp_path / "a.py") + assert [(f.rule, f.line) for f in findings] == [("sql-inline-too-complex", 1), ("sql-inline-too-complex", 7)] + + +def test_sql_named_variable_simple_or_dml_not_flagged(tmp_path: Path) -> None: + source = 'GET_SQL = "select id from t where id = %(id)s"\nINSERT_SQL = """\ninsert into t (\n a,\n b,\n c,\n d\n) values (1, 2, 3, 4)\n"""\n' + assert _findings(source, tmp_path / "a.py") == [] + + +def test_sql_named_variable_used_in_execute_reported_once(tmp_path: Path) -> None: + source = f"ARTICLES_SQL = {_COMPLEX_SELECT}\n\ndef f(cur):\n cur.execute(ARTICLES_SQL)\n" + assert len(_findings(source, tmp_path / "a.py")) == 1