CODEWITHSUNDEEP
pythonmysqlsqlalchemy
buhtla
buhtla

Reputation: 2909

SQLAlchemy cast to binary(N)

Am I the only one who is having this problem?

I want to write expression which will be equivalent to this sql running on MySQL:

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

What I was trying in SQLAlchemy was:

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

However, this is incorrectly translated into:

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

Any ideas?

I tried similar but using Numeric instead of Binary type, and it worked as expected.

Upvotes: 2

Views: 1291

Answers (1)

univerio
univerio

Reputation: 20518

You do not have to change the SQLAlchemy source to get it to compile something differently. The relevant code for compiling Binary into BINARY is 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

So, we just need to intercept that with our own compilation logic:

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)

Upvotes: 2

Related Questions