仅适用于选择和插入
对于选择,您可以使用:
>>> 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):
return func.HEX(bindvalue)
def column_expression(self, col):
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(?));
更多用法:
>>> session.query(Model).first()
Model(col='Iuri Diniz')
这个问题发出了这个查询:
SELECT
model.id AS model_id, UNHEX(model.col) AS model_col
FROM model
LIMIT ? ;
稍微多一点:
>>> 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 ? ;
额外内容:使用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:
return str(value)
def process_result_value(self, value, dialect):
if value is None:
return value
else:
return uuid.UUID(value)
func.HEX/UNHEX
,它返回一个函数对象而不是一个值:func.UNHEX(99)
=><sqlalchemy.sql.functions.Function at 0x7f3e7d2b8c90; UNHEX>
。你可以对_任何东西_获得相同的输出:func.IJUSTFUDGEDYOURMAMA(99)
=><sqlalchemy.sql.functions.Function at 0x7fe72c03b350; IJUSTFUDGEDYOURMAMA>
。 - Matthew Trevor