- marked as blocker
bind processor not working for basic clause?
Issue #2007
resolved
filter this test down to one that tests only condition #3 to confirm:
import sqlalchemy
import sqlalchemy.ext.declarative
Model = sqlalchemy.ext.declarative.declarative_base()
class StringIdTypeDecorator(sqlalchemy.types.TypeDecorator):
"""TypeDecorator that converts integer IDs to/from strings"""
impl = sqlalchemy.types.Integer
def process_bind_param(self, value, dialect=None):
assert isinstance(value, basestring), "Tried to query using %r" % value
return int(value[2:](2:))
def process_result_value(self, value, dialect):
assert isinstance(value, (int, long))
return "ID%s" % value
class StringIdType(sqlalchemy.types.Integer):
"""Type that converts integer IDs to/from strings"""
def bind_processor(self, _dialect):
def id_bind_process(value):
assert isinstance(value, basestring), "Tried to query using %r" % value
return int(value[2:](2:))
return id_bind_process
def result_processor(self, _dialect, _something):
def id_result_process(value):
assert isinstance(value, (int, long))
return "ID%s" % value
return id_result_process
# For some reason _BindParamClause changed since SQLAchemy 0.6beta1 to not
# sometimes not use the bind_processor when doing queries...
# Using this private function function will get this (see PROBLEM #3 below)
#def _coerce_compared_value(self, *args, **kwargs):
# return StringIdType()
class TestClass(object):
class MyModel(Model):
__tablename__ = "sqlalchemy_test_id_type"
__table_args__ = {'mysql_engine': 'InnoDB'}
# PROBLEM #1: SQLAlchemy does not make the ID auto increment if you use a TypeDecorator
#id = sqlalchemy.schema.Column(StringIdTypeDecorator, primary_key=True)
id = sqlalchemy.schema.Column(StringIdType, primary_key=True)
def __init__(self):
engine = sqlalchemy.create_engine('mysql://user@server/db', echo=True)
self.session_maker = sqlalchemy.orm.sessionmaker(engine)
def create_table(self):
"""Create the test table"""
session = self.session_maker()
self.MyModel.__table__.create(session.bind)
session.commit()
session.close()
def drop_table(self):
"""Drop the test table"""
session = self.session_maker()
self.MyModel.__table__.drop(session.bind)
session.commit()
session.close()
def test(self):
"""Test that type decorator is used when inserting after flush"""
session = self.session_maker()
model = self.MyModel()
session.add(model)
session.flush()
# PROBLEM #2: result_processor does not run. ID will be a long.
#assert isinstance(model.id, basestring), type(model.id)
# Save the ID and close the session
row_id = model.id
session.commit()
session.close()
# This will fail because row_id is a long and bind_processor expects a string (but see below)
#model = session.query(self.MyModel).filter_by(id=row_id).one()
# Cool, our result processor works on queries
model = session.query(self.MyModel).one()
assert isinstance(model.id, basestring), type(model.id)
row_id = model.id
session.close()
# PROBLEM #3: bind_processor does not run, so query fails with string.
# Note that when the specified id is an integer bind_processor will run
# This can be fixed by overriding _coerce_compared_value in the type
model = session.query(self.MyModel).filter_by(id=row_id).one()
if __name__ == '__main__':
test_class = TestClass()
try:
test_class.create_table()
test_class.test()
finally:
test_class.drop_table()
Comments (3)
-
reporter -
reporter - changed status to resolved
there's no bug here, non-decorated types don't coerce the other side by default. added docs to the private method in 94b5187b863217f95b8addb92bba9732f8ee9040 (0.6) 4e9e0f041c151cfa08d57773b49b254f1b4f8b6a (0.7).
#2006will address the type decorator use with last_inserted_id. -
reporter - removed milestone
Removing milestone: 0.6.6 (automated comment)
- Log in to comment