Reviews a schema and slow SQL queries, proposes composite indexes in the right column order (equality, sort, range) with covering columns and tradeoffs, flags non-sargable filters like functions on columns and leading-wildcard LIKE, and warns about duplicate, redundant, and low-selectivity indexes. Use when a user shares a slow query, EXPLAIN output, or schema and asks "what index do I need?" or "too many indexes?". Includes a tested stdlib Python helper.
---
name: sql-index-recommendation-reviewer
description: Reviews a schema and slow SQL queries, proposes composite indexes in the right column order (equality, sort, range) with covering columns and tradeoffs, flags non-sargable filters like functions on columns and leading-wildcard LIKE, and warns about duplicate, redundant, and low-selectivity indexes. Use when a user shares a slow query, EXPLAIN output, or schema and asks "what index do I need?" or "too many indexes?". Includes a tested stdlib Python helper.
---
# SQL Index Recommendation Reviewer
You help developers make slow queries fast without burying their tables in indexes. You read the schema and the query the way a database engineer would: which columns are compared with equality, which with ranges, what the query sorts by, what it joins on, and which indexes already exist. You propose the smallest set of indexes that serves the workload, in the right column order, and you are honest that only EXPLAIN on real data can confirm it.
## Files in this skill
- `scripts/suggest_indexes.py` - parses CREATE TABLE / CREATE INDEX statements and SELECT, UPDATE, and DELETE queries (subqueries and CTE bodies separately), then prints candidate indexes with DDL, the queries each one serves, covering columns, overlaps, and schema findings (Python 3 standard library only; no database connection)
- `references/indexing-rules.md` - column order, covering indexes, partial and expression indexes, sargability, and over-indexing rules
- `references/explain-reading-guide.md` - how to read EXPLAIN in PostgreSQL, MySQL, and SQLite to confirm or reject a suggestion
- `templates/index-review.md` - the review report to return
- `examples/example-shop-orders-review.md` - a worked review of an online shop's slow queries
## Workflow
### 1. Collect the facts
Ask, or assume and say so:
- The engine and version (PostgreSQL, MySQL/MariaDB, SQLite, SQL Server), and the schema with existing indexes (`\d table`, `SHOW CREATE TABLE`, or the migration files).
- The slow queries exactly as the app sends them, with typical parameter values, how often each runs, and how slow it is.
- Table sizes (rows), how write-heavy each table is, and any EXPLAIN or EXPLAIN ANALYZE output.
- Constraints: can the team build indexes online, disk space, and replication lag concerns.
### 2. Run the helper
```bash
python3 scripts/suggest_indexes.py --schema schema.sql slow-queries.sql
python3 scripts/suggest_indexes.py all-in-one.sql --dialect mysql
python3 scripts/suggest_indexes.py --schema schema.sql - --json < query.sql
python3 scripts/suggest_indexes.py --schema schema.sql queries.sql --strict # exit 1 on non-sargable filters or duplicate indexes
```
The helper is a heuristic parser, not a planner: it does not know row counts, data distribution, or which plan the database will choose. It can miss predicates in complex SQL (it lists them as "not understood"). Read every query yourself and treat each suggestion as a candidate to test. If you cannot run the script, apply `references/indexing-rules.md` by hand in the same order.
### 3. Decide
1. Fix the query before adding an index: rewrite non-sargable filters (functions on columns, arithmetic, leading-wildcard LIKE, OR across columns) so an index can be used at all.
2. For each query, choose column order: equality columns (most selective first when you know it), then ORDER BY or GROUP BY columns if the query sorts with LIMIT, then one range column.
3. Merge: one composite index can serve several queries when they share its leftmost columns. Prefer extending an existing index over adding a new one.
4. Add covering (INCLUDE) columns only for hot, read-heavy queries, and keep them few.
5. Remove before adding: drop duplicate and redundant-prefix indexes after checking usage statistics (`pg_stat_user_indexes`, `sys.schema_unused_indexes`).
6. Count the cost: every index slows inserts and updates of its columns and uses disk and memory. Write-heavy tables deserve fewer, broader indexes.
### 4. Deliver
Fill in `templates/index-review.md`: per-query diagnosis, the recommended DDL in build order with the queries each serves, indexes to drop, query rewrites, tradeoffs, and how to verify with EXPLAIN before and after (see `references/explain-reading-guide.md`). Follow the style of `examples/example-shop-orders-review.md`.
## Rules
- Never claim a speed-up without saying how to measure it; give expected plan changes ("index scan instead of sequential scan and sort"), not invented timings.
- Use safe build commands for production: `CREATE INDEX CONCURRENTLY` in PostgreSQL, online DDL (`ALGORITHM=INPLACE, LOCK=NONE`) in MySQL, and a maintenance window for SQLite.
- Do not recommend an index on a low-selectivity column alone (status, booleans); use it inside a composite or as a partial index condition.
- Keep primary keys and unique constraints; never suggest dropping them to remove "redundancy".
- Tell the user to build one index at a time and compare plans, and to remove indexes that EXPLAIN never uses.
FILE:references/indexing-rules.md
# Indexing rules
The rules the helper applies, plus the judgment it cannot make. Examples use PostgreSQL syntax unless noted.
## 1. Column order in a composite index
A B-tree index is sorted by its first column, then the second, and so on. A query can seek on a leftmost run of columns compared with equality, then use one more column for a range or for sorting.
Order: **equality columns, then sort columns, then one range column**.
```sql
-- WHERE customer_id = ? AND status = 'paid' ORDER BY created_at DESC LIMIT 20
CREATE INDEX ON orders (customer_id, status, created_at);
```
- Among equality columns, put the one used by the most queries first so the index serves them all; selectivity matters less than reuse once all of them are equalities.
- After a range column (`>`, `<`, `BETWEEN`, `LIKE 'abc%'`), later columns cannot be used for seeking, only for filtering inside the index.
- A sort can use the index only if every ORDER BY column comes from the index in the same order and directions are all the same (or all reversed). Mixed directions need a matching index definition (`created_at DESC, id ASC`).
- With a range and an ORDER BY on a different column, the database must choose: seek with the range and sort, or walk the sort order and filter. With LIMIT, walking the sort order often wins.
## 2. Leftmost prefix and merging
An index on `(a, b, c)` serves queries on `a`, on `a, b`, and on `a, b, c`, but not on `b` or `c` alone. Two candidates where one is a leftmost prefix of the other become one index. An existing index `(a)` is redundant once `(a, b)` exists, unless it is unique, partial, or much smaller and very hot.
## 3. Joins and foreign keys
- Index the column on the side that is looked up: usually the foreign key (`orders.customer_id`). The primary key side is already indexed.
- PostgreSQL does not index foreign keys automatically (MySQL InnoDB does). Missing FK indexes also make deletes on the parent table slow.
- A join column plus that table's own filters can form one composite: `(customer_id, created_at)`.
## 4. Covering indexes
If the index holds every column the query reads, the database can skip the table (index-only scan).
- PostgreSQL 11+ and SQL Server: `INCLUDE (col, ...)` stores extra columns without sorting by them.
- MySQL and SQLite: append the columns to the key; InnoDB secondary indexes already carry the primary key.
- Use it for hot, read-mostly queries with few selected columns. Wide covering indexes are big and slow to write.
- PostgreSQL index-only scans also need a well-vacuumed table (visibility map).
## 5. Partial and expression indexes
- Partial (PostgreSQL, SQLite; filtered index in SQL Server): `CREATE INDEX ON orders (created_at) WHERE status = 'paid' AND shipped_at IS NULL;` Tiny and fast when the condition matches few rows. The query must repeat the same condition.
- Expression: `CREATE INDEX ON users (lower(email));` serves `WHERE lower(email) = ?`. MySQL 8 supports functional indexes `((lower(email)))` or a generated column with an index.
## 6. Sargability: filters an index cannot use
| Pattern | Problem | Fix |
|---|---|---|
| `WHERE date(created_at) = '2026-10-12'` | function on the column | `created_at >= '2026-10-12' AND created_at < '2026-10-13'` |
| `WHERE YEAR(updated_at) = 2026` | function on the column | range on the column |
| `WHERE price * 1.2 > 100` | arithmetic on the column | `price > 100 / 1.2` |
| `WHERE name LIKE '%lamp%'` | leading wildcard | full-text search, or `pg_trgm` GIN index |
| `WHERE lower(name) = ?` | function | expression index on `lower(name)` or a case-insensitive collation |
| `WHERE a = 1 OR b = 2` | OR across columns | `UNION` of two indexed queries, or one index per column (bitmap OR in PostgreSQL) |
| `WHERE status <> 'done'`, `NOT IN` | negative filter | positive list `IN (...)` or a partial index |
| `WHERE id = '42'` on an integer column | implicit cast (MySQL may skip the index on string columns compared to numbers) | compare with the right type |
## 7. Over-indexing
- Every index is updated on INSERT, on DELETE, and on UPDATE of its columns; it also uses disk, memory (buffer cache), and backup and replication bandwidth.
- Duplicates (same columns, same condition) are pure cost: drop one.
- Single-column indexes on low-selectivity columns (booleans, status, type) are rarely used; the planner prefers a scan when a value matches more than about 5 to 20 percent of rows. Use them inside composites or as partial-index conditions.
- As a rule of thumb, more than 5 or 6 indexes on a write-heavy table deserves a review; read-mostly tables can carry more.
- Find unused indexes before dropping: PostgreSQL `pg_stat_user_indexes.idx_scan = 0` (since the last stats reset, on every replica), MySQL `sys.schema_unused_indexes`. Drop with `DROP INDEX CONCURRENTLY` in PostgreSQL.
## 8. Building safely
- PostgreSQL: `CREATE INDEX CONCURRENTLY` avoids blocking writes; it cannot run inside a transaction and leaves an INVALID index on failure (drop and retry).
- MySQL 8: `ALTER TABLE ... ADD INDEX ..., ALGORITHM=INPLACE, LOCK=NONE`; watch replication lag on large tables.
- SQLite: index builds lock the database; run during low traffic.
- Build one index at a time, check the plan, then the next.
FILE:references/explain-reading-guide.md
# Reading EXPLAIN to confirm an index
Always compare the plan before and after, on production-like data and with realistic parameter values. Small test tables make the planner choose scans that are right for small tables.
## PostgreSQL
```sql
EXPLAIN (ANALYZE, BUFFERS) SELECT ...; -- runs the query; wrap writes in BEGIN ... ROLLBACK
```
| Plan node | Meaning |
|---|---|
| Seq Scan | reads the whole table; fine for small tables or when most rows match |
| Index Scan | seeks the index, then reads matching table rows |
| Index Only Scan | answered from the index; check "Heap Fetches" is low (vacuum if not) |
| Bitmap Index Scan + Bitmap Heap Scan | collects many matches, then reads pages in order; common for medium selectivity and OR |
| Sort (with "external merge") | sorting spilled to disk; an index in ORDER BY order can remove it |
| Rows Removed by Filter | rows read and thrown away; high numbers mean the index does not cover the filter |
Look at: actual time of the top node, rows estimated versus actual (a 10x gap means stale statistics: run `ANALYZE table`), and shared hit/read buffers. Confirm a new index is used with `pg_stat_user_indexes.idx_scan` after a day of traffic.
## MySQL 8
```sql
EXPLAIN FORMAT=TREE SELECT ...;
EXPLAIN ANALYZE SELECT ...; -- 8.0.18+, runs the query
```
In classic `EXPLAIN` output check: `type` (ALL = full scan, ref/eq_ref/range = index use, index = full index scan), `key` (the index chosen), `rows` (estimate), and `Extra` ("Using index" = covering, "Using filesort" = extra sort, "Using temporary" = temporary table for GROUP BY or DISTINCT). Refresh statistics with `ANALYZE TABLE t`.
## SQLite
```sql
EXPLAIN QUERY PLAN SELECT ...;
```
"SCAN t" is a full scan; "SEARCH t USING INDEX idx (a=? AND b>?)" shows which columns seek; "USING COVERING INDEX" means no table access; "USE TEMP B-TREE FOR ORDER BY" means an extra sort. Run `ANALYZE;` so the planner has statistics.
## What success looks like
- The slow query's plan switches from a full scan plus sort to an index (only) scan with no sort, and the rows read drop close to the rows returned.
- Writes on the table do not get noticeably slower (check p95 insert and update latency after deploying).
- No other important query changed plan for the worse; run the top queries by total time (`pg_stat_statements`, MySQL performance schema digest tables) before and after.
## When the index is not used
1. Statistics are stale: run ANALYZE.
2. The filter is not sargable (see indexing-rules.md section 6).
3. The value matches too many rows; the planner is right to scan.
4. Column types or collations differ between the query and the column (implicit cast).
5. A parameterized query uses a generic plan; test with the real prepared statement.
6. The index columns are in the wrong order for the query.
FILE:templates/index-review.md
# Index review: app_or_service
**Engine:** engine_and_version | **Tables:** tables_with_row_counts | **Write load:** write_profile
**Helper:** n_queries queries analyzed, n_suggested candidates, n_findings schema findings
## Summary
Two or three sentences: the main cause of slowness, the indexes to add and drop, and the expected plan change.
## Per-query diagnosis
| Query | Runs | Now (plan / time) | Problem | Fix |
|---|---|---|---|---|
| Q1 | per_minute | Seq Scan + Sort | | |
## Query rewrites (do these first)
```sql
-- before
-- after
```
## Indexes to add (build in this order)
| # | DDL | Serves | Why this column order | Cost |
|---|---|---|---|---|
| 1 | | | | |
## Indexes to drop (after checking usage)
| Index | Reason | Check first |
|---|---|---|
| | duplicate / redundant prefix / unused / low selectivity | `idx_scan` on primary and replicas |
## Net effect per table
| Table | Indexes before | After | Note |
|---|---|---|---|
## How to verify
1. Capture `EXPLAIN (ANALYZE, BUFFERS)` for each query before.
2. Build one index (`CONCURRENTLY` / online DDL), run `ANALYZE`, capture the plan again.
3. Expect: (plan change per query).
4. Watch write latency and replication lag for a day; then drop the indexes listed above.
## Caveats
- Suggestions come from query text and schema only; row counts and data skew can change the best choice.
FILE:examples/example-shop-orders-review.md
# Example: index review for an online shop
## User request
"Our PostgreSQL 16 shop database is getting slow. orders has 12 million rows and gets about 40 inserts a second, customers has 900k rows, products 20k. Here are the schema and the six queries that show up in pg_stat_statements. What indexes should we add, and do we have too many already?"
## Helper run
```
$ python3 scripts/suggest_indexes.py --schema shop-schema.sql slow-queries.sql
Schema: 4 table(s), 12 existing index(es) incl. keys | dialect: postgres
Queries analyzed: 7
Q1 SELECT: SELECT o.id, o.total, o.created_at FROM orders o WHERE o.customer_id = $1 AND o.status = 'paid' ORDER BY o.cre...
equality: orders.customer_id (=), orders.status (=)
order by: orders.created_at DESC (LIMIT)
Q2 SELECT: SELECT c.email, count(*) AS orders FROM customers c JOIN orders o ON o.customer_id = c.id WHERE c.country = 'D...
equality: customers.country (=)
range: orders.created_at (range)
joins: orders.customer_id = customers.id
group by: customers.email
Q3 SELECT: SELECT * FROM products WHERE lower(name) LIKE '%lamp%'
WARN non-sargable: lower(name) in the WHERE clause hides the column from a normal index; ... the LIKE pattern '%lamp%' also starts with a wildcard, so only full-text search or a trigram index can help
Q4 SELECT: ... WHERE oi.order_id IN (SELECT id FROM orders WHERE customer_id = 42 AND created_at > now() - interval '30 days')
note: subquery analyzed as Q5
Q6 UPDATE: UPDATE orders SET status = 'shipped', shipped_at = now() WHERE status = 'paid' AND shipped_at IS NULL AND crea...
Q7 SELECT: SELECT id FROM customers WHERE email = 'ann@example.com' OR phone = '+491701234567'
WARN OR across different columns (...): one composite index cannot serve both sides; consider UNION ALL of two indexed queries or one index per side
Suggested indexes (4):
1. CREATE INDEX CONCURRENTLY idx_customers_country_email ON customers (country, email);
2. CREATE INDEX CONCURRENTLY idx_orders_customer_id_created_at ON orders (customer_id, created_at);
serves: Q2, Q5 | overlap: same leading column as #3
3. CREATE INDEX CONCURRENTLY idx_orders_customer_id_status_created_at ON orders (customer_id, status, created_at) INCLUDE (id, total);
serves: Q1 | overlap: same leading column as #2
4. CREATE INDEX CONCURRENTLY idx_orders_status_shipped_at_created_at ON orders (status, shipped_at, created_at);
serves: Q6 | alternative: ... ON orders (status, created_at) WHERE shipped_at IS NULL;
Schema findings (5):
WARN redundant-prefix customers: idx_customers_country (country) is a leftmost prefix of idx_customers_country_created ...
WARN low-selectivity customers: idx_customers_active indexes only 'is_active' ...
HIGH duplicate-index orders: idx_orders_created and orders_created_at_idx have the same columns (created_at); drop one
WARN low-selectivity orders: idx_orders_status indexes only 'status' ...
WARN many-indexes orders: 4 existing + 3 suggested = 7 indexes (limit 6) ...
```
## Review (filled template, shortened)
**Summary:** orders has no index that starts with customer_id, so the order history page (Q1) and the recent-orders lookup (Q5) scan millions of rows, and two of its four indexes are a duplicate and a low-selectivity status index. Add one composite and one partial index on orders, a phone index and a trigram index elsewhere, and drop three indexes: orders ends with the same count of 4 but every one is useful.
| Query | Problem | Fix |
|---|---|---|
| Q1 order history | no index on customer_id; sort on created_at | (customer_id, created_at) with status and total included: newest rows for the customer, status filtered inside the index, LIMIT stops after 20 |
| Q2 country report | join needs orders.customer_id; customers.country is already served by idx_customers_country_created | same orders index; no new customers index (the helper's #1 only helps the GROUP BY, and the report runs once a night) |
| Q3 product search | `lower(name) LIKE '%lamp%'` cannot use a B-tree | pg_trgm GIN index on lower(name) |
| Q4/Q5 recent items | subquery needs (customer_id, created_at); order_items PK (order_id, product_id) serves the IN | same orders index |
| Q6 shipping job | scans by status | partial index on created_at for paid, unshipped orders only |
| Q7 support lookup | OR across email and phone | UNION of two lookups; email is unique already, add a phone index |
Why not the helper's #3 as well: #2 and #3 share customer_id, and on a table with 40 inserts a second one index is better than two. The user confirmed that about 95 percent of orders are paid, so Q1 throws away few rows, and (customer_id, created_at) with status in INCLUDE keeps Q1 fast. If the share of non-paid orders grows, switch to #3's column order.
**Indexes to add (in this order):**
```sql
CREATE INDEX CONCURRENTLY idx_orders_customer_created
ON orders (customer_id, created_at) INCLUDE (id, status, total); -- Q1, Q2, Q5
CREATE INDEX CONCURRENTLY idx_orders_paid_unshipped
ON orders (created_at) WHERE status = 'paid' AND shipped_at IS NULL; -- Q6, holds only paid, unshipped orders
CREATE INDEX CONCURRENTLY idx_customers_phone ON customers (phone); -- Q7 after the rewrite
CREATE EXTENSION IF NOT EXISTS pg_trgm;
CREATE INDEX CONCURRENTLY idx_products_name_trgm
ON products USING gin (lower(name) gin_trgm_ops); -- Q3, query unchanged
```
**Query rewrite (Q7):**
```sql
SELECT id FROM customers WHERE email = 'ann@example.com'
UNION
SELECT id FROM customers WHERE phone = '+491701234567';
```
**Indexes to drop (after checking `pg_stat_user_indexes.idx_scan` on the primary and both replicas):**
| Index | Reason |
|---|---|
| orders_created_at_idx | duplicate of idx_orders_created |
| idx_orders_status | low selectivity; Q6 now uses the partial index |
| idx_customers_country | leftmost prefix of idx_customers_country_created |
| idx_customers_active (check) | boolean; keep only if idx_scan shows real use |
**Net effect:** orders 4 -> 4 indexes (two replaced by better ones), customers 5 -> 5 (one dropped, phone added; 4 if is_active goes too), products 2 -> 3.
**Recheck with the helper** (schema after the change, queries after the rewrite; query numbers follow the new file, so Q3 is the old Q5 and Q4/Q5 are the two UNION branches):
```
$ python3 scripts/suggest_indexes.py --schema shop-schema-after.sql shop-queries-after.sql --no-covering
Already served by existing indexes (5):
customers (email) -> customers_email_key (Q4)
customers (id, country) -> customers_pkey (Q2)
customers (phone) -> idx_customers_phone (Q5)
orders (created_at) -> idx_orders_created (Q2)
orders (customer_id, created_at) -> idx_orders_customer_created (Q2, Q3)
```
The helper still lists (customer_id, status, created_at) for Q1 and (country, email) for Q2; both were rejected on purpose for the reasons above. There are no duplicate or redundant-prefix findings left.
**How to verify:** before and after each build, run `EXPLAIN (ANALYZE, BUFFERS)` for Q1 with a real customer id. Expect "Index Scan using idx_orders_customer_created" (or Index Only Scan) with no Sort node and about 20 rows read instead of a Bitmap Heap Scan over thousands. Watch insert latency on orders for a day before dropping the old indexes.
FILE:scripts/suggest_indexes.py
#!/usr/bin/env python3
"""Suggest candidate indexes for slow SQL queries from a schema and the query text (no database needed).
Usage:
python3 suggest_indexes.py --schema schema.sql queries.sql [options]
python3 suggest_indexes.py all-in-one.sql (CREATE statements and queries in one file)
python3 suggest_indexes.py --schema schema.sql - < slow.sql
Options:
--schema FILE CREATE TABLE / CREATE INDEX statements (may also be inside the query file)
--dialect NAME postgres (default), mysql, or sqlite; changes INCLUDE and online-build hints
--max-indexes N warn when a table would have more than N indexes (default 6)
--no-covering do not suggest covering (INCLUDE) columns
--json print the result as JSON
--strict exit 1 if any query has a non-sargable predicate or the schema has duplicate indexes
What it does: lightweight parsing of SELECT, UPDATE, and DELETE statements (subqueries are analyzed
separately) to find equality filters, range filters, join columns, ORDER BY and GROUP BY columns,
and selected columns per table. It builds composite candidates (equality columns, then sort, then
range), merges candidates that are prefixes of each other, skips those already served by an existing
index, and reports duplicate and redundant indexes, low-selectivity single-column indexes, and tables
with many indexes. It does not know row counts or data distribution: confirm every suggestion with
EXPLAIN on real data.
Exit codes: 0 ok, 1 --strict findings, 2 usage or read error.
Python 3 standard library only.
"""
import json
import re
import signal
import sys
if hasattr(signal, "SIGPIPE"):
signal.signal(signal.SIGPIPE, signal.SIG_DFL) # quiet exit when piped into head
KEYWORDS = set("""select from where and or not in is null like ilike between join inner left right full outer cross on
using group by order having limit offset as asc desc distinct union all exists case when then else end update set
delete insert into values returning with true false interval current_date current_timestamp now lateral natural
fetch first rows only for share nowait skip locked""".split())
LOW_CARD = re.compile(r"^(is_|has_|can_|should_)|^(status|state|type|kind|flag|active|enabled|deleted|archived|gender|"
r"visible|published|verified|priority|level|tier|role)$", re.I)
CLAUSE_END = r"(?=\b(?:where|group\s+by|order\s+by|having|limit|offset|union|returning|for\s+update|window)\b|$)"
class UsageError(Exception):
pass
# ---------------------------------------------------------------- text helpers
def strip_comments(sql):
sql = re.sub(r"/\*.*?\*/", " ", sql, flags=re.S)
return re.sub(r"--[^\n]*", " ", sql)
def split_statements(sql):
out, buf, quote = [], [], None
for ch in sql:
if quote:
buf.append(ch)
if ch == quote:
quote = None
continue
if ch in "'\"`":
quote = ch
if ch == ";":
out.append("".join(buf))
buf = []
else:
buf.append(ch)
out.append("".join(buf))
return [s.strip() for s in out if s.strip()]
def unquote_idents(sql):
sql = re.sub(r'"([A-Za-z_][\w$]*)"', r"\1", sql)
sql = re.sub(r"`([A-Za-z_][\w$]*)`", r"\1", sql)
return re.sub(r"\[([A-Za-z_][\w$]*)\]", r"\1", sql)
def mask_strings(sql):
"""Replace string literals with placeholders, remembering LIKE patterns."""
lits = []
def rep(m):
lits.append(m.group(0)[1:-1])
return f" __lit{len(lits) - 1}__ "
return re.sub(r"'(?:[^']|'')*'", rep, sql), lits
def split_top(text, sep_regex):
"""Split on a keyword at parenthesis depth 0."""
parts, depth, last = [], 0, 0
for m in re.finditer(r"\(|\)|" + sep_regex, text, flags=re.I):
tok = m.group(0)
if tok == "(":
depth += 1
elif tok == ")":
depth -= 1
elif depth == 0:
parts.append(text[last:m.start()])
last = m.end()
parts.append(text[last:])
return [p.strip() for p in parts if p.strip()]
def split_and(text):
"""Split on top-level AND, keeping 'x BETWEEN a AND b' together."""
parts = split_top(text, r"\band\b")
out = []
for part in parts:
if out and re.search(r"\bbetween\s+[^\s]+$", out[-1], re.I):
out[-1] = f"{out[-1]} AND {part}"
else:
out.append(part)
return out
def extract_subqueries(sql):
"""Pull out innermost (SELECT ...) blocks; return (outer_sql, [subqueries])."""
subs = []
pattern = re.compile(r"\(\s*(select\b[^()]*(?:\([^()]*\)[^()]*)*)\)", re.I)
while True:
m = pattern.search(sql)
if not m:
break
subs.append(m.group(1))
sql = sql[:m.start()] + " __subq__ " + sql[m.end():]
return sql, subs
# ---------------------------------------------------------------- schema
def parse_columns_list(text):
return [re.sub(r"\s+(asc|desc)\b.*$", "", c.strip(), flags=re.I).strip().lower()
for c in split_top(text, r",")]
def parse_schema(statements):
tables, indexes = {}, []
for st in statements:
s = unquote_idents(strip_comments(st)).strip()
m = re.match(r"create\s+table\s+(?:if\s+not\s+exists\s+)?([\w.]+)\s*\((.*)\)\s*[^)]*$", s, re.I | re.S)
if m:
name = m.group(1).split(".")[-1].lower()
cols, pk = [], None
for part in split_top(m.group(2), r","):
p = part.strip()
low = p.lower()
c = re.match(r"(?:constraint\s+\w+\s+)?primary\s+key\s*\(([^)]*)\)", low)
if c:
pk = parse_columns_list(c.group(1))
continue
c = re.match(r"(?:constraint\s+(\w+)\s+)?unique\s*(?:key\s+\w+\s*)?\(([^)]*)\)", low)
if c:
indexes.append({"name": c.group(1) or f"{name}_unique", "table": name,
"columns": parse_columns_list(c.group(2)), "unique": True, "kind": "unique"})
continue
c = re.match(r"(?:key|index)\s+(\w+)\s*\(([^)]*)\)", low)
if c:
indexes.append({"name": c.group(1), "table": name, "columns": parse_columns_list(c.group(2)),
"unique": False, "kind": "index"})
continue
if re.match(r"(constraint|foreign\s+key|check|exclude)\b", low):
continue
col = re.match(r"([a-z_][\w$]*)\s+", low + " ")
if col:
cols.append(col.group(1))
if re.search(r"\bprimary\s+key\b", low):
pk = [col.group(1)]
elif re.search(r"\bunique\b", low):
indexes.append({"name": f"{name}_{col.group(1)}_key", "table": name,
"columns": [col.group(1)], "unique": True, "kind": "unique"})
tables[name] = {"columns": cols}
if pk:
indexes.append({"name": f"{name}_pkey", "table": name, "columns": pk, "unique": True,
"kind": "primary"})
continue
m = re.match(r"create\s+(unique\s+)?index\s+(?:concurrently\s+)?(?:if\s+not\s+exists\s+)?(\w+)\s+on\s+"
r"([\w.]+)\s*(?:using\s+\w+\s*)?\(([^;]*?)\)\s*(include\s*\(([^)]*)\))?\s*(where\s+.*)?$",
s, re.I | re.S)
if m:
cols = parse_columns_list(m.group(4))
indexes.append({"name": m.group(2), "table": m.group(3).split(".")[-1].lower(), "columns": cols,
"unique": bool(m.group(1)), "kind": "index",
"include": parse_columns_list(m.group(6)) if m.group(6) else [],
"partial": (m.group(7) or "").strip() or None})
return tables, indexes
# ---------------------------------------------------------------- queries
def parse_from(from_text):
"""Return (tables {alias: table}, order list, join predicates text list)."""
aliases, order, ons = {}, [], []
pieces = re.split(r"\b((?:natural\s+)?(?:inner\s+|left\s+(?:outer\s+)?|right\s+(?:outer\s+)?|full\s+(?:outer\s+)?|"
r"cross\s+)?join)\b", from_text, flags=re.I)
items = [pieces[0]] + [pieces[i + 1] for i in range(1, len(pieces) - 1, 2)]
for item in items:
on_split = re.split(r"\bon\b", item, maxsplit=1, flags=re.I)
using = re.search(r"\busing\s*\(([^)]*)\)", on_split[0], re.I)
head = re.sub(r"\busing\s*\([^)]*\)", "", on_split[0], flags=re.I)
for ref in head.split(","):
toks = [t for t in re.split(r"\s+", ref.strip()) if t and t.lower() not in ("as", "lateral", "only")]
if not toks or toks[0] == "__subq__":
continue
table = toks[0].split(".")[-1].lower()
alias = toks[1].lower() if len(toks) > 1 else table
aliases[alias] = table
order.append(alias)
if len(on_split) > 1:
ons.append(on_split[1])
if using and len(order) >= 2:
for col in parse_columns_list(using.group(1)):
ons.append(f"{order[-2]}.{col} = {order[-1]}.{col}")
return aliases, order, ons
COL = r"((?:[a-z_][\w$]*\.)?[a-z_][\w$]*)"
VALUE = r"(?:__lit\d+__|\?|:\w+|\$\d+|@\w+|%s|-?\d+(?:\.\d+)?|true|false|null|__subq__|current_date|current_timestamp|now\(\))"
def resolve(ref, aliases, tables):
ref = ref.lower()
if "." in ref:
a, c = ref.split(".", 1)
return (aliases.get(a, a), c)
if len(set(aliases.values())) == 1:
return (next(iter(aliases.values())), ref)
owners = [t for t in set(aliases.values()) if ref in tables.get(t, {}).get("columns", [])]
return (owners[0], ref) if len(owners) == 1 else (None, ref)
def analyze_predicate(pred, lits, aliases, tables, q):
p = pred.strip()
while p.startswith("(") and p.endswith(")") and len(split_top(p[1:-1], r"\band\b")) >= 1 and \
p.count("(") == p.count(")") and _balanced_wrap(p):
p = p[1:-1].strip()
if len(split_and(p)) > 1:
for sub in split_and(p):
analyze_predicate(sub, lits, aliases, tables, q)
return
ors = split_top(p, r"\bor\b")
if len(ors) > 1:
cols = set()
for o in ors:
m = re.match(rf"^\(?\s*{COL}\s*(=|in\b)", o.strip(), re.I)
cols.add(m.group(1).lower() if m else None)
if len(cols) == 1 and None not in cols:
t, c = resolve(cols.pop(), aliases, tables)
q["filters"].append({"table": t, "column": c, "kind": "eq", "op": "OR of equalities (same as IN)"})
else:
shown = re.sub(r"__lit(\d+)__", lambda m: "'" + lits[int(m.group(1))] + "'", p)
shown = shown if len(shown) <= 70 else shown[:67] + "..."
q["warnings"].append(f"OR across different columns ({shown}): one composite index cannot serve "
"both sides; consider UNION ALL of two indexed queries or one index per side")
return
if re.match(r"^not\b", p, re.I) or re.search(r"(<>|!=|\bnot\s+in\b|\bnot\s+like\b)", p, re.I):
m = re.match(rf"^(?:not\s+)?{COL}", p, re.I)
shown = re.sub(r"__lit(\d+)__", lambda m: "'" + lits[int(m.group(1))] + "'", p)[:50]
q["notes"].append(f"negative predicate '{shown}' rarely uses an index; it is applied as a filter")
return
m = re.match(rf"^{COL}\s*(=|<=>)\s*{COL}$", p, re.I)
if m and not re.fullmatch(VALUE, m.group(3), re.I) and m.group(3).lower() not in KEYWORDS:
left, right = resolve(m.group(1), aliases, tables), resolve(m.group(3), aliases, tables)
if left[0] and right[0] and left[0] != right[0]:
q["joins"].append((left, right))
return
m = re.match(rf"^{VALUE}\s*(=|<|>|<=|>=)\s*{COL}$", p, re.I)
if m:
flip = {"<": ">", ">": "<", "<=": ">=", ">=": "<=", "=": "="}
p = f"{m.group(2)} {flip[m.group(1)]} __v__"
m = re.match(rf"^{COL}\s*(=|<=>)\s*(.+)$", p, re.I)
if m and m.group(1).lower() not in KEYWORDS:
t, c = resolve(m.group(1), aliases, tables)
q["filters"].append({"table": t, "column": c, "kind": "eq", "op": "="})
return
m = re.match(rf"^{COL}\s+in\s+__subq__$", p, re.I)
if m:
t, c = resolve(m.group(1), aliases, tables)
q["filters"].append({"table": t, "column": c, "kind": "eq", "op": "IN (subquery)"})
return
m = re.match(rf"^{COL}\s+in\s*\((.*)\)$", p, re.I | re.S)
if m:
t, c = resolve(m.group(1), aliases, tables)
q["filters"].append({"table": t, "column": c, "kind": "eq", "op": "IN"})
return
m = re.match(rf"^{COL}\s+is\s+(not\s+)?null$", p, re.I)
if m:
t, c = resolve(m.group(1), aliases, tables)
if m.group(2):
q["notes"].append(f"'{c} IS NOT NULL' is usually not selective; a partial index can help if most rows are NULL")
else:
q["filters"].append({"table": t, "column": c, "kind": "eq", "op": "IS NULL"})
return
m = re.match(rf"^{COL}\s*(<|>|<=|>=)\s*(.+)$", p, re.I) or re.match(rf"^{COL}\s+between\s+(.+)\s+and\s+(.+)$", p, re.I)
if m and m.group(1).lower() not in KEYWORDS:
t, c = resolve(m.group(1), aliases, tables)
q["filters"].append({"table": t, "column": c, "kind": "range", "op": "range"})
return
m = re.match(rf"^{COL}\s+(i?like)\s+__lit(\d+)__", p, re.I)
if m:
t, c = resolve(m.group(1), aliases, tables)
pattern = lits[int(m.group(3))]
if pattern.startswith(("%", "_")):
q["warnings"].append(f"non-sargable: {c} {m.group(2).upper()} '{pattern}' starts with a wildcard, so a "
"B-tree index cannot seek; use full-text search or a trigram index (PostgreSQL pg_trgm)")
q["non_sargable"] = True
else:
q["filters"].append({"table": t, "column": c, "kind": "range", "op": f"LIKE '{pattern}'"})
if m.group(2).lower() == "ilike":
q["notes"].append(f"ILIKE on {c}: a plain B-tree index is case-sensitive; index lower({c}) instead")
return
m = re.match(rf"^([a-z_]+)\s*\(\s*{COL}[^)]*\)\s*(=|<|>|<=|>=|like|between)", p, re.I)
if m and m.group(1).lower() not in KEYWORDS:
func, col = m.group(1).lower(), m.group(2)
t, c = resolve(col, aliases, tables)
wild = re.search(r"like\s+__lit(\d+)__", p, re.I)
extra = ""
if wild and lits[int(wild.group(1))].startswith(("%", "_")):
extra = (f"; the LIKE pattern '{lits[int(wild.group(1))]}' also starts with a wildcard, so only full-text "
"search or a trigram index can help")
q["warnings"].append(f"non-sargable: {func}({c}) in the WHERE clause hides the column from a normal index; "
f"rewrite as a range on {c} or add an expression index on {func}({c}){extra}")
q["non_sargable"] = True
return
m = re.match(rf"^{COL}\s*[-+*/]\s*\S+\s*(=|<|>|<=|>=)", p, re.I)
if m:
t, c = resolve(m.group(1), aliases, tables)
q["warnings"].append(f"non-sargable: arithmetic on {c} ('{p[:40]}'); move the arithmetic to the other side")
q["non_sargable"] = True
return
if "__subq__" in p and re.search(r"\bexists\b", p, re.I):
return
q["notes"].append(f"predicate not understood, check by hand: {p[:60]}")
def _balanced_wrap(p):
depth = 0
for i, ch in enumerate(p):
depth += ch == "("
depth -= ch == ")"
if depth == 0 and i < len(p) - 1:
return False
return True
def parse_query(raw, tables, number, lits=None, outer_aliases=None):
if lits is None:
sql, lits = mask_strings(unquote_idents(strip_comments(raw)))
else:
sql = raw
sql = re.sub(r"\s+", " ", sql).strip()
outer, subs = extract_subqueries(sql)
shown = re.sub(r"__lit(\d+)__", lambda m: "'" + lits[int(m.group(1))] + "'", re.sub(r"\s+", " ", raw.strip()))
shown = re.sub(r"\s+", " ", shown)
q = {"n": number, "text": shown[:110] + ("..." if len(shown) > 110 else ""), "kind": None, "aliases": {},
"filters": [], "joins": [], "order_by": [], "group_by": [], "selected": {}, "select_star": False,
"limit": False, "warnings": [], "notes": [], "non_sargable": False, "subqueries": subs, "lits": lits}
low = outer.lower()
cte_names = set()
if low.startswith("with "):
cte_names = {n.lower() for n in re.findall(r"(?:with|,)\s*(?:recursive\s+)?(\w+)\s+as\b", outer, re.I)}
q["notes"].append(f"CTE {', '.join(sorted(cte_names))} analyzed as separate queries; the final SELECT reads from it")
m = re.search(r"(?:\)|__subq__)\s*((?:select|update|delete)\b.*)$", outer, re.I)
outer = m.group(1) if m else outer
low = outer.lower()
if low.startswith("select"):
q["kind"] = "SELECT"
m = re.match(r"select\s+(?:distinct\s+)?(.*?)\s+from\s+(.*?)" + CLAUSE_END, outer, re.I)
if not m:
q["notes"].append("no FROM clause found")
return q
select_list, from_text = m.group(1), m.group(2)
elif low.startswith("update"):
q["kind"] = "UPDATE"
m = re.match(r"update\s+(.*?)\s+set\s+", outer, re.I)
select_list, from_text = "", m.group(1) if m else ""
elif low.startswith("delete"):
q["kind"] = "DELETE"
m = re.match(r"delete\s+from\s+(.*?)" + CLAUSE_END, outer, re.I)
select_list, from_text = "", m.group(1) if m else ""
else:
return None
own, _order, ons = parse_from(from_text)
own = {a: t for a, t in own.items() if t not in cte_names}
q["aliases"] = own
aliases = {**(outer_aliases or {}), **own} # outer aliases resolve correlated references
unknown = [t for t in set(own.values()) if tables and t not in tables]
if unknown:
q["notes"].append(f"table(s) not in the schema: {', '.join(sorted(unknown))}; existing indexes unknown")
for on in ons:
for pred in split_and(on):
analyze_predicate(pred, lits, aliases, tables, q)
w = re.search(r"\bwhere\b(.*?)(?=\b(?:group\s+by|order\s+by|having|limit|offset|returning|for\s+update)\b|$)", outer, re.I)
if w:
for pred in split_and(w.group(1)):
analyze_predicate(pred, lits, aliases, tables, q)
for clause, key in (("group\\s+by", "group_by"), ("order\\s+by", "order_by")):
m = re.search(rf"\b{clause}\b(.*?)(?=\b(?:having|order\s+by|limit|offset|for\s+update)\b|$)", outer, re.I)
if m:
for item in split_top(m.group(1), r","):
mm = re.match(rf"^{COL}\s*(asc|desc)?(?:\s+nulls\s+(?:first|last))?$", item.strip(), re.I)
if mm:
t, c = resolve(mm.group(1), aliases, tables)
q[key].append({"table": t, "column": c, "dir": (mm.group(2) or "asc").lower()})
else:
q["notes"].append(f"{key.replace('_', ' ')} expression '{item.strip()[:40]}' cannot use a plain column index")
q["limit"] = bool(re.search(r"\b(limit|fetch\s+first)\b", outer, re.I))
if select_list:
if re.search(r"(^|,)\s*(\w+\.)?\*\s*(,|$)", select_list):
q["select_star"] = True
for ref in re.findall(rf"\b{COL}\b(?!\s*\()", select_list, re.I):
if ref.lower() in KEYWORDS or ref.startswith("__") or re.fullmatch(r"\d+", ref):
continue
t, c = resolve(ref, aliases, tables)
if t and (not tables or c in tables.get(t, {}).get("columns", [c])):
q["selected"].setdefault(t, [])
if c not in q["selected"][t]:
q["selected"][t].append(c)
for f in q["filters"]:
if f["table"] is None:
q["notes"].append(f"column '{f['column']}' could belong to several tables; qualify it with an alias")
return q
# ---------------------------------------------------------------- candidates
def candidates_for(q, dialect, covering, pk_cols=None):
out = []
by_table = {}
for f in q["filters"]:
if f["table"]:
by_table.setdefault(f["table"], {"eq": [], "range": []})
lst = by_table[f["table"]][f["kind"]]
if f["column"] not in lst:
lst.append(f["column"])
tables_in_q = set(q["aliases"].values())
correlated = set()
for left, right in q["joins"]:
if left[0] in tables_in_q and right[0] not in tables_in_q:
correlated.add(left[0])
if right[0] in tables_in_q and left[0] not in tables_in_q:
correlated.add(right[0])
order_tables = {o["table"] for o in q["order_by"]}
for t in tables_in_q:
if t in correlated:
continue # reached through the correlated join below
eq = by_table.get(t, {}).get("eq", [])
rng = [c for c in by_table.get(t, {}).get("range", []) if c not in eq]
sort = []
if q["order_by"] and order_tables == {t} and len({o["dir"] for o in q["order_by"]}) == 1:
sort = [o["column"] for o in q["order_by"] if o["column"] not in eq]
elif q["group_by"] and {g["table"] for g in q["group_by"]} == {t}:
sort = [g["column"] for g in q["group_by"] if g["column"] not in eq]
key, why = list(eq), []
if eq:
why.append(f"equality on {', '.join(eq)} first")
if sort and (not rng or q["limit"]):
key += [c for c in sort if c not in key]
purpose = "already sorted" if q["order_by"] else "already grouped"
why.append(f"then {', '.join(sort)} so rows come out {purpose}" + (" (stops early with LIMIT)" if q["limit"] else ""))
if rng and rng[0] not in key:
key.append(rng[0])
why.append(f"{rng[0]} last as an in-index filter")
elif rng:
key.append(rng[0])
why.append(f"then the range on {rng[0]}" if eq else f"range on {rng[0]}")
if len(rng) > 1:
q["notes"].append(f"several range filters on {t} ({', '.join(rng)}): only the first can seek; the rest filter")
if sort:
q["notes"].append(f"ORDER BY on {t} cannot use the same index as the range on {rng[0]}; expect a sort step")
if not key:
continue
include = []
if covering and not q["select_star"] and q["kind"] == "SELECT":
extra = [c for c in q["selected"].get(t, []) if c not in key
and not (dialect == "mysql" and c in (pk_cols or {}).get(t, []))]
if 0 < len(extra) <= 3:
include = extra
out.append({"table": t, "columns": key, "include": include, "why": "; ".join(why), "queries": [q["n"]],
"source": "filter", "eq": list(eq),
"null_filter": [f["column"] for f in q["filters"] if f["table"] == t and f["op"] == "IS NULL"]})
for left, right in q["joins"]:
for (t, c), (ot, oc) in ((left, right), (right, left)):
if t not in tables_in_q:
continue # outer table of a correlated subquery
eq = [e for e in by_table.get(t, {}).get("eq", []) if e != c]
rng = [r for r in by_table.get(t, {}).get("range", []) if r != c and r not in eq][:1]
why = f"join column {t}.{c} = {ot}.{oc} first (lets the planner look up matching rows)"
if eq or rng:
why += f", then {', '.join(eq + rng)} to filter inside the index"
if t in correlated:
why = f"correlated lookup {t}.{c} = outer {ot}.{oc} first" + (f", then {', '.join(eq + rng)}" if eq or rng else "")
out.append({"table": t, "columns": [c] + eq + rng, "include": [], "queries": [q["n"]], "source": "join",
"eq": [c] + eq, "why": why})
return out
def served_by(cand, indexes):
for ix in indexes:
if ix["table"] != cand["table"]:
continue
cols = ix["columns"]
if cand["source"] == "join" and ix["unique"] and cols == cand["columns"][:1] and not ix.get("partial"):
return ix # joining on a unique key: one row per lookup, extra columns add little
if cols[:len(cand["columns"])] == cand["columns"]:
return ix
n_eq = len(cand.get("eq", []))
if n_eq > 1 and len(cols) >= len(cand["columns"]) and set(cols[:n_eq]) == set(cand["columns"][:n_eq]) \
and cols[n_eq:len(cand["columns"])] == cand["columns"][n_eq:]:
return ix # same equality columns in another order serve the query equally well
return None
def served_by_named(c, indexes):
ix = served_by(c, indexes)
return {"existing": ix["name"] if ix else "?"}
def merge(cands):
merged = []
for c in sorted(cands, key=lambda c: -len(c["columns"])):
for m in merged:
if m["table"] == c["table"] and m["columns"][:len(c["columns"])] == c["columns"]:
m["queries"] = sorted(set(m["queries"]) | set(c["queries"]))
m["include"] = m["include"] or []
break
else:
merged.append(dict(c))
return sorted(merged, key=lambda c: (c["table"], c["columns"]))
def schema_findings(indexes, max_indexes, new_by_table):
out = []
by_table = {}
for ix in indexes:
by_table.setdefault(ix["table"], []).append(ix)
for t, ixs in by_table.items():
for i, a in enumerate(ixs):
for b in ixs[i + 1:]:
if a["columns"] == b["columns"] and (a.get("partial") == b.get("partial")):
out.append({"level": "high", "rule": "duplicate-index",
"message": f"{t}: {a['name']} and {b['name']} have the same columns ({', '.join(a['columns'])}); drop one"})
elif a["columns"] != b["columns"]:
short, long_ = (a, b) if len(a["columns"]) < len(b["columns"]) else (b, a)
if long_["columns"][:len(short["columns"])] == short["columns"] and not short["unique"] \
and not short.get("partial"):
out.append({"level": "warn", "rule": "redundant-prefix",
"message": f"{t}: {short['name']} ({', '.join(short['columns'])}) is a leftmost prefix of "
f"{long_['name']} ({', '.join(long_['columns'])}); usually droppable"})
for ix in ixs:
if len(ix["columns"]) == 1 and LOW_CARD.search(ix["columns"][0]) and ix["kind"] == "index" and not ix.get("partial"):
out.append({"level": "warn", "rule": "low-selectivity",
"message": f"{t}: {ix['name']} indexes only '{ix['columns'][0]}', which sounds like a flag or status "
"with few values; use a composite or partial index instead"})
total = len(ixs) + new_by_table.get(t, 0)
if total > max_indexes:
out.append({"level": "warn", "rule": "many-indexes",
"message": f"{t}: {len(ixs)} existing + {new_by_table.get(t, 0)} suggested = {total} indexes "
f"(limit {max_indexes}); every insert and update maintains all of them"})
return out
def ddl(c, dialect):
name = f"idx_{c['table']}_{'_'.join(c['columns'])}"[:63]
cols = ", ".join(c["columns"])
if c["include"]:
if dialect == "postgres":
return f"CREATE INDEX CONCURRENTLY {name} ON {c['table']} ({cols}) INCLUDE ({', '.join(c['include'])});"
cols += ", " + ", ".join(c["include"])
if dialect == "postgres":
return f"CREATE INDEX CONCURRENTLY {name} ON {c['table']} ({cols});"
if dialect == "mysql":
return f"ALTER TABLE {c['table']} ADD INDEX {name} ({cols}), ALGORITHM=INPLACE, LOCK=NONE;"
return f"CREATE INDEX {name} ON {c['table']} ({cols});"
# ---------------------------------------------------------------- main
def parse_args(argv):
opts = {"schema": None, "files": [], "dialect": "postgres", "max_indexes": 6, "covering": True,
"json": False, "strict": False}
i = 0
while i < len(argv):
a = argv[i]
if a in ("-h", "--help"):
print(__doc__)
sys.exit(0)
if a in ("--schema", "--dialect", "--max-indexes"):
if i + 1 >= len(argv):
raise UsageError(f"{a} needs a value")
v = argv[i + 1]
if a == "--schema":
opts["schema"] = v
elif a == "--dialect":
if v not in ("postgres", "mysql", "sqlite"):
raise UsageError("--dialect must be postgres, mysql, or sqlite")
opts["dialect"] = v
else:
try:
opts["max_indexes"] = int(v)
except ValueError:
raise UsageError(f"--max-indexes needs a whole number, got {v!r}")
i += 2
elif a == "--no-covering":
opts["covering"] = False; i += 1
elif a == "--json":
opts["json"] = True; i += 1
elif a == "--strict":
opts["strict"] = True; i += 1
elif a.startswith("--"):
raise UsageError(f"unknown option {a}")
else:
opts["files"].append(a); i += 1
if not opts["files"] and not opts["schema"]:
raise UsageError("give a query file (or - for stdin), optionally with --schema FILE")
return opts
def read(path):
if path == "-":
return sys.stdin.read()
with open(path, encoding="utf-8") as f:
return f.read()
def main(argv):
try:
opts = parse_args(argv)
schema_text = read(opts["schema"]) if opts["schema"] else ""
query_texts = [read(p) for p in opts["files"]]
except UsageError as e:
print(f"error: {e}", file=sys.stderr)
return 2
except OSError as e:
print(f"error: cannot read {e.filename}: {e.strerror}", file=sys.stderr)
return 2
statements = split_statements(strip_comments(schema_text))
query_stmts = []
for text in query_texts:
clean = strip_comments(text)
for st in split_statements(clean):
(statements if re.match(r"\s*create\s", st, re.I) else query_stmts).append(st)
tables, indexes = parse_schema(statements)
queries, n = [], 0
expanded = []
for st in query_stmts:
masked, _l = mask_strings(st)
if len(split_top(masked, r"\bunion(?:\s+all)?\b")) > 1: # a real top-level UNION, not one inside a string
expanded.extend(split_top(st, r"\bunion(?:\s+all)?\b")) # each branch is analyzed as its own query
else:
expanded.append(st)
for st in expanded:
pending = [(st, None, None)]
while pending:
cur, lits, outer = pending.pop(0)
n += 1
q = parse_query(cur, tables, n, lits, outer)
if q is None:
n -= 1
continue
queries.append(q)
for sub in q["subqueries"]:
pending.append((sub, q["lits"], {**(outer or {}), **q["aliases"]}))
q["notes"].append(f"subquery analyzed as Q{n + len(pending)}")
if not queries:
print("error: no SELECT, UPDATE, or DELETE statements found", file=sys.stderr)
return 2
pks = {ix["table"]: ix["columns"] for ix in indexes if ix["kind"] == "primary"}
cands = []
for q in queries:
cands += candidates_for(q, opts["dialect"], opts["covering"], pks)
covered, open_cands = [], []
for c in cands:
ix = served_by(c, indexes)
if ix:
covered.append({**c, "existing": ix["name"]})
else:
open_cands.append(c)
suggested = merge(open_cands)
covered = merge(covered)
for c in covered:
c["existing"] = next(x["existing"] for x in [served_by_named(c, indexes)])
new_by_table = {}
for c in suggested:
new_by_table[c["table"]] = new_by_table.get(c["table"], 0) + 1
c["ddl"] = ddl(c, opts["dialect"])
lead = c["columns"][0]
c["tradeoff"] = ("every INSERT, UPDATE of these columns, and DELETE on " + c["table"] +
" also writes this index; size grows with the row count and column widths")
if LOW_CARD.search(lead) and len(c["columns"]) == 1:
c["tradeoff"] += f"; '{lead}' looks low-selectivity, so the planner may still prefer a scan"
for a in suggested:
others = [k for k, b in enumerate(suggested, 1) if b is not a and b["table"] == a["table"]
and b["columns"][0] == a["columns"][0]]
a["overlaps"] = others
a["makes_redundant"] = [ix["name"] for ix in indexes if ix["table"] == a["table"] and not ix["unique"]
and ix["kind"] != "primary" and len(ix["columns"]) < len(a["columns"])
and a["columns"][:len(ix["columns"])] == ix["columns"]]
if a.get("null_filter") and opts["dialect"] == "postgres":
col = a["null_filter"][0]
rest = [c for c in a["columns"] if c != col]
name = f"idx_{a['table']}_{'_'.join(rest)}_{col}_null"[:63]
a["partial_hint"] = (f"CREATE INDEX CONCURRENTLY {name} ON {a['table']} ({', '.join(rest)}) "
f"WHERE {col} IS NULL; -- smaller if few rows match")
findings = schema_findings(indexes, opts["max_indexes"], new_by_table)
report = {"dialect": opts["dialect"], "tables": {t: len(v["columns"]) for t, v in tables.items()},
"existing_indexes": indexes, "queries": [{k: v for k, v in q.items() if k not in ("subqueries", "lits")}
for q in queries],
"suggested": suggested, "already_covered": covered, "schema_findings": findings}
for q in report["queries"]:
q["joins"] = [f"{a[0]}.{a[1]} = {b[0]}.{b[1]}" for a, b in q["joins"]]
strict_fail = opts["strict"] and (any(q["non_sargable"] for q in queries) or
any(f["rule"] == "duplicate-index" for f in findings))
if opts["json"]:
print(json.dumps(report, indent=2))
return 1 if strict_fail else 0
print(f"Schema: {len(tables)} table(s), {len(indexes)} existing index(es) incl. keys | dialect: {opts['dialect']}")
print(f"Queries analyzed: {len(queries)}")
for q in report["queries"]:
print(f"\nQ{q['n']} {q['kind']}: {q['text']}")
tabs = ", ".join(f"{t} {a}" if a != t else t for a, t in q["aliases"].items())
print(f" tables: {tabs or '-'}")
eqs = [f"{f['table'] or '?'}.{f['column']} ({f['op']})" for f in q["filters"] if f["kind"] == "eq"]
rngs = [f"{f['table'] or '?'}.{f['column']} ({f['op']})" for f in q["filters"] if f["kind"] == "range"]
if eqs:
print(f" equality: {', '.join(eqs)}")
if rngs:
print(f" range: {', '.join(rngs)}")
if q["joins"]:
print(f" joins: {', '.join(q['joins'])}")
if q["order_by"]:
print(" order by: " + ", ".join(f"{o['table']}.{o['column']} {o['dir'].upper()}" for o in q["order_by"])
+ (" (LIMIT)" if q["limit"] else ""))
if q["group_by"]:
print(" group by: " + ", ".join(f"{g['table']}.{g['column']}" for g in q["group_by"]))
for w in q["warnings"]:
print(f" WARN {w}")
for nt in q["notes"]:
print(f" note: {nt}")
print(f"\nSuggested indexes ({len(suggested)}):")
if not suggested:
print(" none: existing indexes already lead with the needed columns")
for k, c in enumerate(suggested, 1):
print(f" {k}. {c['ddl']}")
print(f" serves: {', '.join('Q' + str(x) for x in c['queries'])} | why: {c['why']}")
if c["include"]:
print(f" covering: {', '.join(c['include'])} added so the query can skip the table (optional)")
print(f" tradeoff: {c['tradeoff']}")
if c.get("overlaps"):
print(f" overlap: same leading column as #{', #'.join(map(str, c['overlaps']))}; "
"check whether one index can serve both before creating two")
if c.get("makes_redundant"):
print(f" makes redundant: {', '.join(c['makes_redundant'])} (leftmost prefix of this index; drop after checking usage)")
if c.get("partial_hint"):
print(f" alternative: {c['partial_hint']}")
if covered:
print(f"\nAlready served by existing indexes ({len(covered)}):")
for c in covered:
print(f" {c['table']} ({', '.join(c['columns'])}) -> {c['existing']} (Q{', Q'.join(map(str, c['queries']))})")
print(f"\nSchema findings ({len(findings)}):")
if not findings:
print(" none")
for f in findings:
print(f" {f['level'].upper():<4} {f['rule']:<17} {f['message']}")
print("\nNext: run EXPLAIN (ANALYZE, BUFFERS) or EXPLAIN on real data before and after; build one index at a time.")
return 1 if strict_fail else 0
if __name__ == "__main__":
sys.exit(main(sys.argv[1:]))Reads Conventional Commits or a changelog snippet, classifies fix/feat/breaking changes, and recommends the next SemVer bump (patch, minor, or major) with a clear rationale and a preview of the resulting version. Use when a user asks "what version should I ship?", "is this a breaking change?", or "bump SemVer from these commits". Includes a tested stdlib Python advisor.
---
name: semver-bump-advisor
description: Reads Conventional Commits or a changelog snippet, classifies fix/feat/breaking changes, and recommends the next SemVer bump (patch, minor, or major) with a clear rationale and a preview of the resulting version. Use when a user asks "what version should I ship?", "is this a breaking change?", or "bump SemVer from these commits". Includes a tested stdlib Python advisor.
---
# SemVer Bump Advisor
You help teams pick the next Semantic Version from real change signals: Conventional Commit subject lines, a short changelog, or an explicit list of change types. You separate patch (fixes, docs, chores that do not change the public API), minor (new backward-compatible features), and major (breaking changes), then show the resulting version string. You do not invent release notes the user did not imply.
## Files in this skill
- `scripts/semver_bump.py` - parses commits or a CHANGELOG-ish list, scores the highest bump required, and prints a rationale table (Python 3 standard library only)
- `references/conventional-commits-semver.md` - how commit types map to SemVer, including `BREAKING CHANGE` and `!`
- `references/edge-cases.md` - pre-1.0 rules, build metadata, multiple packages, and when not to bump
- `templates/bump-report.md` - the report shape to return
- `examples/example-library-release.md` - a worked example from a small library release
## Workflow
### 1. Collect the facts
Ask, or assume and say so:
- The current version (`MAJOR.MINOR.PATCH`, optional pre-release).
- The change signals: Conventional Commit lines, a changelog since the last tag, or labeled bullets (fix / feat / breaking).
- Whether the project is still `0.x` (different expectations) and whether public API means CLI flags, HTTP routes, or library exports.
### 2. Run the advisor
```bash
python3 scripts/semver_bump.py 1.4.2 --commits commits.txt
python3 scripts/semver_bump.py 0.9.1 --commits - --pre-1-0-major-as-minor
python3 scripts/semver_bump.py 2.0.0-rc.1 --json < changes.txt
```
A commits file has one subject per line (optional `type(scope):` prefix). Lines may include `BREAKING CHANGE:` footers on following indented lines, or a `!` after the type. Always show the tool output, then explain it in plain words.
### 3. Advise
1. State the highest bump required and why (quote the decisive commit or bullet).
2. Show the next version string, including how pre-release versions reset when graduating to a stable release if relevant.
3. Call out ambiguous lines (for example `refactor` that might be breaking) and ask the user to confirm API impact when needed.
4. Deliver with `templates/bump-report.md`, following `examples/example-library-release.md`.
## Rules
- Never skip a major bump when a breaking change is clearly present, unless the user explicitly opts into pre-1.0 soft-major handling.
- Do not invent commits; if the list is empty, say so and refuse to guess.
- Keep build metadata (`+...`) out of bump math; strip or preserve it only as the user requests.
- Prefer the Conventional Commits mapping in `references/conventional-commits-semver.md` over ad-hoc labels.
FILE:references/conventional-commits-semver.md
# Conventional Commits → SemVer
## Type mapping (default)
| Commit signal | SemVer bump |
|---|---|
| `fix`, `docs`, `style`, `test`, `chore`, `ci`, `perf` (non-breaking) | **patch** |
| `feat` (non-breaking) | **minor** |
| `BREAKING CHANGE` footer, or `type!:` / `type(scope)!:` | **major** |
| `refactor`, `build`, `revert` without breaking marker | **patch** (confirm if public API moved) |
When several commits appear, take the **maximum** bump: major > minor > patch.
## Breaking change markers
1. A footer line starting with `BREAKING CHANGE:` or `BREAKING-CHANGE:`.
2. `!` immediately before the colon: `feat!: ...` or `feat(api)!: ...`.
## Scopes
Scopes do not change the bump by themselves. `feat(cli): add --json` is still a minor. `fix(docs): typo` is still a patch.
## Pre-release versions
Given `1.2.0-rc.3`:
- A patch-level change while still in RC usually becomes `1.2.0-rc.4` if you stay on the RC channel.
- Graduating to stable after RC is a product decision: often `1.2.0` (not `1.2.1`) when RC already represents that minor.
This skill's script treats the numeric core `MAJOR.MINOR.PATCH` for bump math and prints pre-release handling notes separately.
FILE:references/edge-cases.md
# Edge cases
## Pre-1.0 (`0.x.y`)
SemVer allows breaking changes in `0.x` without a major bump, but many teams still use `feat` → minor and breaking → major for clarity. Offer both:
- Strict SemVer comfort: breaking in `0.x` may bump minor.
- Team policy: treat breaking as major even before 1.0 (`--pre-1-0-major-as-minor` off).
## Multiple packages
Monorepos may bump packages independently. Run the advisor once per package with that package's commits only.
## Ambiguous refactors
`refactor: rename export Foo to Bar` is breaking for libraries even without a `!`. Escalate to the user: if public API renamed → major; if internal only → patch.
## Empty or non-conventional lines
Plain sentences without a type still count if they clearly say "breaking" or "add feature"; otherwise mark them **unknown** and ask.
## Build metadata
`1.2.3+build.9` — bump the core version, then decide whether to drop or refresh `+build...`. Do not let metadata affect major/minor/patch math.
FILE:templates/bump-report.md
# SemVer bump report
## Current version
- …
## Recommended bump
- Level: patch | minor | major
- Next version: …
- Decisive change: …
## Change tally
| Level | Count | Examples |
|---|---|---|
| major | | |
| minor | | |
| patch | | |
| unknown | | |
## Notes and questions
- …
## Suggested tag / release command
```bash
# example
git tag -a vX.Y.Z -m "…"
```
FILE:examples/example-library-release.md
# Example: library release from Conventional Commits
**Current version:** `1.4.2`
**Commits since v1.4.2:**
```
fix: handle empty Accept headers
feat(api): add cursor pagination
docs: fix README typo
feat!: remove deprecated /v1/list route
```
**Advisor result:** major → `2.0.0` because of `feat!: remove deprecated /v1/list route`.
**Rationale:** The `!` marks a breaking API removal. The new pagination feature alone would have been a minor (`1.5.0`), and the fix/docs alone a patch (`1.4.3`), but major wins.
**Report excerpt:** Recommend releasing `2.0.0` with a migration note for `/v1/list` callers; mention cursor pagination as the headline feature of the major.
FILE:scripts/semver_bump.py
#!/usr/bin/env python3
"""Recommend a SemVer bump from Conventional Commit subjects or change lines.
Usage:
python3 semver_bump.py 1.4.2 --commits commits.txt
python3 semver_bump.py 0.9.1 --commits - --pre-1-0-major-as-minor
python3 semver_bump.py 2.0.0 --json < changes.txt
"""
from __future__ import annotations
import argparse
import json
import re
import sys
TYPE_PATCH = {"fix", "docs", "style", "test", "chore", "ci", "perf", "build", "refactor", "revert"}
TYPE_MINOR = {"feat"}
COMMIT_RE = re.compile(
r"^(?P<type>[a-zA-Z]+)(?P<scope>\([^)]*\))?(?P<breaking>!)?:\s*(?P<sub>.+)$"
)
VER_RE = re.compile(
r"^(?P<major>0|[1-9]\d*)\.(?P<minor>0|[1-9]\d*)\.(?P<patch>0|[1-9]\d*)"
r"(?:-(?P<pre>[0-9A-Za-z.-]+))?(?:\+(?P<build>[0-9A-Za-z.-]+))?$"
)
LEVELS = {"none": 0, "patch": 1, "minor": 2, "major": 3}
def parse_version(v: str) -> dict:
m = VER_RE.match(v.strip())
if not m:
raise SystemExit(f"error: not a SemVer core version: {v!r}")
return {k: m.group(k) for k in ("major", "minor", "patch", "pre", "build")}
def classify_line(line: str) -> tuple[str, str]:
raw = line.strip()
if not raw or raw.startswith("#"):
return "skip", raw
lower = raw.lower()
if lower.startswith("breaking change:") or lower.startswith("breaking-change:"):
return "major", raw
m = COMMIT_RE.match(raw)
if m:
if m.group("breaking"):
return "major", raw
t = m.group("type").lower()
if t in TYPE_MINOR:
return "minor", raw
if t in TYPE_PATCH:
return "patch", raw
return "unknown", raw
if "breaking" in lower:
return "major", raw
if lower.startswith("feat") or "new feature" in lower or lower.startswith("add "):
return "minor", raw
if lower.startswith("fix") or "bug" in lower:
return "patch", raw
return "unknown", raw
def read_lines(path: str) -> list[str]:
if path == "-":
return sys.stdin.read().splitlines()
with open(path, encoding="utf-8") as f:
return f.read().splitlines()
def bump(ver: dict, level: str, pre10_soft: bool) -> str:
major, minor, patch = int(ver["major"]), int(ver["minor"]), int(ver["patch"])
if level == "none":
core = f"{major}.{minor}.{patch}"
elif level == "major":
if pre10_soft and major == 0:
minor += 1
patch = 0
else:
major += 1
minor = 0
patch = 0
core = f"{major}.{minor}.{patch}"
elif level == "minor":
minor += 1
patch = 0
core = f"{major}.{minor}.{patch}"
else:
patch += 1
core = f"{major}.{minor}.{patch}"
return core
def main(argv: list[str]) -> int:
ap = argparse.ArgumentParser(description=__doc__)
ap.add_argument("version", help="Current SemVer, e.g. 1.4.2 or 0.9.1-rc.1")
ap.add_argument("--commits", default="-", help="File of commit subjects (default: stdin)")
ap.add_argument("--pre-1-0-major-as-minor", action="store_true",
help="In 0.x, treat breaking as minor bump")
ap.add_argument("--json", action="store_true", help="Machine-readable JSON output")
args = ap.parse_args(argv)
ver = parse_version(args.version)
lines = read_lines(args.commits)
tallies = {"major": [], "minor": [], "patch": [], "unknown": []}
best = "none"
for line in lines:
level, raw = classify_line(line)
if level == "skip":
continue
tallies[level].append(raw)
if LEVELS[level] > LEVELS[best]:
best = level
if best == "none":
print("error: no usable change lines", file=sys.stderr)
return 1
next_v = bump(ver, best, args.pre_1_0_major_as_minor)
decisive = (tallies[best][0] if tallies[best] else "")
note = None
if args.pre_1_0_major_as_minor and int(ver["major"]) == 0 and best == "major":
note = "pre-1.0 soft major: breaking treated as minor bump"
# best label stays major for tally, but version used soft path
result = {
"current": args.version,
"bump": best,
"next": next_v,
"decisive": decisive,
"tally": {k: len(v) for k, v in tallies.items()},
"examples": {k: v[:5] for k, v in tallies.items() if v},
"note": note,
}
if args.json:
json.dump(result, sys.stdout, indent=2)
sys.stdout.write("\n")
else:
print(f"current: {args.version}")
print(f"bump: {best}")
print(f"next: {next_v}")
if note:
print(f"note: {note}")
print(f"decisive:{decisive}")
print("tally:")
for k in ("major", "minor", "patch", "unknown"):
print(f" {k}: {len(tallies[k])}")
for ex in tallies[k][:5]:
print(f" - {ex}")
return 0
if __name__ == "__main__":
sys.exit(main(sys.argv[1:]))Picks the correct HTTP status code for an API response from a plain description of what happened (created, validation failed, not found, conflict, rate limited, upstream down) and explains why, with common mistakes to avoid. Use when designing REST endpoints, reviewing API error handling, or asking "should this be 400, 404, 409, or 422?". Includes a stdlib Python suggester.
---
name: http-status-code-advisor
description: Picks the correct HTTP status code for an API response from a plain description of what happened (created, validation failed, not found, conflict, rate limited, upstream down) and explains why, with common mistakes to avoid. Use when designing REST endpoints, reviewing API error handling, or asking "should this be 400, 404, 409, or 422?". Includes a stdlib Python suggester.
---
# HTTP Status Code Advisor
You help API designers and reviewers choose the HTTP status code that best matches what actually happened during a request. You reason from the request method, the outcome, and who is at fault (client, server, or upstream), and you prefer the most specific standard code that clients can act on. You follow RFC 9110 semantics and do not invent non-standard codes.
## Files in this skill
- `scripts/suggest_status.py` - maps a short scenario description (keywords plus method) to a recommended code, alternatives, and a one-line reason (Python 3 standard library only)
- `references/status-cheatsheet.md` - the codes that matter for APIs, grouped by class, with when to use each
- `references/common-mistakes.md` - frequent misuses (200 with an error body, 401 vs 403, 404 for empty lists, 500 for validation)
- `templates/decision-tree.md` - a step-by-step question flow to reach a code
- `examples/api-scenarios.md` - worked scenarios with the chosen code and response shape
## Workflow
### 1. Collect the facts
Ask for, or infer from the code or description:
- HTTP method and endpoint (e.g. `POST /orders`)
- What happened: success, partial success, async accepted, input problem, auth problem, missing resource, state conflict, rate limit, server or dependency failure
- Whether a response body is returned, and whether a new resource was created
- Any existing API conventions the team already follows
### 2. Walk the decision tree
Use `templates/decision-tree.md`. Decide the class first (2xx, 3xx, 4xx, 5xx), then the specific code. Who caused the failure decides 4xx vs 5xx.
### 3. Run the helper (optional)
```
python3 scripts/suggest_status.py --method POST "created new order"
python3 scripts/suggest_status.py --method GET "user not logged in"
python3 scripts/suggest_status.py --json "email already exists"
```
Treat the script as a first guess; confirm against the cheatsheet.
### 4. Check for mistakes
Compare the choice against `references/common-mistakes.md`, especially 401 vs 403, 400 vs 422, and 404 vs an empty 200 list.
### 5. Answer
Return:
1. **Recommended code** and reason phrase (e.g. `409 Conflict`)
2. **Why** in one or two sentences, tied to the facts
3. **Alternatives** considered and why they lose
4. **Headers to include** when relevant (`Location` for 201, `Retry-After` for 429/503, `WWW-Authenticate` for 401, `Allow` for 405)
5. **Error body sketch** using RFC 9457 Problem Details (`type`, `title`, `status`, `detail`, `instance`) for 4xx/5xx
## Rules
- Never return 2xx for a failed operation.
- Use 5xx only when the server or an upstream it depends on failed, never for bad client input.
- Keep consistency with the team's existing conventions if they are reasonable; mention the standard choice once if they differ.
- If security requires hiding a resource's existence, say so explicitly when recommending 404 over 403.
FILE:references/common-mistakes.md
# Common Status Code Mistakes
1. **200 with an error body** (`{"success": false}`). Clients, proxies, caches, and monitoring treat 200 as success. Use the right 4xx/5xx and put details in the body.
2. **401 vs 403 confusion.** 401 means "who are you?" (missing or bad credentials). 403 means "I know who you are, and the answer is no." A logged-in user without permission gets 403, not 401.
3. **500 for bad input.** Validation failures are the client's fault: 400 or 422. Reserve 500 for bugs and unexpected server failures. A spike in 500s should mean something is broken on your side.
4. **404 for an empty collection.** `GET /users?role=admin` with no matches is `200` with `[]`. 404 is for a missing resource such as `/users/9999`.
5. **400 for everything.** Prefer the specific code when it helps clients act: 409 for duplicates, 413 for oversize, 415 for wrong content type, 429 for rate limits.
6. **201 without `Location`.** When you create a resource with its own URL, send `Location: /orders/123`.
7. **204 with a body.** 204 must not include content. If you return the resource, use 200.
8. **302 in APIs.** Clients may change POST to GET. Use 307 or 308 when the method must be preserved.
9. **503 vs 500.** 503 means temporary and retryable (maintenance, overload), ideally with `Retry-After`. 500 means an unexpected error.
10. **Leaking existence with 403.** For private resources, returning 404 to unauthorized users can prevent enumeration. Decide deliberately and document it.
11. **Using 422 and 400 inconsistently.** Pick one rule (400 = cannot parse, 422 = parsed but invalid) and apply it on every endpoint.
12. **Non-standard codes** (e.g. 418 in production, 499, custom 6xx). Many clients and proxies handle them poorly. Stick to registered codes.
FILE:references/status-cheatsheet.md
# HTTP Status Cheatsheet for APIs
Based on RFC 9110 (HTTP Semantics), RFC 6585 (429), and RFC 9457 (Problem Details).
## 2xx Success
| Code | Name | Use when |
|---|---|---|
| 200 | OK | The request succeeded and the body contains the result (GET, or PUT/PATCH returning the updated resource). |
| 201 | Created | A new resource was created. Send `Location` with its URL; body usually contains the resource. |
| 202 | Accepted | Work was queued and will finish later. Return a status URL or job id. |
| 204 | No Content | Success with no body (DELETE, or PUT/PATCH when you return nothing). |
| 206 | Partial Content | Serving a byte range in response to a `Range` header. |
## 3xx Redirection
| Code | Name | Use when |
|---|---|---|
| 301 | Moved Permanently | Resource has a new permanent URL (method may change to GET on old clients). |
| 304 | Not Modified | Conditional GET (`If-None-Match` / `If-Modified-Since`) and nothing changed. |
| 307 | Temporary Redirect | Temporary move; client must repeat the same method and body. |
| 308 | Permanent Redirect | Permanent move; same method and body preserved. |
## 4xx Client errors
| Code | Name | Use when |
|---|---|---|
| 400 | Bad Request | Malformed syntax: invalid JSON, wrong content shape, missing required query param. |
| 401 | Unauthorized | No credentials or invalid/expired credentials. Send `WWW-Authenticate`. |
| 403 | Forbidden | Authenticated, but not allowed to do this. |
| 404 | Not Found | The resource does not exist (or you hide that it exists). |
| 405 | Method Not Allowed | The URL exists but not for this method. Send `Allow`. |
| 406 | Not Acceptable | Cannot produce any format listed in `Accept`. |
| 409 | Conflict | Request conflicts with current state: duplicate unique key, version conflict, invalid state transition. |
| 410 | Gone | Resource existed and was permanently removed. |
| 412 | Precondition Failed | `If-Match` / `If-Unmodified-Since` failed (optimistic locking). |
| 413 | Content Too Large | Body exceeds the size limit. |
| 415 | Unsupported Media Type | `Content-Type` not supported. |
| 422 | Unprocessable Content | Syntax is fine but values fail validation (bad email, end date before start date). |
| 428 | Precondition Required | Server requires a conditional request to prevent lost updates. |
| 429 | Too Many Requests | Rate limit exceeded. Send `Retry-After`. |
## 5xx Server errors
| Code | Name | Use when |
|---|---|---|
| 500 | Internal Server Error | Unexpected failure in your code. |
| 501 | Not Implemented | The server does not support this functionality at all. |
| 502 | Bad Gateway | An upstream returned an invalid response. |
| 503 | Service Unavailable | Temporarily overloaded or in maintenance. Send `Retry-After` if known. |
| 504 | Gateway Timeout | An upstream did not respond in time. |
FILE:templates/decision-tree.md
# Status Code Decision Tree
Answer in order. Stop at the first match.
1. **Did the request succeed?**
- Yes, work finished and a new resource exists -> **201** (+ `Location`)
- Yes, work accepted but finishes later -> **202** (+ status URL)
- Yes, nothing to return -> **204**
- Yes, returning data -> **200**
- No -> go to 2
2. **Is the client being redirected?**
- Conditional GET, unchanged -> **304**
- Moved, must keep the method -> **307** (temporary) / **308** (permanent)
- No -> go to 3
3. **Is it an authentication problem?**
- No credentials, or invalid / expired token -> **401** (+ `WWW-Authenticate`)
- No -> go to 4
4. **Is the authenticated caller not allowed?**
- Yes, and existence can be revealed -> **403**
- Yes, but existence must be hidden -> **404**
- No -> go to 5
5. **Is the target missing or the method wrong?**
- Resource does not exist -> **404**
- Resource removed permanently -> **410**
- URL exists, method unsupported -> **405** (+ `Allow`)
- No -> go to 6
6. **Is the request itself unusable?**
- Cannot parse (bad JSON, wrong types) -> **400**
- Wrong `Content-Type` -> **415**
- Body too large -> **413**
- Cannot satisfy `Accept` -> **406**
- No -> go to 7
7. **Is the input valid but rejected?**
- Fails business validation -> **422**
- Conflicts with current state (duplicate, version, state machine) -> **409**
- `If-Match` precondition failed -> **412**
- Too many requests -> **429** (+ `Retry-After`)
- No -> go to 8
8. **Server side failure**
- Upstream returned garbage -> **502**
- Upstream timed out -> **504**
- Temporarily unavailable / maintenance -> **503** (+ `Retry-After`)
- Feature not supported at all -> **501**
- Unexpected bug -> **500**
## Answer template
```
Recommended: <code> <reason phrase>
Why: <1-2 sentences>
Alternatives: <code> - <why not>; <code> - <why not>
Headers: <header: value>
Body (Problem Details):
{
"type": "https://example.com/problems/<slug>",
"title": "<short title>",
"status": <code>,
"detail": "<specific explanation>",
"instance": "<request path or id>"
}
```
FILE:examples/api-scenarios.md
# Worked API Scenarios
## 1. Sign-up with an email that already exists
`POST /users` with `{"email": "ana@example.com"}` and that email is taken.
- **Recommended:** 409 Conflict
- **Why:** The input is well-formed and valid, but it conflicts with existing state (a unique key).
- **Alternatives:** 422 (works if your team uses 422 for all rule failures, but 409 tells the client the specific problem is state); 400 (too vague).
- **Body:**
```json
{"type": "https://example.com/problems/email-taken", "title": "Email already registered", "status": 409, "detail": "An account with this email already exists.", "instance": "/users"}
```
## 2. Order created
`POST /orders` succeeds and returns the order.
- **Recommended:** 201 Created, `Location: /orders/8812`
- **Alternatives:** 200 (loses the "created" signal), 202 (only if fulfillment is async and the order is not yet persisted).
## 3. Report export queued
`POST /reports/exports` starts a 5-minute job.
- **Recommended:** 202 Accepted with `{"jobId": "exp_41", "statusUrl": "/reports/exports/exp_41"}`
## 4. Viewer tries to delete a project
Logged-in user with read-only role calls `DELETE /projects/7`.
- **Recommended:** 403 Forbidden
- **Why:** Identity is known; permission is missing. 401 would wrongly suggest logging in again.
## 5. Expired access token
`GET /me` with an expired JWT.
- **Recommended:** 401 Unauthorized, `WWW-Authenticate: Bearer error="invalid_token"`
## 6. Search with no matches
`GET /products?q=zzzz`
- **Recommended:** 200 OK with `{"items": [], "total": 0}`
- **Why:** The collection exists; it is just empty. 404 would break clients that treat it as an error.
## 7. Start date after end date
`POST /bookings` with `start=2026-10-20`, `end=2026-10-18`.
- **Recommended:** 422 Unprocessable Content
- **Alternatives:** 400 (acceptable if your convention merges parse and validation errors).
## 8. Payment provider timed out
`POST /checkout` and the payment gateway did not answer within 10 seconds.
- **Recommended:** 504 Gateway Timeout
- **Alternatives:** 502 (if the gateway answered with an invalid response), 503 (if you are deliberately shedding load).
## 9. Too many login attempts
- **Recommended:** 429 Too Many Requests, `Retry-After: 60`
## 10. Stale update
`PUT /docs/3` with `If-Match: "v4"` but the current version is `"v5"`.
- **Recommended:** 412 Precondition Failed
FILE:scripts/suggest_status.py
#!/usr/bin/env python3
"""Suggest an HTTP status code from a short scenario description.
Usage:
python3 suggest_status.py [--method METHOD] [--json] "scenario text"
python3 suggest_status.py --self-test
Keyword-based and deliberately simple: it gives a first guess plus
alternatives. Always confirm against references/status-cheatsheet.md.
Python 3 standard library only.
"""
import argparse
import json
import re
import sys
# Ordered rules: first match wins. Each rule: (keywords, code, reason, alternatives, headers)
RULES = [
(["rate limit", "too many", "throttl", "quota exceeded"], 429,
"Client exceeded a rate limit or quota.", [503], ["Retry-After"]),
(["not logged in", "no token", "missing token", "expired token", "invalid token",
"unauthenticated", "no credentials", "bad credentials", "login required"], 401,
"Credentials are missing or invalid.", [403], ["WWW-Authenticate"]),
(["no permission", "not allowed", "forbidden", "insufficient role", "not authorized",
"read-only user", "not the owner"], 403,
"Caller is authenticated but lacks permission.", [404], []),
(["already exists", "duplicate", "conflict", "version mismatch", "already taken",
"invalid state", "state transition"], 409,
"Request conflicts with the current state of the resource.", [422, 412], []),
(["precondition", "if-match", "etag mismatch", "stale"], 412,
"A conditional request header did not match.", [409], ["ETag"]),
(["too large", "payload size", "file too big", "body too big"], 413,
"Request body exceeds the allowed size.", [400], []),
(["unsupported media", "content-type", "wrong content type"], 415,
"Content-Type is not supported.", [400], ["Accept"]),
(["malformed", "invalid json", "parse error", "cannot parse", "syntax error",
"missing required"], 400,
"Request cannot be parsed or is missing required parts.", [422], []),
(["validation", "invalid email", "out of range", "must be", "before start",
"invalid value", "fails rule"], 422,
"Request is well-formed but values fail validation.", [400], []),
(["method not allowed", "wrong method", "unsupported method"], 405,
"The URL exists but does not support this method.", [404], ["Allow"]),
(["gone", "permanently deleted", "removed permanently"], 410,
"Resource existed and was permanently removed.", [404], []),
(["not found", "does not exist", "no such", "unknown id"], 404,
"Target resource does not exist.", [410], []),
(["upstream timeout", "gateway timeout", "timed out", "timeout"], 504,
"An upstream dependency did not respond in time.", [502, 503], []),
(["bad gateway", "upstream error", "invalid upstream", "upstream returned"], 502,
"An upstream dependency returned an invalid response.", [504], []),
(["maintenance", "overloaded", "unavailable", "down for"], 503,
"Service is temporarily unavailable.", [500], ["Retry-After"]),
(["not implemented", "unsupported feature"], 501,
"The server does not support this functionality.", [405], []),
(["crash", "exception", "bug", "unexpected error", "internal error"], 500,
"Unexpected server-side failure.", [503], []),
(["queued", "async", "background job", "accepted for processing", "will process"], 202,
"Work was accepted and will complete later.", [201], ["Location"]),
(["created", "new resource", "signed up", "registered"], 201,
"A new resource was created.", [200, 202], ["Location"]),
(["deleted", "no content", "nothing to return"], 204,
"Success with no response body.", [200], []),
(["not modified", "unchanged", "cache hit"], 304,
"Conditional request and resource is unchanged.", [200], ["ETag"]),
(["empty list", "no results", "no matches", "empty collection"], 200,
"Collection exists but is empty: return 200 with an empty list.", [404], []),
(["success", "ok", "returned", "fetched", "updated"], 200,
"Request succeeded and returns content.", [204], []),
]
REASONS = {
200: "OK", 201: "Created", 202: "Accepted", 204: "No Content", 304: "Not Modified",
400: "Bad Request", 401: "Unauthorized", 403: "Forbidden", 404: "Not Found",
405: "Method Not Allowed", 409: "Conflict", 410: "Gone", 412: "Precondition Failed",
413: "Content Too Large", 415: "Unsupported Media Type", 422: "Unprocessable Content",
429: "Too Many Requests", 500: "Internal Server Error", 501: "Not Implemented",
502: "Bad Gateway", 503: "Service Unavailable", 504: "Gateway Timeout",
}
def suggest(text, method=None):
t = " " + re.sub(r"\s+", " ", text.lower()) + " "
for keywords, code, reason, alts, headers in RULES:
for kw in keywords:
# word-boundary match for very short keywords like "ok"
if len(kw) <= 3:
if re.search(r"\b" + re.escape(kw) + r"\b", t):
break
elif kw in t:
break
else:
continue
m = (method or "").upper()
notes = []
if code == 200 and m == "DELETE":
code, reason, alts = 204, "DELETE succeeded; usually no body.", [200]
if code == 201 and m and m not in ("POST", "PUT"):
notes.append("201 is normally returned by POST or PUT.")
if code == 204 and m == "GET":
notes.append("GET rarely returns 204; prefer 200 with a body.")
return {
"code": code,
"reason_phrase": REASONS.get(code, ""),
"why": reason,
"alternatives": [f"{a} {REASONS.get(a, '')}".strip() for a in alts],
"headers": headers,
"matched_keyword": kw,
"notes": notes,
}
return {
"code": None,
"reason_phrase": "",
"why": "No rule matched. Walk templates/decision-tree.md manually.",
"alternatives": [],
"headers": [],
"matched_keyword": None,
"notes": [],
}
def self_test():
cases = [
("created new order", "POST", 201),
("user not logged in", "GET", 401),
("read-only user tries to delete", "DELETE", 403),
("email already exists", "POST", 409),
("search returned no results", "GET", 200),
("end date before start date validation", "POST", 422),
("invalid json body", "POST", 400),
("payment gateway timed out", "POST", 504),
("too many login attempts", "POST", 429),
("order id does not exist", "GET", 404),
("report export queued as background job", "POST", 202),
("record deleted", "DELETE", 204),
("unexpected exception in handler", "GET", 500),
("something odd happened", "GET", None),
]
failed = 0
for text, method, expected in cases:
got = suggest(text, method)["code"]
ok = got == expected
failed += not ok
print(f"{'PASS' if ok else 'FAIL'} {method:6} {text!r} -> {got} (expected {expected})")
print(f"\n{len(cases) - failed}/{len(cases)} passed")
return 1 if failed else 0
def main(argv=None):
p = argparse.ArgumentParser(description="Suggest an HTTP status code for an API scenario.")
p.add_argument("scenario", nargs="*", help="Short description of what happened")
p.add_argument("--method", help="HTTP method, e.g. POST")
p.add_argument("--json", action="store_true", help="Print JSON output")
p.add_argument("--self-test", action="store_true", help="Run built-in tests")
a = p.parse_args(argv)
if a.self_test:
return self_test()
if not a.scenario:
p.error("scenario text is required (or use --self-test)")
r = suggest(" ".join(a.scenario), a.method)
if a.json:
print(json.dumps(r, indent=2))
return 0
if r["code"] is None:
print(r["why"])
return 2
print(f"Recommended: {r['code']} {r['reason_phrase']}")
print(f"Why: {r['why']}")
if r["alternatives"]:
print("Alternatives: " + ", ".join(r["alternatives"]))
if r["headers"]:
print("Headers: " + ", ".join(r["headers"]))
for n in r["notes"]:
print("Note: " + n)
return 0
if __name__ == "__main__":
sys.exit(main())Audits the scripts section of an npm package.json for safety and clarity: risky install hooks, curl-pipe-to-shell, rm -rf, sudo, inline secrets, unpinned npx, non-portable shell syntax, placeholder test scripts, missing standard scripts, and confusing names. Use when reviewing a repo, a dependency, or a PR that changes package.json scripts. Includes a stdlib Python auditor with severity levels.
---
name: package-json-scripts-auditor
description: Audits the scripts section of an npm package.json for safety and clarity: risky install hooks, curl-pipe-to-shell, rm -rf, sudo, inline secrets, unpinned npx, non-portable shell syntax, placeholder test scripts, missing standard scripts, and confusing names. Use when reviewing a repo, a dependency, or a PR that changes package.json scripts. Includes a stdlib Python auditor with severity levels.
---
# package.json Scripts Auditor
You review the `scripts` block of a Node.js `package.json` and explain what each script does, which ones are risky, and how to make the set clearer and more portable. You cover two audiences: maintainers tidying their own project, and reviewers checking whether a dependency or PR runs anything dangerous at install time. You never run the scripts yourself; you read them.
## Files in this skill
- `scripts/audit_scripts.py` - reads a package.json and reports findings with severity (high, medium, low, info) and a suggested fix (Python 3 standard library only)
- `references/script-conventions.md` - standard script names, lifecycle hooks, naming, and portability conventions
- `references/dangerous-patterns.md` - patterns that deserve a closer look, why they matter, and safer alternatives
- `templates/audit-report.md` - the report shape to return
- `examples/package.json` - a deliberately messy example to practice on
## Workflow
### 1. Collect context
Ask for the full `package.json` (not only `scripts`, since `name`, `private`, `bin`, `workspaces`, and `packageManager` matter), and whether this is the user's own project or a third-party package. Ask which OS the team uses (Windows support changes portability advice).
### 2. Run the auditor
```
python3 scripts/audit_scripts.py path/to/package.json
python3 scripts/audit_scripts.py --json path/to/package.json
python3 scripts/audit_scripts.py --self-test
```
Exit code 0 = no high findings, 1 = at least one high finding, 2 = file error.
### 3. Review manually
The script is pattern-based. Also read each script yourself and check:
- Lifecycle hooks (`preinstall`, `install`, `postinstall`, `prepare`) run automatically for consumers; any network call or file write outside the project there is a red flag.
- Scripts that call other files (`node scripts/setup.js`, `sh ./build.sh`): ask to see those files before calling the package safe.
- Obfuscation (base64, hex strings, `eval`, `node -e` with long one-liners) deserves high suspicion.
### 4. Recommend
Use `references/script-conventions.md` and `references/dangerous-patterns.md` to propose concrete replacements: `rimraf` or `del-cli` instead of `rm -rf`, `cross-env` instead of `VAR=value cmd`, pinned `npx pkg@1.2.3`, `prepublishOnly` running tests, and grouped names like `lint:js`, `test:unit`.
### 5. Report
Fill `templates/audit-report.md`: verdict, findings table, a rewritten `scripts` block, and follow-up questions.
## Rules
- Rate severity by impact: anything that can run remote code or delete files outside the project on install is high.
- Do not accuse authors of malice; describe the behavior and the risk.
- Keep the user's existing script names when renaming would break CI, and say which CI files would need updating if you do suggest a rename.
- Suggest `npm install --ignore-scripts` for auditing untrusted packages, and `npm config set ignore-scripts true` only with the trade-offs explained.
FILE:references/dangerous-patterns.md
# Dangerous and Risky Script Patterns
| Pattern | Example | Why it matters | Safer alternative | Severity |
|---|---|---|---|---|
| Remote script piped to a shell | `curl -fsSL https://x.sh \| sh` | Runs whatever the server returns, unpinned and unverified | Vendor the script, pin by checksum, or use a published package | high |
| Network fetch in install hooks | `"postinstall": "node download.js"` / `wget` | Runs on every consumer's machine at install time | Ship artifacts in the package, or make the step opt-in | high |
| Recursive delete of broad paths | `rm -rf /`, `rm -rf ~`, `rm -rf $DIR/` with unset var | Can wipe data outside the project | `rimraf dist`, explicit relative paths | high |
| `sudo` | `sudo npm i -g x` | Escalates privileges from a script | Never require root in scripts; document manual steps | high |
| `eval` / encoded payloads | `eval "$(echo ... \| base64 -d)"`, `node -e "...Buffer.from('..','base64')"` | Hides what runs | Plain, readable commands in files | high |
| Inline secrets | `API_TOKEN=abc123 node deploy.js` | Leaks in git history and CI logs | Environment variables from a secret store or `.env` not committed | high |
| World-writable permissions | `chmod 777`, `chmod -R a+w` | Lets any user modify files | Minimal permissions (`chmod +x file`) | medium |
| Skipping safety checks | `git commit --no-verify`, `npm publish --force`, `--unsafe-perm` | Bypasses hooks and protections | Fix the failing check instead | medium |
| Unpinned `npx` / `dlx` | `npx create-thing` | Downloads and runs the latest version at run time | `npx create-thing@3.2.1` or add as devDependency | medium |
| Global installs in scripts | `npm i -g typescript` | Changes the user's machine; versions drift | devDependency + local bin | medium |
| Plain HTTP URLs | `curl http://...` | Can be tampered with in transit | HTTPS | medium |
| `git push` / `npm publish` in hooks | `"postbuild": "npm publish"` | Accidental releases | Explicit `release` script run by CI | medium |
| Non-portable shell syntax | `NODE_ENV=prod`, `export`, `cp -r` | Breaks on Windows | `cross-env`, `cpy-cli`, Node scripts | low |
| Placeholder test | `echo "Error: no test specified" && exit 1` | CI fails or tests silently missing | Real test runner | low |
| Very long one-liners | > 200 chars | Hard to review; hides risk | Move to a script file | low |
## Review tips
- For third-party packages: inspect with `npm pack <pkg>` and read the tarball, or install with `--ignore-scripts` first.
- Follow file references: `node scripts/postinstall.js` is only as safe as that file.
- Watch for `prepare` in git dependencies: it runs when installing from a git URL.
FILE:references/script-conventions.md
# npm Script Conventions
## Standard names
| Script | Purpose | Run with |
|---|---|---|
| `start` | Run the app in production mode | `npm start` |
| `dev` | Run with watch / hot reload | `npm run dev` |
| `build` | Produce build output | `npm run build` |
| `test` | Run the test suite (CI default) | `npm test` |
| `lint` | Static checks | `npm run lint` |
| `format` | Apply formatter | `npm run format` |
| `typecheck` | `tsc --noEmit` or similar | `npm run typecheck` |
| `clean` | Remove build artifacts | `npm run clean` |
Use a colon to group variants: `test:unit`, `test:e2e`, `lint:css`, `build:docs`. Use `npm-run-all` / `run-p` / `run-s`, or `npm run a && npm run b`, to compose them.
## Lifecycle hooks
| Hook | Runs when | Notes |
|---|---|---|
| `preinstall`, `install`, `postinstall` | Your package is installed, including as a dependency | Runs on every consumer's machine. Keep empty unless compiling native code. |
| `prepare` | After local `npm install` and before `npm publish`/`npm pack`; also for git dependencies | Good for building or setting up git hooks (e.g. `husky`). |
| `prepublishOnly` | Only before `npm publish` | Put `npm test && npm run build` here. |
| `prepack`, `postpack` | Around `npm pack` / publish | |
| `preX` / `postX` | Automatically before/after any script `X` | Can surprise readers; prefer explicit chaining. pnpm does not run them by default. |
## Portability
- Windows `cmd.exe` does not understand `VAR=value cmd`, `export`, `rm -rf`, `cp -r`, `&` backgrounding, single quotes, or `$VAR`. Use `cross-env`, `rimraf`, `cpy-cli`, `concurrently`, and Node scripts.
- `&&` works in both shells; `;` does not run the same way in `cmd.exe`.
- Prefer local binaries (from `devDependencies`) over globally installed ones; npm adds `node_modules/.bin` to PATH inside scripts.
## Clarity
- One job per script; move long logic (> ~120 characters) into `scripts/*.js` or `*.mjs` files.
- Avoid names that shadow npm commands or each other (`install`, `publish`, `test2`).
- Replace the default `echo "Error: no test specified" && exit 1` with a real test command or remove it.
- Document unusual scripts in the README.
- Set `"private": true` for apps that must never be published.
FILE:templates/audit-report.md
# package.json Scripts Audit
**Package:** `<name>@<version>` | **Owner:** own project / third-party | **Private:** yes/no
**Verdict:** Safe / Needs changes / Do not install until reviewed
**Counts:** high `<n>`, medium `<n>`, low `<n>`, info `<n>`
## Script inventory
| Script | What it does (plain words) | Runs automatically? |
|---|---|---|
| `build` | | no |
| `postinstall` | | yes, on every install |
## Findings
| # | Severity | Script | Pattern | Why it matters | Fix |
|---|---|---|---|---|---|
| 1 | high | | | | |
## Suggested scripts block
```json
{
"scripts": {
}
}
```
## Also check
- Files referenced by scripts: `<list>`
- CI files that call renamed scripts: `<list>`
- devDependencies to add: `<rimraf, cross-env, ...>`
## Questions for the maintainer
1.
FILE:examples/package.json
{
"name": "acme-dashboard",
"version": "0.4.2",
"description": "Internal dashboard (example with deliberately messy scripts for auditing practice)",
"main": "dist/index.js",
"scripts": {
"preinstall": "curl -fsSL http://get.example.com/setup.sh | sh",
"postinstall": "node scripts/fetch-assets.js",
"start": "NODE_ENV=production node dist/index.js",
"dev": "nodemon src/index.ts",
"build": "rm -rf dist && tsc -p .",
"test": "echo \"Error: no test specified\" && exit 1",
"deploy": "API_TOKEN=sk_live_example123 node scripts/deploy.js",
"fix-perms": "chmod -R 777 ./uploads",
"commit": "git add -A && git commit --no-verify -m wip",
"gen": "npx create-acme-widget",
"postbuild": "npm publish",
"setup": "sudo npm install -g typescript"
},
"devDependencies": {
"nodemon": "^3.1.0",
"typescript": "^5.6.0"
}
}
FILE:scripts/audit_scripts.py
#!/usr/bin/env python3
"""Audit the "scripts" section of an npm package.json.
Usage:
python3 audit_scripts.py path/to/package.json [--json]
python3 audit_scripts.py --self-test
Exit codes: 0 = no high findings, 1 = high findings present, 2 = file error.
Pattern-based: always read the scripts and any files they call yourself.
Python 3 standard library only.
"""
import argparse
import json
import re
import sys
INSTALL_HOOKS = {"preinstall", "install", "postinstall", "prepare"}
SEVERITY_ORDER = {"high": 0, "medium": 1, "low": 2, "info": 3}
# (id, regex, severity, message, fix)
PATTERNS = [
("pipe-to-shell", r"\b(curl|wget)\b[^|]*\|\s*(sudo\s+)?(ba|z|da)?sh\b", "high",
"Downloads a remote script and pipes it into a shell.",
"Vendor the script, verify a checksum, or use a pinned package."),
("broad-rm", r"\brm\s+-[a-z]*r[a-z]*f?[a-z]*\s+(/|~|\$HOME|\$\{?\w+\}?/?)(\s|$|&|;)", "high",
"Recursive delete of a root, home, or variable-based path.",
"Delete an explicit relative path, e.g. `rimraf dist`."),
("sudo", r"(^|[\s;&|])sudo\s", "high",
"Uses sudo inside an npm script.",
"Never require root from scripts; document any manual setup."),
("eval-or-encoded", r"\beval\b|base64\s+(-d|--decode)|Buffer\.from\([^)]*base64", "high",
"Uses eval or decodes an encoded payload, hiding what runs.",
"Replace with plain, readable commands in a script file."),
("inline-secret", r"\b[A-Z0-9_]*(TOKEN|SECRET|PASSWORD|PASSWD|API_KEY|APIKEY|PRIVATE_KEY)[A-Z0-9_]*=(?!\$)\S+", "high",
"Hard-coded secret value in a script.",
"Read secrets from the environment or a secret manager; rotate the exposed value."),
("chmod-777", r"\bchmod\s+(-R\s+)?(0?777|a\+w|o\+w)\b", "medium",
"Makes files world-writable.",
"Use minimal permissions, e.g. `chmod +x bin/cli.js`."),
("skip-checks", r"--no-verify|--force\b|--unsafe-perm|--ignore-engines", "medium",
"Bypasses safety checks or hooks.",
"Fix the underlying failure instead of skipping the check."),
("unpinned-npx", r"\b(npx|pnpm\s+dlx|bunx|yarn\s+dlx)\s+(-y\s+|--yes\s+)?(?!-)(@?[\w./-]+)(?=\s|$)(?<!@latest)", "medium",
"Runs a package via npx/dlx without a pinned version.",
"Pin it (`npx pkg@1.2.3`) or add it to devDependencies."),
("global-install", r"\bnpm\s+(i|install)\s+(-g|--global)\b|\byarn\s+global\s+add\b", "medium",
"Installs packages globally from a script.",
"Add as a devDependency and use the local binary."),
("plain-http", r"http://(?!localhost|127\.0\.0\.1)", "medium",
"Fetches over plain HTTP.",
"Use HTTPS."),
("publish-or-push", r"\bnpm\s+publish\b|\bgit\s+push\b", "medium",
"Publishes or pushes from a script.",
"Keep releases in an explicit `release` script or CI job."),
("non-portable-env", r"(^|&&\s*|;\s*)[A-Z_][A-Z0-9_]*=\S+\s+\S", "low",
"Inline env assignment `VAR=value cmd` does not work in Windows cmd.",
"Use `cross-env VAR=value cmd`."),
("non-portable-cmd", r"(^|[\s;&|])(rm\s+-|cp\s+-r|export\s+\w+=|mv\s)", "low",
"Unix-only shell command.",
"Use rimraf / cpy-cli / a Node script for Windows support."),
]
PLACEHOLDER_TEST = re.compile(r'no test specified', re.I)
NETWORK = re.compile(r"\b(curl|wget|git\s+clone|Invoke-WebRequest)\b|https?://", re.I)
FILE_REF = re.compile(r"\b(?:node|sh|bash|python3?)\s+(\.?/?[\w./-]+\.(?:m?js|cjs|ts|sh|py))")
def audit(pkg):
findings = []
def add(sev, script, rule, msg, fix):
findings.append({"severity": sev, "script": script, "rule": rule, "message": msg, "fix": fix})
scripts = pkg.get("scripts")
if not isinstance(scripts, dict) or not scripts:
add("info", None, "no-scripts", "No scripts section found.", "Add standard scripts such as build, test, lint.")
return findings
for name, cmd in scripts.items():
if not isinstance(cmd, str):
add("medium", name, "not-string", "Script value is not a string.", "Make it a shell command string.")
continue
for rule, rx, sev, msg, fix in PATTERNS:
if re.search(rx, cmd):
add(sev, name, rule, msg, fix)
if name in INSTALL_HOOKS:
if name == "prepare":
add("info", name, "install-hook", "`prepare` runs on local install and for git dependencies.",
"Keep it limited to build or git-hook setup.")
else:
add("medium", name, "install-hook", f"`{name}` runs automatically on every consumer's install.",
"Remove it unless it compiles native code; move setup to an explicit script.")
if NETWORK.search(cmd):
add("high", name, "network-in-install-hook", "Install hook performs network access.",
"Ship assets in the package or make the download opt-in.")
for ref in FILE_REF.findall(cmd):
add("high" if name != "prepare" else "medium", name, "install-hook-runs-file",
f"Install hook executes `{ref}`; its contents decide the real risk.",
f"Read `{ref}` before trusting this package.")
if name == "test" and PLACEHOLDER_TEST.search(cmd):
add("low", name, "placeholder-test", "Default placeholder test script.", "Wire up a real test runner.")
if len(cmd) > 200:
add("low", name, "long-script", f"Script is {len(cmd)} characters long.", "Move the logic into a script file.")
if name.startswith(("pre", "post")) and name not in INSTALL_HOOKS and name not in ("prepublishOnly", "prepack", "postpack"):
base = name[3:] if name.startswith("pre") else name[4:]
if base in scripts:
add("info", name, "implicit-hook", f"Runs automatically around `{base}` (pnpm skips these by default).",
f"Consider chaining explicitly inside `{base}`.")
if not re.match(r"^[a-z0-9][a-z0-9:._-]*$", name):
add("low", name, "naming", "Script name uses unusual characters or uppercase.",
"Use lowercase names grouped with colons, e.g. `test:unit`.")
for std in ("build", "test", "lint"):
if std not in scripts and not any(k.startswith(std + ":") for k in scripts):
add("info", None, "missing-standard", f"No `{std}` script.", f"Add a `{std}` script if the project needs one.")
if not pkg.get("private") and "prepublishOnly" not in scripts and "publish-or-push" not in [f["rule"] for f in findings]:
add("info", None, "no-prepublishonly", "Package is publishable but has no `prepublishOnly` check.",
"Add `\"prepublishOnly\": \"npm test && npm run build\"` or set `\"private\": true`.")
findings.sort(key=lambda f: (SEVERITY_ORDER[f["severity"]], f["script"] or ""))
return findings
def self_test():
pkg = {
"name": "t", "scripts": {
"preinstall": "curl -fsSL http://x.example/s.sh | sh",
"postinstall": "node scripts/setup.js",
"clean": "rm -rf $OUT/",
"deploy": "API_TOKEN=abc123 node deploy.js",
"perm": "chmod -R 777 uploads",
"gen": "npx create-foo",
"gen2": "npx create-foo@1.2.3",
"start": "NODE_ENV=production node index.js",
"test": "echo \"Error: no test specified\" && exit 1",
"setup": "sudo npm i -g typescript",
"build": "tsc",
}}
f = audit(pkg)
rules = {(x["script"], x["rule"]) for x in f}
checks = [
(("preinstall", "pipe-to-shell"), True),
(("preinstall", "network-in-install-hook"), True),
(("preinstall", "plain-http"), True),
(("postinstall", "install-hook-runs-file"), True),
(("clean", "broad-rm"), True),
(("deploy", "inline-secret"), True),
(("perm", "chmod-777"), True),
(("gen", "unpinned-npx"), True),
(("gen2", "unpinned-npx"), False),
(("start", "non-portable-env"), True),
(("test", "placeholder-test"), True),
(("setup", "sudo"), True),
(("setup", "global-install"), True),
(("build", "broad-rm"), False),
]
failed = 0
for key, expected in checks:
ok = (key in rules) == expected
failed += not ok
print(f"{'PASS' if ok else 'FAIL'} {key[0]}:{key[1]} expected={'present' if expected else 'absent'}")
clean = audit({"name": "c", "private": True, "scripts": {"build": "tsc -p .", "test": "vitest run", "lint": "eslint ."}})
ok = not [x for x in clean if x["severity"] in ("high", "medium", "low")]
failed += not ok
print(f"{'PASS' if ok else 'FAIL'} clean package has no high/medium/low findings")
total = len(checks) + 1
print(f"\n{total - failed}/{total} passed")
return 1 if failed else 0
def main(argv=None):
p = argparse.ArgumentParser(description="Audit npm package.json scripts.")
p.add_argument("file", nargs="?", help="Path to package.json")
p.add_argument("--json", action="store_true", help="Output JSON")
p.add_argument("--self-test", action="store_true", help="Run built-in tests")
a = p.parse_args(argv)
if a.self_test:
return self_test()
if not a.file:
p.error("file is required (or use --self-test)")
try:
with open(a.file, encoding="utf-8") as fh:
pkg = json.load(fh)
except (OSError, json.JSONDecodeError) as exc:
print(f"Cannot read {a.file}: {exc}", file=sys.stderr)
return 2
findings = audit(pkg)
counts = {s: sum(1 for f in findings if f["severity"] == s) for s in SEVERITY_ORDER}
if a.json:
print(json.dumps({"package": pkg.get("name"), "counts": counts, "findings": findings}, indent=2))
else:
print(f"Package: {pkg.get('name', '?')}@{pkg.get('version', '?')}")
for f in findings:
where = f["script"] or "-"
print(f"[{f['severity'].upper():6}] {where:12} {f['rule']}: {f['message']}\n{'':22}fix: {f['fix']}")
print("\n" + ", ".join(f"{k} {v}" for k, v in counts.items()))
return 1 if counts["high"] else 0
if __name__ == "__main__":
sys.exit(main())Compares an OpenAPI YAML/JSON spec to a list of implemented routes and flags drift: missing operations, extra routes, method mismatches, and path-parameter shape differences. Use when reviewing API PRs, before a release, or asking "does our code still match the OpenAPI?". Includes a stdlib Python checker with --self-test.
---
name: openapi-spec-drift-checker
description: Compares an OpenAPI YAML/JSON spec to a list of implemented routes and flags drift: missing operations, extra routes, method mismatches, and path-parameter shape differences. Use when reviewing API PRs, before a release, or asking "does our code still match the OpenAPI?". Includes a stdlib Python checker with --self-test.
---
# OpenAPI Spec Drift Checker
You help API teams find drift between an OpenAPI 3.x document and the routes their code actually serves. You compare paths and methods, flag missing or extra operations, and call out parameter or response mismatches that break clients. You do not invent endpoints; you only report what the inputs show.
## Files in this skill
- `scripts/check_drift.py` - compares a small OpenAPI snippet to a route list and prints a drift report (Python 3 standard library only)
- `references/openapi-checklist.md` - what to verify in the spec before comparing
- `references/common-drift.md` - frequent causes of drift and how to fix them
- `templates/drift-report.md` - report structure for PRs and release notes
- `examples/openapi-snippet.yaml` - sample OpenAPI fragment for demos
## Workflow
### 1. Collect inputs
Ask for, or infer:
- OpenAPI file (YAML or JSON) or a pasted paths section
- Implemented routes: method + path list from the framework router, generated from code, or a gateway config
- Whether path-parameter names must match exactly (`{id}` vs `{userId}`)
### 2. Normalize
Lowercase methods. Strip trailing slashes. Treat `/users/{id}` and `/users/{userId}` as the same path shape unless the team requires exact parameter names.
### 3. Run the checker
```
python3 scripts/check_drift.py --spec examples/openapi-snippet.yaml --routes examples/routes.txt
python3 scripts/check_drift.py --self-test
```
If routes are not in a file yet, help the user extract them from code first.
### 4. Classify findings
- **missing_in_code**: in OpenAPI, not implemented
- **missing_in_spec**: implemented, not in OpenAPI
- **method_mismatch**: path exists but verbs differ
- **param_shape_diff**: same path pattern, different parameter names (optional strict mode)
### 5. Answer
Return a short summary counts table, then each finding with severity (high/medium/low) and a suggested fix. Use `templates/drift-report.md` for PR comments.
## Rules
- Prefer documenting reality when the code is intentional and the spec is stale; prefer implementing when the spec is the contract already published to clients.
- Never claim security review was done; only report structural drift.
- If the OpenAPI cannot be parsed, stop and show the parse error.
FILE:references/openapi-checklist.md
# OpenAPI pre-compare checklist
Before running drift checks, confirm the spec is coherent:
1. **openapi** field is `3.0.x` or `3.1.x`.
2. **paths** exist and use leading `/`.
3. Every operation has an HTTP method (`get`, `post`, `put`, `patch`, `delete`, `head`, `options`).
4. Path parameters declared in the path (`{id}`) also appear under `parameters` with `in: path` and `required: true`.
5. `$ref` targets resolve (or you are comparing a resolved/bundled document).
6. Servers URL base path is understood: if `servers[].url` includes `/api/v1`, routes may be listed with or without that prefix—pick one convention and stick to it for the compare.
7. Deprecated operations are still listed until removed from code; mark them in the report rather than ignoring them.
## Minimal paths shape
```yaml
paths:
/pets:
get:
summary: List pets
post:
summary: Create pet
/pets/{petId}:
get:
parameters:
- name: petId
in: path
required: true
schema: { type: string }
```
FILE:references/common-drift.md
# Common OpenAPI drift
| Symptom | Likely cause | Fix |
| --- | --- | --- |
| Spec has `GET /users`, code has none | Spec written ahead of implementation | Implement or remove from published spec |
| Code has `POST /users/{id}/reset`, missing in spec | Hotfix shipped without doc update | Add operation; bump minor if public |
| Spec `PUT`, code `PATCH` | Partial-update semantics changed | Align method; note breaking if clients call PUT |
| `{userId}` vs `{id}` | Framework auto-names vs hand-written spec | Normalize in compare; document naming rule |
| Trailing slash only in code | Router is slash-tolerant | Strip for compare; pick canonical form in spec |
| Base path `/api` only in gateway | Spec servers omit gateway prefix | Strip or add the same prefix on both sides |
| Webhooks / callbacks only in spec | Not HTTP routes | Exclude from route compare |
| File upload route undocumented | Multipart added late | Document `multipart/form-data` requestBody |
## Severity guide
- **High**: public clients already call the operation; method or path missing on either side.
- **Medium**: internal-only route missing from spec; param rename with aliases.
- **Low**: trailing slash, optional HEAD/OPTIONS noise, deprecated-but-present.
FILE:templates/drift-report.md
# OpenAPI drift report
- **Spec**: <path or URL>
- **Routes source**: <router dump / gateway / manual>
- **Compared at**: <ISO timestamp>
- **Summary**: missing_in_code=<n>, missing_in_spec=<n>, method_mismatch=<n>, param_shape_diff=<n>
## Findings
| Severity | Kind | Spec | Code | Notes |
| --- | --- | --- | --- | --- |
| high | missing_in_code | `GET /pets/{petId}` | — | Implement or unpublish |
| medium | missing_in_spec | — | `POST /pets/{petId}/adopt` | Document before release |
## Decision
- [ ] Update OpenAPI
- [ ] Update code
- [ ] Accept as known exception (list why)
## Follow-ups
- [ ]
FILE:examples/openapi-snippet.yaml
openapi: 3.0.3
info:
title: Pets demo
version: 0.1.0
paths:
/pets:
get:
summary: List pets
operationId: listPets
post:
summary: Create pet
operationId: createPet
/pets/{petId}:
get:
summary: Get pet by id
operationId: getPet
parameters:
- name: petId
in: path
required: true
schema:
type: string
delete:
summary: Delete pet
operationId: deletePet
parameters:
- name: petId
in: path
required: true
schema:
type: string
/health:
get:
summary: Health check
operationId: health
FILE:scripts/check_drift.py
#!/usr/bin/env python3
"""Compare OpenAPI paths to an implemented route list.
Usage:
python3 check_drift.py --spec openapi.yaml --routes routes.txt
python3 check_drift.py --spec openapi.yaml --route 'GET /pets' --route 'POST /pets'
python3 check_drift.py --self-test
Route file format: one "METHOD /path" per line. Comments (#) and blank lines ignored.
Python 3 standard library only.
"""
from __future__ import annotations
import argparse
import json
import re
import sys
from typing import Dict, List, Set, Tuple
try:
import yaml # type: ignore
except ImportError:
yaml = None
PathKey = str # normalized path template
Method = str
def normalize_path(path: str) -> str:
path = path.strip()
if not path.startswith("/"):
path = "/" + path
if len(path) > 1 and path.endswith("/"):
path = path.rstrip("/")
# unify parameter names to {}
return re.sub(r"\{[^}]+\}", "{}", path)
def normalize_method(m: str) -> str:
return m.strip().upper()
def parse_routes(lines: List[str]) -> Dict[PathKey, Set[Method]]:
out: Dict[PathKey, Set[Method]] = {}
for raw in lines:
line = raw.split("#", 1)[0].strip()
if not line:
continue
parts = line.split()
if len(parts) < 2:
raise ValueError(f"Bad route line: {raw!r}")
method, path = normalize_method(parts[0]), normalize_path(parts[1])
out.setdefault(path, set()).add(method)
return out
def _parse_openapi_json(text: str) -> dict:
return json.loads(text)
def _parse_openapi_yaml_simple(text: str) -> dict:
"""Minimal YAML subset parser for simple OpenAPI path maps when PyYAML is absent.
Supports the examples shipped with this skill (no anchors, no multiline).
Prefer PyYAML when installed.
"""
# Very small fallback: extract path keys and methods via regex for our demos.
paths: Dict[str, dict] = {}
current_path = None
for raw in text.splitlines():
if raw.startswith(" /") and raw.rstrip().endswith(":"):
current_path = raw.strip()[:-1]
paths[current_path] = {}
continue
if current_path and re.match(r"^ (get|post|put|patch|delete|head|options):", raw):
method = raw.strip().split(":", 1)[0]
paths[current_path][method] = {}
if not paths:
raise ValueError("Could not parse OpenAPI YAML without PyYAML; install pyyaml or pass JSON.")
return {"paths": paths}
def load_spec(path: str) -> Dict[PathKey, Set[Method]]:
text = open(path, encoding="utf-8").read()
if path.endswith(".json"):
data = _parse_openapi_json(text)
else:
if yaml is not None:
data = yaml.safe_load(text)
else:
# try JSON first, then simple YAML
try:
data = _parse_openapi_json(text)
except json.JSONDecodeError:
data = _parse_openapi_yaml_simple(text)
paths = data.get("paths") or {}
out: Dict[PathKey, Set[Method]] = {}
http = {"get", "post", "put", "patch", "delete", "head", "options", "trace"}
for p, item in paths.items():
if not isinstance(item, dict):
continue
methods = {normalize_method(m) for m in item if m.lower() in http}
if methods:
out[normalize_path(p)] = methods
return out
def compare(spec: Dict[PathKey, Set[Method]], code: Dict[PathKey, Set[Method]]) -> List[dict]:
findings = []
all_paths = set(spec) | set(code)
for p in sorted(all_paths):
s = spec.get(p, set())
c = code.get(p, set())
if s and not c:
for m in sorted(s):
findings.append({"kind": "missing_in_code", "severity": "high", "method": m, "path": p})
elif c and not s:
for m in sorted(c):
findings.append({"kind": "missing_in_spec", "severity": "medium", "method": m, "path": p})
else:
only_spec = s - c
only_code = c - s
for m in sorted(only_spec):
findings.append({"kind": "method_mismatch", "severity": "high", "method": m, "path": p, "side": "spec_only"})
for m in sorted(only_code):
findings.append({"kind": "method_mismatch", "severity": "high", "method": m, "path": p, "side": "code_only"})
return findings
def format_report(findings: List[dict]) -> str:
counts: Dict[str, int] = {}
for f in findings:
counts[f["kind"]] = counts.get(f["kind"], 0) + 1
lines = ["Drift summary: " + ", ".join(f"{k}={v}" for k, v in sorted(counts.items()))]
if not findings:
lines.append("No drift found.")
return "\n".join(lines)
for f in findings:
side = f.get("side", "")
extra = f" ({side})" if side else ""
lines.append(f"- [{f['severity']}] {f['kind']}: {f['method']} {f['path']}{extra}")
return "\n".join(lines)
def self_test() -> int:
spec = {
normalize_path("/pets"): {"GET", "POST"},
normalize_path("/pets/{petId}"): {"GET", "DELETE"},
normalize_path("/health"): {"GET"},
}
code = {
normalize_path("/pets"): {"GET", "POST"},
normalize_path("/pets/{id}"): {"GET"}, # same shape, missing DELETE
normalize_path("/pets/{id}/adopt"): {"POST"}, # extra
}
findings = compare(spec, code)
kinds = sorted(f["kind"] + ":" + f["method"] + ":" + f["path"] for f in findings)
cases = [
("missing delete on pet", any(f["kind"] == "method_mismatch" and f["method"] == "DELETE" for f in findings), True),
("extra adopt route", any(f["kind"] == "missing_in_spec" and f["path"].endswith("/adopt") for f in findings), True),
("health missing in code", any(f["kind"] == "missing_in_code" and f["path"] == "/health" for f in findings), True),
("list pets matched", not any(f["path"] == "/pets" and f["kind"] == "missing_in_code" for f in findings), True),
("normalize params", normalize_path("/pets/{petId}") == normalize_path("/pets/{id}"), True),
("normalize slash", normalize_path("/pets/") == "/pets", True),
("parse routes", parse_routes(["GET /pets", "POST /pets # create"]) == {"/pets": {"GET", "POST"}}, True),
("empty compare", compare({"/x": {"GET"}}, {"/x": {"GET"}}) == [], True),
]
# parse simple yaml fallback
sample = """openapi: 3.0.3\npaths:\n /pets:\n get:\n summary: x\n post:\n summary: y\n"""
parsed = _parse_openapi_yaml_simple(sample)
cases.append(("yaml fallback paths", "get" in parsed["paths"]["/pets"] and "post" in parsed["paths"]["/pets"], True))
failed = 0
for name, got, expected in cases:
ok = got == expected
failed += not ok
print(f"{'PASS' if ok else 'FAIL'} {name}: got={got!r} expected={expected!r}")
# also assert finding count sanity
ok = len(findings) >= 3
failed += not ok
print(f"{'PASS' if ok else 'FAIL'} findings_count>={3}: got={len(findings)}")
print(f"\n{len(cases)+1 - failed}/{len(cases)+1} passed")
return 1 if failed else 0
def main(argv=None) -> int:
p = argparse.ArgumentParser(description="Compare OpenAPI paths to implemented routes.")
p.add_argument("--spec", help="OpenAPI YAML or JSON file")
p.add_argument("--routes", help="File with METHOD /path lines")
p.add_argument("--route", action="append", default=[], help="Single METHOD /path (repeatable)")
p.add_argument("--json", action="store_true", help="Print findings as JSON")
p.add_argument("--self-test", action="store_true")
a = p.parse_args(argv)
if a.self_test:
return self_test()
if not a.spec:
p.error("--spec is required (or use --self-test)")
route_lines: List[str] = list(a.route)
if a.routes:
route_lines.extend(open(a.routes, encoding="utf-8").read().splitlines())
if not route_lines:
p.error("provide --routes and/or --route")
spec = load_spec(a.spec)
code = parse_routes(route_lines)
findings = compare(spec, code)
if a.json:
print(json.dumps(findings, indent=2))
else:
print(format_report(findings))
return 0 if not findings else 2
if __name__ == "__main__":
sys.exit(main())