Python-Sqlalchemy Binary Column Type HEX() and UNHEX()

Question

I'm attempting to learn Sqlalchemy and utilize an ORM. One of my columns stores file hashes as binary. In SQL, the select would simply be

SELECT type, column FROM table WHERE hash = UNHEX('somehash')

How do I achieve a select like this (ideally with an insert example, too) using my ORM? I've begun reading about column overrides, but I'm confused/not certain that that's really what I'm after.

eg res = session.query.filter(Model.hash == __something__? )

Thoughts?

Answer 1

Only for select's and insert's

Well, for select you could use:

>>> from sqlalchemy import func
>>> session = (...)
>>> (...)
>>> engine = create_engine('sqlite:///:memory:', echo=True)
>>> q = session.query(Model.id).filter(Model.some == func.HEX('asd'))
>>> print q.statement.compile(bind=engine)
SELECT model.id
FROM model
WHERE model.some = HEX(?)

For insert:

>>> from sqlalchemy import func
>>> session = (...)
>>> (...)
>>> engine = create_engine('sqlite:///:memory:', echo=True)
>>> m = new Model(hash=func.HEX('asd'))
>>> session.add(m)
>>> session.commit()
INSERT INTO model (hash) VALUES (HEX(%s))

A better approach: Custom column that converts data by using sql functions

But, I think the best for you is a custom column on sqlalchemy using any process_bind_param, process_result_value, bind_expression and column_expression see this example.

Check this code below, it create a custom column that I think fit your needs:

from sqlalchemy.types import VARCHAR
from sqlalchemy import func

class HashColumn(VARCHAR):

    def bind_expression(self, bindvalue):
        # convert the bind's type from String to HEX encoded 
        return func.HEX(bindvalue)

    def column_expression(self, col):
        # convert select value from HEX encoded to String
        return func.UNHEX(col)

You could model your a table like:

from sqlalchemy import Column, types
from sqlalchemy.ext.declarative import declarative_base

Base = declarative_base()

class Model(Base):
    __tablename__ = "model"
    id = Column(types.Integer, primary_key=True)
    col = Column(HashColumn(20))

    def __repr__(self):
        return "Model(col=%r)" % self.col

Some usage:

>>> (...)
>>> session = create_session(...)
>>> (...)
>>> model = Model(col='Iuri Diniz')
>>> session.add(model)
>>> session.commit()

this issues this query:

INSERT INTO model (col) VALUES (HEX(?)); -- ('Iuri Diniz',)

More usage:

>>> session.query(Model).first()
Model(col='Iuri Diniz')

this issues this query:

SELECT 
    model.id AS model_id, UNHEX(model.col) AS model_col 
FROM model 
LIMIT ? ; -- (1,)

A bit more:

>>> session.query(Model).filter(Model.col == "Iuri Diniz").first()
Model(col='Iuri Diniz')

this issues this query:

SELECT 
    model.id AS model_id, UNHEX(model.col) AS model_col 
FROM model 
WHERE model.col = HEX(?) 
LIMIT ? ; -- ('Iuri Diniz', 1)

Extra: Custom column that converts data by using python types

Maybe you want to use some beautiful custom type and want to convert it between python and the database.

In the following example I convert UUID's between python and the database (the code is based on this link):

import uuid
from sqlalchemy.types import TypeDecorator, VARCHAR

class UUID4(TypeDecorator):
    """Portable UUID implementation

    >>> str(UUID4())
    'VARCHAR(36)'
    """

    impl = VARCHAR(36)

    def process_bind_param(self, value, dialect):
        if value is None:
            return value
        else:
            if not isinstance(value, uuid.UUID):
                return str(uuid.UUID(value))
            else:
                # hexstring
                return str(value)

    def process_result_value(self, value, dialect):
        if value is None:
            return value
        else:
            return uuid.UUID(value)