如何在执行之前修改 SQLAlchemy SQL 语句？答案

【问题标题】：How to modify SQLAlchemy SQL statement, right before execution?如何在执行之前修改 SQLAlchemy SQL 语句？
【发布时间】：2021-03-08 06:51:57
【问题描述】：

摘要：
为了我大学的学期项目，我需要为 SQLAlchemy 实现实体级访问模块。 ORM 查询应该只返回用户有权访问的那些对象。例如：

# python-pseudocode
# SomeModel contains multiple rows in database, but some_user only have acces to some_obj1, some_obj2

session.query(SomeModel).all()
[]  # returns empty list because user is not set

ACL.set_user(some_user)
session.query(SomeModel).all()
['some_obj1', 'some_obj2']  # returns object that user have acces to

问题：
我已经implemented 知道，通过扩展 SQLAlchemy 的 BaseQuery 并覆盖迭代器，几乎类似于 here 的解释，但我的讲师指出，由于效率低下，这不是正确的方法。他提出我在从数据库中检索对象后过滤对象，所以如果数据库中有数百万行，而用户只能访问其中的几个，我将无缘无故地检索这百万个对象。他建议我应该在执行之前对 SQL 语句进行拦截。

我做了一些研究，发现 SQLAlchemy events, before_cursor_execute 似乎是拦截语句的好地方。我知道我可以解析语句并注入WHERE <tablename>.id IN <ids_that_user_have_access_to>。不幸的是，我遇到了三个问题：

我需要确保在语句中的正确位置注入WHERE 子句。我有这样的直觉，当查询很简单时很容易，但对于更复杂的查询，它可能会很棘手。有没有办法将WHERE 插入总是可以正常工作的地方？
在解析时，statement 在before_cursor_execute 事件中被截获，具有这种带有问号的奇怪格式，例如： SELECT sm.id AS sm_id, sm.some_field AS sm_some_field FROM sm WHERE sm.id = ? ，与语句关联的参数在parameters 元组中（对于示例性语句，它将是(1, )），它与statement 一起传递给函数。我用moz_sql_parser测试了解析语句，但是当然问号在SQL语句中是无效的，所以无法解析。如何解析这样的语句？
即使我能够解析语句并注入WHERE 子句，我怎么知道我应该在新创建的parameters 元组中的哪个位置放置适当的参数？

【问题讨论】：

如果<ids_that_user_have_access_to> 的数量在数百万左右，那么您仍然会遇到同样的问题。理想情况下，您希望在单个查询中在数据库中完成选择和过滤。如果访问控制标准也存储在数据库中，这应该是可能的。
此外，对于任意大小和复杂性的查询，拦截、解析、剖析和修改原始 SQL 可能会变得非常困难且容易出错。您最好将原始查询包装在 SELECT * FROM (…) WHERE id IN <list>
实际上，包装可能是一种方式（我一开始没有考虑过），但我不知道在语句包装器中使用* 是否正确。如果没有，我仍然需要解析原始语句以提取列名。
原来，使用*应该没有问题，但还是需要解析语句。首先 - 我只需要修改 SELECT 语句，最简单的方法是检查字符串是否包含 SELECT 但我不知道这是否是正确的方法。其次 - 我需要知道表名来建立条件。 WHERE id IN <list> 不起作用，因为 id 不会被识别，而是需要 <tablename>_id。

标签： sql python-3.x parsing orm sqlalchemy

【解决方案1】：

我正在回复自己，但我找到了合适的解决方案。

原来before_execute，事件是更好的拦截和修改查询的地方，因为未编译的SQL构造ClauseElement是可用的。这使得修改查询变得更加容易。

@event.listens_for(engine, 'before_execute', retval=True)
def intercept(conn, clauseelement, multiparams, params):
    from sqlalchemy.sql.selectable import Select

    # check if it's select statement
    if isinstance(clauseelement, Select):
        # 'froms' represents list of tables that statement is querying
        table = clauseelement.froms[0]

        # adding filter in clause
        clauseelement = clauseelement.where(table.c.id.in_(list_of_ids))

    return clauseelement, multiparams, params

如需了解更多信息，请阅读before_execute 事件和expression API。

【讨论】：