feat(lint): allow simple inline insert/update/delete

README.md

@@ -535,6 +535,10 @@ 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.
+  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()`

bmsdna/devtools/lint_sql.py

@@ -280,20 +280,26 @@ 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()])
-    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(
@@ -304,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(
@@ -430,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
@@ -470,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)
 
 

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

tests/test_lint_sql.py

@@ -294,3 +294,76 @@ 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
+
+
+_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