and TypeDecorator classes which define convert_bind_param()/convert_result_value()
will continue to function. Also supports calling the super() version of
those methods.
- Fix to bind param processing such that "False" values (like blank strings)
still get processed/encoded.
+- Added a "legacy" adapter to types, such that user-defined TypeEngine
+ and TypeDecorator classes which define convert_bind_param()/convert_result_value()
+ will continue to function. Also supports calling the super() version of
+ those methods.
+
- Added session.prune(), trims away instances cached in a session that are no
longer referenced elsewhere. (A utility for strong-ref identity maps).
attribute_manager.reset_class_managed(self.class_)
oldinit = self.class_.__init__
+ if oldinit is object.__init__:
+ oldinit = None
+
def init(instance, *args, **kwargs):
self.compile()
self.extension.init_instance(self, self.class_, oldinit, instance, args, kwargs)
from sqlalchemy import exceptions
from sqlalchemy.util import Decimal, pickle
+class _UserTypeAdapter(type):
+ """adapts 0.3 style user-defined types with convert_bind_param/convert_result_value
+ to use newer bind_processor()/result_processor() methods."""
+
+ def __init__(cls, clsname, bases, dict):
+ if not hasattr(cls.convert_result_value, '_sa_override'):
+ cls.__instrument_result_proc(cls)
+
+ if not hasattr(cls.convert_bind_param, '_sa_override'):
+ cls.__instrument_bind_proc(cls)
+
+ return super(_UserTypeAdapter, cls).__init__(clsname, bases, dict)
+
+ def __instrument_bind_proc(cls, class_):
+ def bind_processor(self, dialect):
+ def process(value):
+ return self.convert_bind_param(value, dialect)
+ return process
+ class_.super_bind_processor = class_.bind_processor
+ class_.bind_processor = bind_processor
+
+ def __instrument_result_proc(cls, class_):
+ def result_processor(self, dialect):
+ def process(value):
+ return self.convert_result_value(value, dialect)
+ return process
+ class_.super_result_processor = class_.result_processor
+ class_.result_processor = result_processor
+
+
class AbstractType(object):
+ __metaclass__ = _UserTypeAdapter
+
def __init__(self, *args, **kwargs):
pass
def copy_value(self, value):
return value
+ def convert_result_value(self, value, dialect):
+ """legacy convert_result_value() method.
+
+ this method is only used when called via a user-defined
+ subclass' own convert_result_value() method, and adapts
+ the call to use the result_processor() callable.
+
+ The method is configured at class definition time
+ by a legacy adapter metaclass, and
+ will not work with a subclass that does not
+ define a convert_result_value() method of its own.
+ """
+ return self.super_result_processor(dialect)(value)
+ convert_result_value._sa_override = True
+
+ def convert_bind_param(self, value, dialect):
+ """legacy convert_bind_param() method.
+
+ this method is only used when called via a user-defined
+ subclass' own convert_bind_param() method, and adapts
+ the call to use the bind_processor() callable.
+
+ The method is configured at class definition time
+ by a legacy adapter metaclass, and
+ will not work with a subclass that does not
+ define a convert_bind_param() method of its own.
+ """
+ return self.super_bind_processor(dialect)(value)
+ convert_bind_param._sa_override = True
+
+ def bind_processor(self, dialect):
+ """define a bind parameter processing function."""
+
+ return None
+
+ def result_processor(self, dialect):
+ """define a result-column processing function."""
+
+ return None
+
def compare_values(self, x, y):
+ """compare two values for equality."""
+
return x == y
def is_mutable(self):
+ """return True if the target Python type is 'mutable'.
+
+ This allows systems like the ORM to know if an object
+ can be considered 'not changed' by identity alone.
+ """
+
return False
def get_dbapi_type(self, dbapi):
return None
def adapt(self, cls):
- return cls()
+ return cls()
def get_search_list(self):
"""return a list of classes to test for a match
warnings.warn(logging.SADeprecationWarning("Call to deprecated function %s" % func.__name__),
stacklevel=2)
return func(*args, **kwargs)
- func_with_warning.__doc__ = (add_deprecation_to_docstring and 'Deprecated.\n' or '') + func.__doc__
+ func_with_warning.__doc__ = (add_deprecation_to_docstring and 'Deprecated.\n' or '') + str(func.__doc__)
func_with_warning.__dict__.update(func.__dict__)
try:
func_with_warning.__name__ = func.__name__
def copy(self):
return MyUnicodeType(self.impl.length)
+class LegacyType(types.TypeEngine):
+ def get_col_spec(self):
+ return "VARCHAR(100)"
+ def convert_bind_param(self, value, dialect):
+ return "BIND_IN"+ value
+ def convert_result_value(self, value, dialect):
+ return value + "BIND_OUT"
+ def adapt(self, typeobj):
+ return typeobj()
+
+class LegacyUnicodeType(types.TypeDecorator):
+ impl = Unicode
+
+ def convert_bind_param(self, value, dialect):
+ return "UNI_BIND_IN" + super(LegacyUnicodeType, self).convert_bind_param(value, dialect)
+
+ def convert_result_value(self, value, dialect):
+ return super(LegacyUnicodeType, self).convert_result_value(value, dialect) + "UNI_BIND_OUT"
+
+ def copy(self):
+ return LegacyUnicodeType(self.impl.length)
+
class AdaptTest(PersistTest):
def testadapt(self):
e1 = url.URL('postgres').get_dialect()()
assert isinstance(dialect.type_descriptor(t2), mysql.MSVarBinary)
-class OverrideTest(PersistTest):
- """tests user-defined types, including a full type as well as a TypeDecorator"""
+class UserDefinedTest(PersistTest):
+ """tests user-defined types."""
def testbasic(self):
print users.c.goofy4.type
def testprocessing(self):
global users
- users.insert().execute(user_id = 2, goofy = 'jack', goofy2='jack', goofy3='jack', goofy4='jack')
- users.insert().execute(user_id = 3, goofy = 'lala', goofy2='lala', goofy3='lala', goofy4='lala')
- users.insert().execute(user_id = 4, goofy = 'fred', goofy2='fred', goofy3='fred', goofy4='fred')
+ users.insert().execute(user_id = 2, goofy = 'jack', goofy2='jack', goofy3='jack', goofy4='jack', goofy5='jack', goofy6='jack')
+ users.insert().execute(user_id = 3, goofy = 'lala', goofy2='lala', goofy3='lala', goofy4='lala', goofy5='lala', goofy6='lala')
+ users.insert().execute(user_id = 4, goofy = 'fred', goofy2='fred', goofy3='fred', goofy4='fred', goofy5='fred', goofy6='fred')
l = users.select().execute().fetchall()
- print repr(l)
- self.assert_(l == [(2, 'BIND_INjackBIND_OUT', 'BIND_INjackBIND_OUT', 'BIND_INjackBIND_OUT', u'UNI_BIND_INjackUNI_BIND_OUT'), (3, 'BIND_INlalaBIND_OUT', 'BIND_INlalaBIND_OUT', 'BIND_INlalaBIND_OUT', u'UNI_BIND_INlalaUNI_BIND_OUT'), (4, 'BIND_INfredBIND_OUT', 'BIND_INfredBIND_OUT', 'BIND_INfredBIND_OUT', u'UNI_BIND_INfredUNI_BIND_OUT')])
+ assert l == [
+ (2, 'BIND_INjackBIND_OUT', 'BIND_INjackBIND_OUT', 'BIND_INjackBIND_OUT', u'UNI_BIND_INjackUNI_BIND_OUT', u'UNI_BIND_INjackUNI_BIND_OUT', 'BIND_INjackBIND_OUT'),
+ (3, 'BIND_INlalaBIND_OUT', 'BIND_INlalaBIND_OUT', 'BIND_INlalaBIND_OUT', u'UNI_BIND_INlalaUNI_BIND_OUT', u'UNI_BIND_INlalaUNI_BIND_OUT', 'BIND_INlalaBIND_OUT'),
+ (4, 'BIND_INfredBIND_OUT', 'BIND_INfredBIND_OUT', 'BIND_INfredBIND_OUT', u'UNI_BIND_INfredUNI_BIND_OUT', u'UNI_BIND_INfredUNI_BIND_OUT', 'BIND_INfredBIND_OUT')
+ ]
def setUpAll(self):
- global users
- users = Table('type_users', MetaData(testbase.db),
+ global users, metadata
+ metadata = MetaData(testbase.db)
+ users = Table('type_users', metadata,
Column('user_id', Integer, primary_key = True),
# totall custom type
Column('goofy', MyType, nullable = False),
Column('goofy3', MyDecoratedType, nullable = False),
Column('goofy4', MyUnicodeType, nullable = False),
+ Column('goofy5', LegacyUnicodeType, nullable = False),
+ Column('goofy6', LegacyType, nullable = False),
)
- users.create()
+ metadata.create_all()
+
def tearDownAll(self):
- global users
- users.drop()
-
+ metadata.drop_all()
class ColumnsTest(AssertMixin):
projects / thirdparty / sqlalchemy / sqlalchemy.git / commitdiff
