迁移安全审查提示

--- name: migration-safety-review description: 审查数据库结构迁移（原始 SQL 或由 Rails、Django、Alembic、Prisma、Knex、Laravel、Flyway 生成的 ORM 迁移），在生产环境部署前评估风险 - 表锁 DDL、全表重写、数据丢失、对运行中应用代码的破坏性变更、以及缺失的回滚路径 - 并提出安全的零停机重写方案。当 diff 或 PR 新增或修改迁移文件时，当用户询问"此迁移是否安全？"时，或在向繁忙的 PostgreSQL 或 MySQL 数据库部署结构变更之前使用。 --- # 迁移安全审查 你将以资深 DBA 在生产部署前的谨慎态度审查结构迁移。目标是给出明确结论和具体、更安全的 SQL - 而非泛泛的数据库说教。 ## 本技能包含的文件 - `scripts/scan_migration.py` - 用于检测高风险 SQL 语句的快速启发式扫描器 - `references/risk-catalog.md` - 逐操作风险和安全模式 - `references/expand-contract.md` - 在滚动发布期间保持新旧应用代码同时可用 - `templates/review-report.md` - 你必须产出的报告格式 - `examples/example-review.md` - 一个完整的示例审查，用于校准语气和深度 ## 工作流程 ### 1. 确定审查范围内的迁移 - 如果审查分支或 PR：`git diff --name-only origin/main...HEAD`，保留迁移文件夹下的文件（`migrations/`、`db/migrate/`、`alembic/versions/`、`prisma/migrations/`、`database/migrations/`、`db/migration/`）。 - 否则使用用户指定的文件或 SQL。 - 注意哪些迁移是新增的，哪些已在任何环境中应用。切勿建议编辑已应用的迁移；应提出新的后续迁移。 ### 2. 建立上下文 根据配置文件、docker-compose 或询问用户来确定： - 引擎和主版本（例如 PostgreSQL 15、MySQL 8.0）。锁行为取决于此。 - 每个受影响表的近似大小和写入流量。 - 部署方式：迁移是在新代码发布之前、期间还是之后运行？ 如果大小或流量未知，则假设表很大且繁忙，并明确说明。 ### 3. 获取真实 SQL ORM 代码隐藏了实际执行的内容。首先渲染 SQL： | 框架 | 命令 | |-----------|---------| | Django | `python manage.py sqlmigrate <app> <migration>` | | Rails | 在临时数据库上执行 `rails db:migrate`，然后检查 `db/structure.sql` 的差异 | | Alembic | `alembic upgrade <from>:<to> --sql` | | Prisma | 读取 `prisma/migrations/<name>/migration.sql` | | Laravel | `php artisan migrate --pretend` | | Knex | 在临时数据库上运行，使用 `DEBUG=knex:query` 并复制记录的 SQL | | Flyway / Liquibase | `.sql` 文件或 `liquibase update-sql` | 如果渲染后的 SQL 还不是 `.sql` 文件，请将其保存到临时文件。 ### 4. 运行扫描器 ```bash python3 scripts/scan_migration.py --dialect postgres path/to/migration.sql python3 scripts/scan_migration.py --dialect mysql db/*.sql ``` 它输出 `file:line [SEVERITY] RULE message`，如果存在任何 HIGH 发现则退出码为 1。 将其输出视为线索而非结论：它使用正则表达式，可能遗漏动态 SQL，且无法知道表的大小。 ### 5. 手动审查每条语句 对每条语句，使用 `references/risk-catalog.md` 回答： 1. 它获取什么锁，持续多久（瞬间、全表扫描或全表重写）？ 2. 它会导致数据丢失或损坏吗？这是有意为之且有备份吗？ 3. 它会排在长事务后面吗？是否设置了 `lock_timeout`（Postgres）或 `lock_wait_timeout`（MySQL），以便快速失败而不是阻塞所有流量？ 4. 它是否在不应有的事务中运行（例如 `CREATE INDEX CONCURRENTLY`）？ 5. 大型数据回填是否分批处理并与 DDL 分离？ ### 6. 检查应用兼容性 在滚动部署期间，旧代码和新代码会同时针对新结构运行。 遵循 `references/expand-contract.md`： - 在代码库中搜索（`rg -n '<column_or_table_name>'`）每个重命名、删除或更改类型的对象，包括原始 SQL、序列化器和分析查询。 - 标记当前部署的代码无法容忍的任何变更。 ### 7. 验证回滚路径 - 是否存在 down 迁移，它是否真的能恢复之前的状态？ - 删除和有损类型更改是单向的：需要备份或分阶段计划。 ### 8. 撰写报告 严格按照 `templates/review-report.md` 填写。匹配 `examples/example-review.md` 的深度。对每个 HIGH 或 MEDIUM 发现，给出能达到相同最终状态的安全替代 SQL 或迁移代码，必要时拆分为有序的部署步骤。 ## 结论 - **SAFE** - 大表上无阻塞锁，无数据丢失，向后兼容。 - **SAFE WITH CHANGES** - 应用列出的重写后可发布。 - **UNSAFE** - 按现状会导致停机、数据丢失或运行中代码出错。 ## 规则 - 切勿自行在生产或共享数据库上运行迁移。 - 除非用户要求，否则不要修改迁移文件；在报告中提出更改建议。 - 要具体：指明表名、锁类型和失败模式。跳过泛泛的建议。 - 如果不确定某个版本特定行为，请说明，并建议在接近生产规模的副本上使用 `\timing` / `EXPLAIN` 和锁监控进行测试。 FILE:references/risk-catalog.md # 风险目录：常见迁移操作 锁名称基于 PostgreSQL。ACCESS EXCLUSIVE 阻止所有读写；SHARE 阻止写入；SHARE UPDATE EXCLUSIVE 两者都不阻止。 ## 锁队列问题（适用于以下所有内容） 即使是"瞬间"的 ALTER TABLE 也需要短暂的 ACCESS EXCLUSIVE。如果长查询或空闲事务会话持有该表，ALTER 将等待 - 而每个新查询都会排在它后面。1 毫秒的变更可能导致数分钟的中断。 始终以以下内容开始有风险的迁移： ```sql SET lock_timeout = '5s'; -- 快速失败，稍后重试 SET statement_timeout = '15min'; -- 可选上限 ``` MySQL 等效：`SET SESSION lock_wait_timeout = 5;`（元数据锁）。 ## PostgreSQL 操作 | 操作 | 风险 | 安全模式 | |-----------|------|--------------| | `CREATE INDEX` | SHARE 锁：整个构建期间阻止写入 | `CREATE INDEX CONCURRENTLY`，在事务外；失败时删除 INVALID 索引并重试。Rails：`disable_ddl_transaction!`；Django：`atomic = False` | | `DROP INDEX` | ACCESS EXCLUSIVE | `DROP INDEX CONCURRENTLY` | | `ADD COLUMN` 可空，无默认值 | 瞬间 | 安全（仍设置 lock_timeout） | | `ADD COLUMN ... DEFAULT <常量>` | PG 11+ 瞬间，11 之前重写 | 11+ 安全 | | `ADD COLUMN ... DEFAULT now()/random()/gen_random_uuid()` | 易变默认值：全表重写 | 添加可空列，分批回填，然后设置默认值 | | `ADD COLUMN ... NOT NULL` 无默认值 | 在非空表上失败 | 添加可空列，回填，然后强制执行 NOT NULL（见下文） | | `ALTER COLUMN ... SET NOT NULL` | 在 ACCESS EXCLUSIVE 下全表扫描 | `ADD CONSTRAINT c CHECK (col IS NOT NULL) NOT VALID`；`VALIDATE CONSTRAINT c`；然后 `SET NOT NULL`（PG 12+ 跳过扫描）；删除 `c` | | `ALTER COLUMN ... TYPE` | 通常在 ACCESS EXCLUSIVE 下全表重写 + 索引重建 | 仅当二进制可强制转换时安全（varchar(n) 到更大的 n 或 text）。否则新列 + 双写 + 回填 + 切换 | | `ADD FOREIGN KEY` | 验证所有行时锁定两个表 | `ADD CONSTRAINT ... NOT VALID`，然后在单独步骤中 `VALIDATE CONSTRAINT` | | `ADD CHECK` | 在 ACCESS EXCLUSIVE 下扫描 | 相同的 NOT VALID + VALIDATE 模式 | | `ADD UNIQUE` / `ADD PRIMARY KEY` | 在锁下构建索引 | `CREATE UNIQUE INDEX CONCURRENTLY idx ...`；然后 `ADD CONSTRAINT ... UNIQUE USING INDEX idx` | | `RENAME COLUMN` / `RENAME TO` | 瞬间，但破坏运行中的代码 | 扩展/收缩（见 expand-contract.md） | | `DROP COLUMN` | 瞬间，但不可逆；选择它的旧代码报错 | 先删除所有代码引用并部署；然后删除 | | `DROP TABLE` / `TRUNCATE` | 不可逆的数据丢失 | 确认备份且无读取者；考虑先重命名为 `_deprecated` | | `ALTER TYPE ... ADD VALUE` | 新值在同一事务中不可用；PG 12 之前完全无事务 | 将其放在单独的迁移中 | | `VACUUM FULL` / `CLUSTER` / `REINDEX` | 在 ACCESS EXCLUSIVE 下全表重写 | `REINDEX CONCURRENTLY`（PG 12+），使用 `pg_repack` 处理膨胀 | | 大型 `UPDATE` / `DELETE` | 长行锁、WAL 峰值、副本延迟 | 按主键分批（1k-10k 行），每批提交，在 DDL 迁移之外运行 | ## MySQL 8.0（InnoDB）注意事项 - 始终声明算法，以便 MySQL 报错而不是静默复制表： `ALTER TABLE t ADD COLUMN c INT, ALGORITHM=INSTANT;` 或 `ALTER TABLE t ADD INDEX i (c), ALGORITHM=INPLACE, LOCK=NONE;` - `ADD COLUMN` 在 8.0.12+（最后位置）和 8.0.29+（任意位置）为 INSTANT。 - `MODIFY` / `CHANGE COLUMN` 类型更改使用 ALGORITHM=COPY：阻止写入。 - 对于仅支持 COPY 的大表更改，使用 `gh-ost` 或 `pt-online-schema-change`。 - MySQL 中的 DDL 不是事务性的：失败的多语句迁移会使结构半应用。每个迁移只保留一条 DDL 语句。 FILE:references/expand-contract.md # 扩展 / 收缩：向后兼容的结构变更 在滚动部署期间，旧版本和新版本的应用会并行运行。 如果迁移在新代码上线前运行，旧代码必须能使用新结构。如果在之后运行，新代码必须能使用旧结构。 扩展/收缩使每一步都与两者兼容。 ## 三个阶段 1. **扩展** - 仅添加新结构（列、表、索引）。不删除或重命名任何内容。旧代码忽略新增内容。 2. **迁移** - 部署同时写入新旧结构的代码，分批回填现有行，然后将读取切换到新结构。 3. **收缩** - 一旦没有部署的代码再触及旧结构，在单独的、更晚的迁移中删除它。 每个阶段都是独立的部署。切勿在一个迁移中同时进行扩展和收缩。 ## 配方 ### 重命名列（`users.name` 到 `users.full_name`） 1. 迁移：添加可空的 `full_name`。 2. 代码：同时写入 `name` 和 `full_name`；读取 `name`。 3. 在 `full_name IS NULL` 处分批回填 `full_name = name`。 4. 代码：读取 `full_name`；继续写入两者。 5. 代码：停止写入 `name`。（Rails：在此处将 `name` 添加到 `ignored_columns`。） 6. 迁移：删除 `name`。 ### 更改列类型（`orders.amount` int 到 numeric） 与重命名相同：添加 `amount_numeric`，双写，回填，切换读取，删除旧列。 如果应用更改困难，触发器可以处理双写。 ### 使列 NOT NULL 1. 代码：始终写入值。 2. 分批回填 NULL 行。 3. 迁移：CHECK ... NOT VALID，VALIDATE，SET NOT NULL（见 risk-catalog.md）。 ### 删除列或表 1. 代码：移除所有读写（搜索 ORM 模型、原始 SQL、视图、报告、ETL 作业以及共享数据库的其他服务）。 2. 部署并等待至少一个完整的发布周期。 3. 迁移：删除。如果数据重要，先进行备份或快照。 ### 拆分或移动表 创建新表，双写，回填，切换读取，停止旧写入，删除。 ## 每次变更需要回答的兼容性问题 - 是否有部署的代码执行 `SELECT *` 或映射所有列（ORM 通常在启动时缓存列列表，当一列消失时失败）？ - 旧代码的插入是否会因新列 NOT NULL 且无默认值而失败？ - 其他服务、cron 作业、BI 仪表板或副本是否读取此表？ - 部署能否在没有 down 迁移的情况下回滚到之前的代码版本？ 如果最后一个问题的答案是"否"，则该变更不向后兼容。 FILE:templates/review-report.md # 迁移安全审查：<迁移名称或 PR 标题> **结论：** SAFE | SAFE WITH CHANGES | UNSAFE **引擎：** <例如 PostgreSQL 15> | **审查的文件：** <数量> **假设：** <表大小、流量、部署顺序 - 标记任何猜测> ## 摘要 <2-4 句话：迁移做什么、最大的风险以及需要更改什么。> ## 发现 | # | 严重性 | 文件:行 | 语句 | 风险 | |---|----------|-----------|-----------|------| | 1 | HIGH | <路径:行> | `<简短 SQL>` | <锁 / 数据丢失 / 破坏旧代码> | ### 1. <发现的简短标题> - **会发生什么：** <获取的锁、持续时间、谁被阻塞或什么被破坏> - **为什么在这里重要：** <表大小、流量、依赖它的代码> - **安全替代方案：** ```sql -- 替代 SQL 或迁移代码，按运行顺序 ``` <对每个 HIGH 和 MEDIUM 发现重复。将 LOW 发现分组到一个列表中。> ## 应用兼容性 - <每个重命名 / 删除 / 更改类型的对象以及代码仍在何处使用它> - <或："没有受影响的代码引用 - 已用 rg 检查 X、Y。"> ## 回滚计划 - <down 迁移是否恢复状态？什么不可逆？> - <运行前需要备份或快照：是/否> ## 推荐的部署顺序 1. <迁移或代码部署步骤> 2. <...> ## 扫描器输出 ``` <粘贴 scripts/scan_migration.py 的输出，或注明误报> ``` FILE:examples/example-review.md # 示例审查 **输入：** PR "添加订单状态跟踪"，包含一个针对 PostgreSQL 15 的 Rails 迁移。 `orders` 约有 4000 万行，持续有写入。迁移在新代码之前运行。 ```sql -- 从 db/migrate/20261002_add_status_to_orders.rb 渲染 ALTER TABLE orders ADD COLUMN status varchar NOT NULL DEFAULT 'pending'; ALTER TABLE orders RENAME COLUMN shipped_on TO shipped_at; CREATE INDEX index_orders_on_status ON orders (status); ALTER TABLE orders ADD CONSTRAINT fk_orders_carrier FOREIGN KEY (carrier_id) REFERENCES carriers (id); ``` **扫描器：** 3 个 HIGH（重命名、非并发索引、已验证外键），1 个 MEDIUM（无锁超时）。 --- # 迁移安全审查：添加订单状态跟踪 **结论：** UNSAFE **引擎：** PostgreSQL 15 | **审查的文件：** 1 **假设：** orders 约 4000 万行，高写入流量（来自用户）；carriers 很小。 ## 摘要 添加订单状态列，重命名 `shipped_on`，为 status 建索引，并添加 carrier 外键。 status 列本身在 PG 15 上是安全的，但重命名会破坏运行中的应用，索引和外键会阻塞 `orders` 上的写入数分钟。 拆分为三个迁移并使用并发 / NOT VALID 变体。 ## 发现 | # | 严重性 | 文件:行 | 语句 | 风险 | |---|----------|-----------|-----------|------| | 1 | HIGH | rendered.sql:3 | `RENAME COLUMN shipped_on` | 旧代码在部署时出错 | | 2 | HIGH | rendered.sql:4 | `CREATE INDEX ... (status)` | 构建期间阻止写入 | | 3 | HIGH | rendered.sql:5 | `ADD ... FOREIGN KEY` | 在锁下全量验证扫描 | | 4 | MEDIUM | rendered.sql:1 | 无 `lock_timeout` | ALTER 可能排队并阻塞流量 | ### 1. 列重命名破坏运行中的代码 - **会发生什么：** 重命名是瞬间的，但仍在旧版本上的应用服务器查询 `shipped_on` 并失败，报 `column does not exist`，直到部署完成。 - **为什么在这里重要：** `rg -n shipped_on` 找到 7 处引用，包括 `app/serializers/order_serializer.rb` 和夜间 `reports/fulfillment.sql`。 - **安全替代方案：** 扩展/收缩。添加 `shipped_at`，双写，分批回填，切换读取，然后在后续版本中删除 `shipped_on`。 ### 2. 索引构建阻止写入 - **安全替代方案**（单独迁移，`disable_ddl_transaction!`）： ```sql CREATE INDEX CONCURRENTLY index_orders_on_status ON orders (status); ``` ### 3. 外键在锁下验证 4000 万行 - **安全替代方案：** ```sql SET lock_timeout = '5s'; ALTER TABLE orders ADD CONSTRAINT fk_orders_carrier FOREIGN KEY (carrier_id) REFERENCES carriers (id) NOT VALID; -- 下一个迁移（在 orders 上仅获取 SHARE UPDATE EXCLUSIVE）： ALTER TABLE orders VALIDATE CONSTRAINT fk_orders_carrier; ``` **LOW：** 无。注意 `ADD COLUMN ... DEFAULT 'pending'` 在 PG 11+ 上仅元数据操作。 ## 应用兼容性 - `shipped_on`：7 处代码引用加一个 SQL 报告；必须保留到收缩阶段。 ## 回滚计划 - down 迁移删除 `status`（数据丢失可接受：新列）。重命名可逆。 - 移除重命名后，此变更集无需备份。 ## 推荐的部署顺序 1. 迁移 A：`SET lock_timeout`；添加 `status`；添加 `shipped_at`；添加 FK NOT VALID。 2. 迁移 B（无事务）：并发创建 status 索引。 3. 迁移 C：验证 FK。部署双写 `shipped_on`/`shipped_at` 的代码。 4. 回填 `shipped_at`；切换读取；后续版本删除 `shipped_on`。 FILE:scripts/scan_migration.py #!/usr/bin/env python3 """用于检测迁移文件中高风险 SQL 的启发式扫描器（PostgreSQL / MySQL）。 用法：python3 scan_migration.py [--dialect postgres|mysql] FILE [FILE ...] 退出码：0 = 无 HIGH 发现，1 = 有 HIGH 发现，2 = 用法错误。""" import re, sys F = re.I | re.S COLDEF = r"(?:\([^)]*\)|[^,(])*" # 一个列定义，允许 numeric(10,2) RULES = [ # (严重性, 规则 ID, 方言或 None 表示两者, 正则表达式, 消息) ("HIGH", "drop-table", None, r"^DROP\s+TABLE\b", "不可逆的数据丢失；确认备份且无读取者"), ("HIGH", "truncate", None, r"^TRUNCATE\b", "不可逆的数据丢失"), ("HIGH", "drop-column", None, r"^ALTER\s+TABLE\b.*\bDROP\s+(COLUMN\b|(?!CONSTRAINT|INDEX|KEY|PRIMARY|FOREIGN|CHECK|DEFAULT|NOT|IDENTITY|EXPRESSION)\w)", "数据丢失；读取它的部署代码将失败 - 先移除代码引用"), ("HIGH", "rename", None, r"^ALTER\s+TABLE\b.*\bRENAME\b", "破坏运行中的代码；使用扩展/收缩"), ("HIGH", "type-change", "postgres", r"^ALTER\s+TABLE\b.*\bALTER\s+(COLUMN\s+)?\S+\s+(SET\s+DATA\s+)?TYPE\b", "通常在 ACCESS EXCLUSIVE 下全表重写"), ("HIGH", "type-change", "mysql", r"^ALTER\s+TABLE\b.*\b(MODIFY|CHANGE)\s+(COLUMN\s+)?\S+", "列重新定义通常使用 ALGORITHM=COPY（阻止写入）"), ("HIGH", "index-not-concurrent", "postgres", r"^CREATE\s+(UNIQUE\s+)?INDEX\s+(?!CONCURRENTLY)", "构建期间阻止写入；使用 CREATE INDEX CONCURRENTLY"), ("MEDIUM", "drop-index-not-concurrent", "postgres", r"^DROP\s+INDEX\s+(?!CONCURRENTLY)", "获取 ACCESS EXCLUSIVE；使用 DROP INDEX CONCURRENTLY"), ("HIGH", "fk-validated", "postgres", r"^ALTER\s+TABLE\b(?!.*\bNOT\s+VALID\b).*\b(FOREIGN\s+KEY|REFERENCES)\b", "锁定两个表时验证所有行；添加 NOT VALID，然后 VALIDATE"), ("MEDIUM", "check-validated", "postgres", r"^ALTER\s+TABLE\b(?!.*\bNOT\s+VALID\b).*\bADD\s+(CONSTRAINT\s+\S+\s+)?CHECK\b", "在锁下全表扫描；添加 NOT VALID，然后 VALIDATE"), ("MEDIUM", "set-not-null", "postgres", r"\bSET\s+NOT\s+NULL\b", "在 ACCESS EXCLUSIVE 下全表扫描；先验证 CHECK (col IS NOT NULL)"), ("HIGH", "add-not-null-no-default", None, r"^ALTER\s+TABLE\b.*\bADD\s+(COLUMN\s+)?(?!" + COLDEF + r"\bDEFAULT\b)" + COLDEF + r"\bNOT\s+NULL\b", "在非空表上失败（或旧代码插入失败）；添加可空列，回填，然后强制执行"), ("MEDIUM", "volatile-default", "postgres", r"^ALTER\s+TABLE\b.*\bADD\b.*\bDEFAULT\s+(now|random|clock_timestamp|gen_random_uuid|uuid_generate_v\d)\s*\(", "易变默认值重写表；添加可空列，回填，然后设置默认值"), ("MEDIUM", "unique-without-index", "postgres", r"^ALTER\s+TABLE\b(?!.*\bUSING\s+INDEX\b).*\bADD\s+(CONSTRAINT\s+\S+\s+)?(UNIQUE|PRIMARY\s+KEY)\b", "在锁下构建索引；先并发创建，然后 ADD CONSTRAINT ... USING INDEX"), ("MEDIUM", "mysql-no-algorithm", "mysql", r"^(ALTER\s+TABLE|CREATE\s+(UNIQUE\s+)?INDEX)\b(?!.*\bALGORITHM\s*=)", "声明 ALGORITHM=INSTANT|INPLACE, LOCK=NONE，以便 MySQL 拒绝阻塞性复制"), ("HIGH", "dml-no-where", None, r"^(UPDATE|DELETE)\b(?!.*\bWHERE\b)", "在单个事务中触及每一行；分批处理"), ("LOW", "dml-in-migration", None, r"^(UPDATE|DELETE|INSERT)\b.*\bWHERE\b", "迁移中的数据变更；如果表很大则分批处理"), ("MEDIUM", "table-rewrite", "postgres", r"^(VACUUM\s+FULL|CLUSTER|REINDEX\s+(?!.*CONCURRENTLY))", "在 ACCESS EXCLUSIVE 下重写；使用 REINDEX CONCURRENTLY 或 pg_repack"), ("LOW", "enum-add-value", "postgres", r"^ALTER\s+TYPE\b.*\bADD\s+VALUE\b", "新值在同一事务中不可用；放在单独的迁移中"), ] def statements(sql): """生成 (行号, 语句)，去除注释后。朴素的 ';' 分割。""" sql = re.sub(r"/\*.*?\*/", lambda m: re.sub(r"[^\n]", " ", m.group()), sql, flags=re.S) sql = re.sub(r"--[^\n]*", "", sql) pos = 0 for part in sql.split(";"): stripped = part.lstrip() line = sql.count("\n", 0, pos + len(part) - len(stripped)) + 1 pos += len(part) + 1 if stripped.strip(): yield line, " ".join(stripped.split()) def scan(path, dialect): text = open(path, encoding="utf-8", errors="replace").read() stmts, out = list(statements(text)), [] for line, st in stmts: for sev, rid, dia, rx, msg in RULES: if (dia is None or dia == dialect) and re.search(rx, st, F): out.append((sev, f"{path}:{line} [{sev}] {rid}: {msg}\n > {st[:110]}")) has_ddl = any(re.match(r"(ALTER|CREATE\s+(UNIQUE\s+)?INDEX|DROP)\b", s, re.I) for _, s in stmts) timeout = "lock_timeout" if dialect == "postgres" else "lock_wait_timeout" if has_ddl and timeout not in text.lower(): out.append(("MEDIUM", f"{path}:1 [MEDIUM] no-lock-timeout: DDL without {timeout}; it may queue and block all traffic")) if re.search(r"\bCONCURRENTLY\b", text, re.I) and re.search(r"^\s*(BEGIN|START\s+TRANSACTION)\b", text, re.I | re.M): out.append(("HIGH", f"{path}:1 [HIGH] concurrently-in-transaction: CONCURRENTLY cannot run inside a transaction block")) return out def main(argv): dialect = "postgres" if len(argv) >= 2 and argv[0] == "--dialect": dialect, argv = argv[1].lower(), argv[2:] if dialect not in ("postgres", "mysql") or not argv: print(__doc__, file=sys.stderr) return 2 try: findings = [f for p in argv for f in scan(p, dialect)] except OSError as e: print(f"error: {e}", file=sys.stderr) return 2 for _, text in findings: print(text) counts = {s: sum(1 for f in findings if f[0] == s) for s in ("HIGH", "MEDIUM", "LOW")} print(f"\n{len(argv)} file(s) scanned: {counts['HIGH']} HIGH, {counts['MEDIUM']} MEDIUM, {counts['LOW']} LOW") print("Heuristic only: confirm each finding against references/risk-catalog.md.") return 1 if counts["HIGH"] else 0 if __name__ == "__main__": sys.exit(main(sys.argv[1:]))