新建/覆盖 SQLAlchemy 运算符编译器输出

Question

编译in_ 表达式的默认 SQLAlchemy 行为对于非常大的列表是病态的，我想为运算符创建一个自定义的、更快的编译器。如果解决方案是一个新的运算符（即：in_list_）或者它是否覆盖了in_ 的默认编译器，这对应用程序来说并不重要。但是，我还没有找到任何关于如何具体执行此操作的文档。

subclassing guidelines for compilation extension 不包含任何关于运算符的内容，这表明这不是开始的地方。 documentation on redefining and creating new operators 专注于更改或创建新的运算符行为，但运算符的行为不是问题，只是编译器。

这是我正在尝试完成的一个非常无效的示例：

from sqlalchemy.types import TypeEngine

class in_list_(TypeEngine.Comparator):
  pass

@compiles(in_list_)
def in_list_impl(element, compiler, **kwargs):
  return "IN ('Now', 'I', 'can', 'inline', 'the', 'list')"

然后在一个表达式中：

select([mytable.c.x, mytable.c.y]).where(mytable.c.x.in_list_(long_list))

Answer 1

将IN 用于非常 大型列表确实是病态的，您可能会更好地为using a temporary table 和IN 提供子查询或连接。但问题是“如何覆盖特定运算符的编译器输出”。对于像IN 和NOT IN 这样的二元运算符，您需要覆盖的是SQLAlchemy 如何处理BinaryExpressions 的编译：

from sqlalchemy.ext.compiler import compiles
from sqlalchemy.sql.elements import BinaryExpression
from sqlalchemy.sql.operators import in_op, notin_op

def visit_in_op_binary(compiler, binary, operator, **kw):
    return "%s IN %s" % (
        compiler.process(binary.left, **kw),
        compiler.process(binary.right, **{**kw, "literal_binds": True}))

def visit_notin_op_binary(compiler, binary, operator, **kw):
    return "%s NOT IN %s" % (
        compiler.process(binary.left, **kw),
        compiler.process(binary.right, **{**kw, "literal_binds": True}))

@compiles(BinaryExpression)
def compile_binary(binary, compiler, override_operator=None, **kw):
    operator = override_operator or binary.operator

    if operator is in_op:
        return visit_in_op_binary(
            compiler, binary, operator, override_operator=override_operator,
            **kw)

    if operator is notin_op:
        return visit_notin_op_binary(
            compiler, binary, operator, override_operator=override_operator,
            **kw)

    return compiler.visit_binary(binary, override_operator=override_operator, **kw)

请注意，简单地生成包含绑定参数的分组和子句列表的二进制表达式对于非常大的列表会花费大量时间，更不用说即使使用文字绑定也要编译所有内容，因此您可能不会观察到显着的性能提升.另一方面，许多实现对您可以在语句中使用多少个占位符/参数有限制，因此内联绑定允许这样的查询完全运行。

另一方面，如果您的列表确实符合您的实现设置的限制（Postgresql 似乎只受可用 RAM 的限制），您可能不需要任何具有最新 SQLAlchemy 的编译器解决方法； use expanding bind parameters instead:

In [15]: %%time
    ...: session.query(Foo).\
    ...:     filter(Foo.data.in_(range(250000))).\
    ...:     all()
    ...: 
CPU times: user 5.09 s, sys: 91.9 ms, total: 5.18 s
Wall time: 5.18 s
Out[15]: []

In [16]: %%time
    ...: session.query(Foo).\
    ...:     filter(Foo.data.in_(bindparam('xs', range(250000), expanding=True))).\
    ...:     all()
    ...: 
CPU times: user 310 ms, sys: 8.05 ms, total: 318 ms
Wall time: 317 ms
Out[16]: []

正如 cmets 中所述，在 1.4 版中，扩展 bindparam 将支持开箱即用的文字执行：

In [4]: session.query(Foo).\
   ...:     filter(Foo.data.in_(
   ...:         bindparam('xs', range(10), expanding=True, literal_execute=True))).\
   ...:     all()
2019-09-07 20:35:04,560 INFO sqlalchemy.engine.base.Engine BEGIN (implicit)
2019-09-07 20:35:04,561 INFO sqlalchemy.engine.base.Engine SELECT foo.id AS foo_id, foo.data AS foo_data 
FROM foo 
WHERE foo.data IN (0, 1, 2, 3, 4, 5, 6, 7, 8, 9)
2019-09-07 20:35:04,561 INFO sqlalchemy.engine.base.Engine ()
Out[4]: []

Answer 2

您可以做的一件事是使用原始 sql 并手动构建查询。但是，这是至关重要的，您必须使用 BINDS。

一旦你承诺，你需要管理整个查询的变量，而不仅仅是 IN 列表位。所以你完全靠自己。经常这样做是不现实的，除非您有专门的、经过高度测试的实用程序功能，但它确实有效。而且也很快：我必须在 999 分块，因为 Oracle 没有超过，但 Postgresql 或 Oracle 都没有抱怨太多。而且，是的，这是在 SQLAlchemy (1.3.8) 下。

这里有一些示例代码。列表的位都是动态生成的，以 Postgresql 为目标。不幸的是，每个 RDBMS 都有自己的占位符格式和绑定变量。参考PEP249 paramstyles 了解更多详情。

生成的查询是什么样的：

qry = """select recname, objecttype
            from bme_mvprd
            where ignore = false
YOU HAVE TO BUILD THIS ?
            and objecttype 
in ( 
%(objecttypes_000__)s
, %(objecttypes_001__)s
, %(objecttypes_002__)s
, %(objecttypes_003__)s
)
...
"""

条件最初来自这个数组：[0, 1, 2, 4]

您作为绑定参数传递的内容如下所示：

以及绑定参数的样子，再次，Postgres 特定的：

（是的，你也需要生成这个）

sub = {
    'objecttypes_000__': 0,
    'objecttypes_001__': 1,
    'objecttypes_002__': 2,
    'objecttypes_003__': 4,
}

您必须execute(qry, sub) 才能使用绑定。

NO CEATING with execute(?qry % sub?) 可以在 Postgresql 中工作，但会让你回到 SQL 注入领域。

Oracle 使用 :1, :2, :3 类型的占位符，这样会出错，但 Postgresql 使用 Python 类型的占位符，因此您需要非常小心，以免意外绕过参数绑定。

注意：大型IN SQL 操作有时可以替换为EXISTS 测试，如果可能的话，应该首选这些操作。我上面的 chunk-by-999 示例是因为除了首先填充临时表之外别无他法：这是一个示例，而不是最佳实践。

PPS：如果您的列表为空怎么办？ I asked a question about that => answer = "... in (%(var001)s)...", {'var001':None} 但您根本不能添加IN (...)。

Answer 3

完全披露，我不知道如何覆盖in_() 的编译器，而且我知道我有过度简化的风险，所以请善意接受这一点，但鉴于你上面的例子，我什至不会尝试。相反，只需创建一个辅助函数：

from sqlalchemy import Table, Column, String, MetaData
from sqlalchemy.sql import text


long_list = ['Now', 'I', 'can', 'inline', 'the', 'list']
tbl = Table("mytable", MetaData(), Column("x", String), Column("y", String))


def col_in_list(col, l):
    # do something safe to generate your in clause here.
    return text("IN ('Now', 'I', 'can', 'inline', 'the', 'list')")


if __name__ == "__main__":
    print(tbl.select().where(col_in_list(tbl.c.x, long_list)))

渲染：

SELECT mytable.x, mytable.y
FROM mytable
WHERE mytable.x IN ('Now','I','can','inline','the','list')