SQLAlchemy 中的位运算符：使用 Oracle 数据库时出错

Question

我需要使用按位逻辑 AND/OR 编写一些查询，但只有当我使用 oracle 数据库时，才会收到以下错误：

sqlalchemy.exc.DatabaseError: (cx_Oracle.DatabaseError) ORA-01036: 非法变量名称/编号 [SQL: 'SELECT "Cars"."Id", "Cars"."Name", "Cars"."Price" \nFROM "Cars" \nWHERE ("Cars"."Price" & :Price_1) > :param_1'] [参数：{'Price_1': 32768, 'param_1': 0}]

如果我使用 PostgreSql 或 Sqlite，我会收到预期的答案。

create_engine('sqlite:///cars.sqlite3') 好的！ create_engine('postgresql+psycopg2://xxx:yyy@localhost:5432/db_sql_alchemy') 好的！ create_engine('oracle+cx_oracle://xxx:yyy@localhost:49161/xe') 错误！

select、where 子句、表创建等其他操作在所有 3 个数据库中都按预期工作。

查看错误日志，查询似乎没有正确转换为 oracle 语法。我期待这样的事情：

从“汽车”中选择“汽车”。“ID”，“汽车”。“名称”，“汽车”。“价格” (BitAnd("汽车"."价格", 32768) > 0);

产生错误的操作是：

stm = stm.where(cars.c.Price.op('&')(0x8000) > 0)

我正在使用 Python 2.7.12 和 SQLAlchemy==1.1.2。

Answer 1

这是一个创建自定义运算符的示例，该运算符在不同的后端进行不同的编译：

from sqlalchemy.ext.compiler import compiles
from sqlalchemy.sql.elements import ColumnElement, Visitable, Grouping
from sqlalchemy.sql.operators import custom_op, is_precedent

class BitwiseAnd(ColumnElement):
    type = Integer()
    operator = custom_op("&", precedence=6)

    def __init__(self, left, right):
        if not isinstance(left, Visitable):
            left = bindparam("bitand", left, unique=True)
        if not isinstance(right, Visitable):
            right = bindparam("bitand", right, unique=True)
        self.left = left
        self.right = right

    def self_group(self, against=None):
        if is_precedent(self.operator, against):
            return Grouping(self)
        else:
            return self


@compiles(BitwiseAnd)
def _compile_bitwise_and(element, compiler, **kwargs):
    left = element.left.self_group(against=element.operator)
    right = element.right.self_group(against=element.operator)
    return compiler.process(element.operator(left, right))


@compiles(BitwiseAnd, "oracle")
def _compile_bitwise_and_oracle(element, compiler, **kwargs):
    return compiler.process(func.BITAND(element.left, element.right))

q = select([BitwiseAnd(BitwiseAnd(1, 2), BitwiseAnd(3, 4))])
print(q.compile(dialect=mysql.dialect()))
# SELECT (%s & %s) & (%s & %s) AS anon_1
print(q.compile(dialect=oracle.dialect()))
# SELECT BITAND(BITAND(:bitand_1, :bitand_2), BITAND(:bitand_3, :bitand_4)) AS anon_1 FROM DUAL