import sqlalchemy.util as util
class TypeEngine(object):
+ basetypes = []
+ def __init__(self, *args, **kwargs):
+ pass
+ def _get_impl(self):
+ if hasattr(self, '_impl'):
+ return self._impl
+ else:
+ return NULLTYPE
+ def _set_impl(self, impl):
+ self._impl = impl
+ impl = property(_get_impl, _set_impl)
def get_col_spec(self):
- raise NotImplementedError()
+ return self.impl.get_col_spec()
def convert_bind_param(self, value, engine):
- raise NotImplementedError()
+ return self.impl.convert_bind_param(value, engine)
def convert_result_value(self, value, engine):
- raise NotImplementedError()
- def adapt(self, typeobj):
- """given a class that is a subclass of this TypeEngine's class, produces a new
- instance of that class with an equivalent state to this TypeEngine. The given
- class is a database-specific subclass which is obtained via a lookup dictionary,
- mapped against the class returned by the class_to_adapt() method."""
- return typeobj()
+ return self.impl.convert_result_value(value, engine)
+ def set_impl(self, impltype):
+ self.impl = impltype(**self.get_constructor_args())
+ def get_constructor_args(self):
+ return {}
def adapt_args(self):
- """Returns an instance of this TypeEngine instance's class, adapted according
- to the constructor arguments of this TypeEngine. Default return value is
- just this object instance."""
return self
- def class_to_adapt(self):
- """returns the class that should be sent to the adapt() method. This class
- will be used to lookup an approprate database-specific subclass."""
- return self.__class__
-# def __repr__(self):
- # return util.generic_repr(self)
-
+
def adapt_type(typeobj, colspecs):
- """given a generic type from this package, and a dictionary of
- "conversion" specs from a DB-specific package, adapts the type
- to a correctly-configured type instance from the DB-specific package."""
- if type(typeobj) is type:
+ if isinstance(typeobj, type):
typeobj = typeobj()
- # if the type is not a base type, i.e. not from our module, or its Null,
- # we return the type as is
- if (typeobj.__module__ != 'sqlalchemy.types' or typeobj.__class__ is NullTypeEngine) and not isinstance(typeobj, TypeDecorator):
- return typeobj
- typeobj = typeobj.adapt_args()
- t = typeobj.class_to_adapt()
- for t in t.__mro__[0:-1]:
+ t2 = typeobj.adapt_args()
+ for t in t2.__class__.__mro__[0:-1]:
try:
- return typeobj.adapt(colspecs[t])
- except KeyError, e:
+ impltype = colspecs[t]
+ break
+ except KeyError:
pass
- return typeobj.adapt(typeobj.__class__)
+ else:
+ # couldnt adapt...raise exception ?
+ return typeobj
+ typeobj.set_impl(impltype)
+ typeobj.impl.impl = NULLTYPE
+ return typeobj
class NullTypeEngine(TypeEngine):
- def __init__(self, *args, **kwargs):
- pass
def get_col_spec(self):
raise NotImplementedError()
def convert_bind_param(self, value, engine):
return value
class TypeDecorator(object):
- def get_col_spec(self):
- return self.extended.get_col_spec()
- def adapt(self, typeobj):
- if self.extended is self:
- t = self.__class__.__mro__[2]
- self.extended = t.adapt(self, typeobj)
- else:
- self.extended = self.extended.adapt(typeobj)
- return self
- def adapt_args(self):
- t = self.__class__.__mro__[2]
- self.extended = t.adapt_args(self)
- return self
- def class_to_adapt(self):
- return self.extended.__class__
+ """TypeDecorator is deprecated"""
+ pass
-class String(NullTypeEngine):
+
+class String(TypeEngine):
def __init__(self, length = None):
self.length = length
- def adapt(self, typeobj):
- return typeobj(self.length)
- def adapt_args(self):
- if self.length is None:
- return TEXT()
- else:
- return self
+ def get_constructor_args(self):
+ return {'length':self.length}
def convert_bind_param(self, value, engine):
if not engine.convert_unicode or value is None or not isinstance(value, unicode):
return value
return value
else:
return value.decode('utf-8')
-
-class Unicode(TypeDecorator,String):
- def __init__(self, length=None):
- String.__init__(self, length)
+ def adapt_args(self):
+ if self.length is None:
+ return TEXT()
+ else:
+ return self
+
+class Unicode(String):
def convert_bind_param(self, value, engine):
if isinstance(value, unicode):
return value.encode('utf-8')
else:
return value
-class Integer(NullTypeEngine):
+class Integer(TypeEngine):
"""integer datatype"""
- # TODO: do string bind params need int(value) performed before sending ?
- # seems to be not needed with SQLite, Postgres
pass
-
-class Smallinteger(Integer):
+
+class SmallInteger(Integer):
""" smallint datatype """
pass
-
-class Numeric(NullTypeEngine):
+Smallinteger = SmallInteger
+
+class Numeric(TypeEngine):
def __init__(self, precision = 10, length = 2):
self.precision = precision
self.length = length
- def adapt(self, typeobj):
- return typeobj(self.precision, self.length)
+ def get_constructor_args(self):
+ return {'precision':self.precision, 'length':self.length}
-class Float(NullTypeEngine):
+class Float(TypeEngine):
def __init__(self, precision = 10):
self.precision = precision
- def adapt(self, typeobj):
- return typeobj(self.precision)
+ def get_constructor_args(self):
+ return {'precision':self.precision}
-class DateTime(NullTypeEngine):
+class DateTime(TypeEngine):
pass
-class Date(NullTypeEngine):
+class Date(TypeEngine):
pass
-class Time(NullTypeEngine):
+class Time(TypeEngine):
pass
-class Binary(NullTypeEngine):
+class Binary(TypeEngine):
def __init__(self, length=None):
self.length = length
def convert_bind_param(self, value, engine):
return engine.dbapi().Binary(value)
def convert_result_value(self, value, engine):
return value
- def adapt(self, typeobj):
- return typeobj(self.length)
+ def get_constructor_args(self):
+ return {'length':self.length}
-class Boolean(NullTypeEngine):
+class Boolean(TypeEngine):
pass
class FLOAT(Float):pass
def adapt_args(self):
return self
-class MyDecoratedType(types.TypeDecorator, types.String):
+class MyDecoratedType(types.String):
def convert_bind_param(self, value, engine):
return "BIND_IN"+ value
def convert_result_value(self, value, engine):
return value + "BIND_OUT"
+class MyUnicodeType(types.Unicode):
+ def convert_bind_param(self, value, engine):
+ return "UNI_BIND_IN"+ value
+ def convert_result_value(self, value, engine):
+ return value + "UNI_BIND_OUT"
+
class OverrideTest(PersistTest):
"""tests user-defined types, including a full type as well as a TypeDecorator"""
def testprocessing(self):
global users
- users.insert().execute(user_id = 2, goofy = 'jack', goofy2='jack', goofy3='jack')
- users.insert().execute(user_id = 3, goofy = 'lala', goofy2='lala', goofy3='lala')
- users.insert().execute(user_id = 4, goofy = 'fred', goofy2='fred', goofy3='fred')
+ 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')
l = users.select().execute().fetchall()
print repr(l)
- self.assert_(l == [(2, 'BIND_INjackBIND_OUT', 'BIND_INjackBIND_OUT', 'BIND_INjackBIND_OUT'), (3, 'BIND_INlalaBIND_OUT', 'BIND_INlalaBIND_OUT', 'BIND_INlalaBIND_OUT'), (4, 'BIND_INfredBIND_OUT', 'BIND_INfredBIND_OUT', 'BIND_INfredBIND_OUT')])
+ 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')])
def setUpAll(self):
global users
# decorated type without an argument, it will adapt_args to TEXT
Column('goofy3', MyDecoratedType, nullable = False),
+
+ Column('goofy4', MyUnicodeType, nullable = False),
+
)
users.create()
projects / thirdparty / sqlalchemy / sqlalchemy.git / commitdiff
