bind processor not working for basic clause?

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)
