python - SQLAlchemy 转换为二进制(N)

Question

只有我一个人遇到这个问题吗？

我想编写与 MySQL 上运行的 SQL 等效的表达式:

select * from table1 where table1.some_column = cast('xyz' as Binary(32))

我在 SQLAlchemy 中尝试的是:

session.Query(table1)
.filter(table1.some_column == cast('xyz', Binary(length=32))
.all()

但是，这被错误地翻译为:

select * from table1 where table1.some_column = cast('xyz' as Binary)

有什么想法吗？

我尝试了类似的方法，但使用数字而不是二进制类型，并且它按预期工作。

Answer 1

您不必更改 SQLAlchemy 源代码即可以不同方式编译某些内容。将 Binary 编译为 BINARY 的相关代码为 here :

class MySQLCompiler(compiler.SQLCompiler):
    ...

    def visit_typeclause(self, typeclause, type_=None):
        if type_ is None:
            type_ = typeclause.type.dialect_impl(self.dialect)
        if isinstance(type_, sqltypes.TypeDecorator):
            return self.visit_typeclause(typeclause, type_.impl)
        elif isinstance(type_, sqltypes.Integer):
            if getattr(type_, 'unsigned', False):
                return 'UNSIGNED INTEGER'
            else:
                return 'SIGNED INTEGER'
        elif isinstance(type_, sqltypes.TIMESTAMP):
            return 'DATETIME'
        elif isinstance(type_, (sqltypes.DECIMAL, sqltypes.DateTime,
                                sqltypes.Date, sqltypes.Time)):
            return self.dialect.type_compiler.process(type_)
        elif isinstance(type_, sqltypes.String) \
                and not isinstance(type_, (ENUM, SET)):
            adapted = CHAR._adapt_string_for_cast(type_)
            return self.dialect.type_compiler.process(adapted)
        elif isinstance(type_, sqltypes._Binary):
            return 'BINARY'  # <-------------------------------- RIGHT HERE
        elif isinstance(type_, sqltypes.JSON):
            return "JSON"
        elif isinstance(type_, sqltypes.NUMERIC):
            return self.dialect.type_compiler.process(
                type_).replace('NUMERIC', 'DECIMAL')
        else:
            return None

所以，我们只需要用我们自己的编译逻辑来拦截它:

from sqlalchemy.sql.elements import TypeClause
from sqlalchemy.types import _Binary
from sqlalchemy.ext.compiler import compiles

@compiles(TypeClause, "mysql")
def _compile_typeclause(element, compiler, **kwargs):
    if isinstance(element.type, _Binary):
        t = "BINARY"
        if element.type.length:
            t += "({})".format(element.type.length)
        return t
    return compiler.visit_typeclause(element, **kwargs)