Python-Sqlalchemy二进制列类型HEX()和UNHEX()

Question

Python-Sqlalchemy二进制列类型HEX()和UNHEX()

mysqlpython-2.7sqlalchemy

4

我正尝试学习Sqlalchemy并使用ORM。我的一个列存储文件哈希作为二进制。在SQL中，选择将简单地是：

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

如何使用我的ORM实现此类选择（最好包括插入示例）？我已经开始阅读有关列覆盖的文章，但我感到困惑/不确定那是否真正符合我的需求。

例如：res = session.query.filter(Model.hash == __something__? )

您有什么想法吗？

- geudrik

3个回答

1

我非常喜欢iuridiniz approach更好的方法：使用SQL函数自定义列来转换数据，但是当在MySQL 5.7中使用BINARY和VARBINARY存储十六进制字符串时，我遇到了一些问题。我尝试了不同的方法，但是SQLAlchemy一直抱怨编码和/或在无法使用它们的上下文中使用func.HEX和func.UNHEX。使用python3和SQLAlchemy 1.2.8，我扩展了基类并替换了其处理器，以使SQLAlchemy不需要从数据库中获取函数来绑定数据并计算结果，而是在Python中完成，如下所示：

import codecs
from sqlalchemy.types import VARBINARY

class VarBinaryHex(VARBINARY):
    """Extend VARBINARY to handle hex strings."""

    impl = VARBINARY

    def bind_processor(self, dialect):
        """Return a processor that decodes hex values."""
        def process(value):
            return codecs.decode(value, 'hex')
        return process

    def result_processor(self, dialect, coltype):
        """Return a processor that encodes hex values."""
        def process(value):
            return codecs.encode(value, 'hex')
        return process

    def adapt(self, impltype):
        """Produce an adapted form of this type, given an impl class."""
        return VarBinaryHex()

这个想法是用Python函数替换需要DBMS干预的HEX和UNHEX，这些函数就像HEX和UNHEX一样编码和解码十六进制字符串。如果你直接连接数据库，可以使用HEX和UNHEX，但从SQLAlchemy开始，codecs.enconde和codecs.decode函数为你完成了这项工作。

我敢打赌，如果有人感兴趣，编写适当的处理器，甚至可以从Python的角度将十六进制值管理为整数，允许存储大于BIGINT的整数。

一些考虑事项：

如果十六进制字符串的长度已知，可以使用BINARY代替VARBINARY。
根据您要执行的操作，可能值得在将使用此类型列的类的构造函数中取消/大写字符串，以便在对象初始化的那一刻使用一致的大小写。例如，'aa' != 'AA'但是0xaa == 0xAA。
如前所述，您可以考虑一个处理器，将数据库二进制十六进制值转换为Python整数。
使用VARBINARY时要小心，因为'aa' != '00aa'
如果使用BINARY，假设您的列为col = Column(BinaryHex(length=4))，请注意，任何少于length字节的值都将用零完成。我的意思是，如果你这样做obj.col = 'aabb'并提交它，当你稍后从数据中检索它时，你会得到obj.col == 'aabb0000'，这是完全不同的东西。

- TrilceAC

1

由于以下错误，我无法使@iuridiniz的自定义列解决方案工作：

sqlalchemy.exc.StatementError: (builtins.TypeError) encoding without a string argument

对于类似以下表达式：

m = Model(col='FFFF')
session.add(m)
session.commit()

我通过重写process_bind_param来解决这个问题，它会在将参数传递给bind_expression进行插值到查询语言之前对其进行处理。

from sqlalchemy.types import VARCHAR
from sqlalchemy import func

class HashColumn(VARCHAR):

    def process_bind_param(self, value, dialect):
        # encode value as a binary
        if value:
            return bytes(value, 'utf-8')

    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)

然后定义表格是相同的：

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

- muirbot

网页内容由stack overflow 提供, 点击上面的

可以查看英文原文，
原文链接

- iuridiniz · Accepted Answer

仅适用于选择和插入

对于选择，您可以使用：

>>> 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(?)

用于插入：

>>> 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))

更好的方法：使用SQL函数转换数据的自定义列

但是，我认为最好的方法是使用任何process_bind_param, process_result_value, bind_expression和column_expression来创建SQLAlchemy上的自定义列。请参考示例。

请查看下面的代码，它创建了一个自定义列，我认为它符合您的需求:

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)

你可以像这样建模你的表格：

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

一些用法：

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

这个问题发出了这个查询：

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

更多用法：

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

这个问题发出了这个查询：

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

稍微多一点：

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

这个问题发出了这个查询：

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

额外内容：使用Python类型的自定义列转换数据

也许您想使用一些漂亮的自定义类型，并希望在Python和数据库之间进行转换。

在下面的示例中，我将UUID在Python和数据库之间进行转换（代码基于此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)