如何借助SQLAlchemy跨多数据库可靠识别只读SQL语句？

方案1：通过SQLAlchemy解析语句类型 #

SQLAlchemy的text()对象可以解析SQL语句并识别其操作类型，直接通过内置属性判断是否属于只读操作：

from sqlalchemy import text

raw_query = "SELECT * FROM table"
parsed_stmt = text(raw_query)

# 仅SELECT类语句属于只读
is_read_only = parsed_stmt.is_select or parsed_stmt.is_textual_select

# 检查是否为修改类语句
if parsed_stmt.is_insert or parsed_stmt.is_update or parsed_stmt.is_delete or parsed_stmt.is_ddl:
    print("该语句需要非只读权限")
else:
    print("该语句仅需只读权限")

注意：对于包含多命令的复杂SQL脚本，需要遍历所有子语句逐一判断，避免遗漏修改操作。

方案2：用只读连接测试执行（最可靠） #

直接借助数据库自身的权限校验机制，用只读权限的连接尝试执行语句（在事务中执行后回滚），如果触发权限错误则说明需要非只读权限：

from sqlalchemy import text
from sqlalchemy.exc import ProgrammingError, OperationalError

def requires_write_permission(raw_query, read_only_engine):
    try:
        with read_only_engine.connect() as conn:
            trans = conn.begin()
            conn.execute(text(raw_query))
            trans.rollback()  # 回滚避免意外修改
        return False
    except (ProgrammingError, OperationalError) as e:
        err_msg = str(e).lower()
        if "permission denied" in err_msg or "read-only" in err_msg:
            return True
        # 语法错误等其他异常不属于权限问题，直接抛出
        raise

这种方式完全适配所有数据库类型，不会因为SQL解析漏洞误判，是最可靠的方案。

方案3：第三方SQL解析库（轻量场景） #

用sqlparse库直接分析SQL字符串中的关键字，适合不需要数据库连接的轻量场景：

import sqlparse
from sqlparse.tokens import DML, DDL

def is_write_sql(raw_query):
    parsed = sqlparse.parse(raw_query)[0]
    for token in parsed.tokens:
        if token.ttype == DML and token.value.upper() in ('INSERT', 'UPDATE', 'DELETE', 'MERGE'):
            return True
        if token.ttype == DDL and token.value.upper() in ('CREATE', 'ALTER', 'DROP', 'TRUNCATE'):
            return True
    return False

注意：需要自行处理注释、大小写、多语句等边界情况，复杂SQL可能出现误判。

选择建议 #

• 优先选方案2，可靠性最高；
• 若不想连接数据库，选方案1，SQLAlchemy的解析已适配多数主流数据库；
• 简单场景或离线分析可选方案3，需额外完善解析逻辑。

内容的提问来源于stack exchange，提问作者pwwolff