# This module is part of SQLAlchemy and is released under
# the MIT License: http://www.opensource.org/licenses/mit-license.php
-"""defines genericized SQL types, each represented by a subclass of
-[sqlalchemy.types#AbstractType]. Dialects define further subclasses of these
+"""defines genericized SQL types, each represented by a subclass of
+[sqlalchemy.types#AbstractType]. Dialects define further subclasses of these
types.
For more information see the SQLAlchemy documentation on types.
'BOOLEAN', 'SMALLINT', 'DATE', 'TIME',
'String', 'Integer', 'SmallInteger','Smallinteger',
'Numeric', 'Float', 'DateTime', 'Date', 'Time', 'Binary',
- 'Boolean', 'Unicode', 'PickleType', 'Interval',
+ 'Boolean', 'Unicode', 'UnicodeText', 'PickleType', 'Interval',
'type_map'
]
by default calls dialect.type_descriptor(self.impl), but
can be overridden to provide different behavior.
"""
-
+
return dialect.type_descriptor(self.impl)
def __getattr__(self, key):
def process_bind_param(self, value, dialect):
raise NotImplementedError()
-
+
def process_result_value(self, value, dialect):
raise NotImplementedError()
-
+
def bind_processor(self, dialect):
if self.__class__.process_bind_param.func_code is not TypeDecorator.process_bind_param.func_code:
impl_processor = self.impl.bind_processor(dialect)
class String(Concatenable, TypeEngine):
"""A sized string type.
-
+
Usually corresponds to VARCHAR. Can also take Python unicode objects
- and encode to the database's encoding in bind params (and the reverse for
+ and encode to the database's encoding in bind params (and the reverse for
result sets.)
-
- a String with no length will adapt itself automatically to a Text
+
+ a String with no length will adapt itself automatically to a Text
object at the dialect level (this behavior is deprecated in 0.4).
"""
def __init__(self, length=None, convert_unicode=False, assert_unicode=None):
if _for_ddl and self.length is None:
warn_deprecated("Using String type with no length for CREATE TABLE is deprecated; use the Text type explicitly")
return TypeEngine.dialect_impl(self, dialect, **kwargs)
-
+
def get_search_list(self):
l = super(String, self).get_search_list()
# if we are String or Unicode with no length,
def get_dbapi_type(self, dbapi):
return dbapi.STRING
+class Text(String):
+ def dialect_impl(self, dialect, **kwargs):
+ return TypeEngine.dialect_impl(self, dialect, **kwargs)
+
class Unicode(String):
"""A synonym for String(length, convert_unicode=True, assert_unicode='warn')."""
-
+
def __init__(self, length=None, **kwargs):
kwargs['convert_unicode'] = True
kwargs['assert_unicode'] = 'warn'
super(Unicode, self).__init__(length=length, **kwargs)
+class UnicodeText(Text):
+ """A synonym for Text(convert_unicode=True, assert_unicode='warn')."""
+
+ def __init__(self, length=None, **kwargs):
+ kwargs['convert_unicode'] = True
+ kwargs['assert_unicode'] = 'warn'
+ super(UnicodeText, self).__init__(length=length, **kwargs)
+
class Integer(TypeEngine):
"""Integer datatype."""
class SmallInteger(Integer):
"""Smallint datatype."""
- pass
-
Smallinteger = SmallInteger
class Numeric(TypeEngine):
"""Numeric datatype, usually resolves to DECIMAL or NUMERIC."""
-
+
def __init__(self, precision=10, length=2, asdecimal=True):
self.precision = precision
self.length = length
return process
class FLOAT(Float): pass
-class Text(String):
- def dialect_impl(self, dialect, **kwargs):
- return TypeEngine.dialect_impl(self, dialect, **kwargs)
-
+
TEXT = Text
class NUMERIC(Numeric): pass
class DECIMAL(Numeric): pass
dt.timedelta : Interval,
type(None): NullType
}
-
import testbase
-import sets
+import sets, warnings
from sqlalchemy import *
from sqlalchemy import sql, exceptions
from sqlalchemy.databases import mysql
specs = [( String(), mysql.MSText(), ),
( String(1), mysql.MSString(1), ),
( String(3), mysql.MSString(3), ),
+ ( Text(), mysql.MSText(), ),
+ ( Unicode(), mysql.MSText(), ),
+ ( Unicode(1), mysql.MSString(1), ),
+ ( Unicode(3), mysql.MSString(3), ),
+ ( UnicodeText(), mysql.MSText(), ),
( mysql.MSChar(1), ),
( mysql.MSChar(3), ),
( NCHAR(2), mysql.MSChar(2), ),
m = MetaData(db)
t_table = Table('mysql_types', m, *columns)
try:
- m.create_all()
+ try:
+ warnings.filterwarnings('ignore',
+ 'Using String type with no length.*')
+ m.create_all()
+ finally:
+ warnings.filterwarnings("always",
+ 'Using String type with no length.*')
m2 = MetaData(db)
rt = Table('mysql_types', m2, autoload=True)
(String, "CAST(t.col AS CHAR)"),
(Unicode, "CAST(t.col AS CHAR)"),
+ (UnicodeText, "CAST(t.col AS CHAR)"),
(VARCHAR, "CAST(t.col AS CHAR)"),
(NCHAR, "CAST(t.col AS CHAR)"),
(CHAR, "CAST(t.col AS CHAR)"),
impl = String
def copy(self):
return MyDecoratedType()
-
+
col = Column('', MyDecoratedType)
dialect_type = col.type.dialect_impl(dialect)
assert isinstance(dialect_type.impl, oracle.OracleText), repr(dialect_type.impl)
(oracle_dialect, VARCHAR(), oracle.OracleString),
(oracle_dialect, String(50), oracle.OracleString),
(oracle_dialect, Unicode(), oracle.OracleText),
+ (oracle_dialect, UnicodeText(), oracle.OracleText),
(oracle_dialect, NCHAR(), oracle.OracleString),
(mysql_dialect, String(), mysql.MSText),
(mysql_dialect, VARCHAR(), mysql.MSString),
(mysql_dialect, String(50), mysql.MSString),
(mysql_dialect, Unicode(), mysql.MSText),
+ (mysql_dialect, UnicodeText(), mysql.MSText),
(mysql_dialect, NCHAR(), mysql.MSNChar),
(postgres_dialect, String(), postgres.PGText),
(postgres_dialect, VARCHAR(), postgres.PGString),
(postgres_dialect, String(50), postgres.PGString),
(postgres_dialect, Unicode(), postgres.PGText),
+ (postgres_dialect, UnicodeText(), postgres.PGText),
(postgres_dialect, NCHAR(), postgres.PGString),
(firebird_dialect, String(), firebird.FBText),
(firebird_dialect, VARCHAR(), firebird.FBString),
(firebird_dialect, String(50), firebird.FBString),
(firebird_dialect, Unicode(), firebird.FBText),
+ (firebird_dialect, UnicodeText(), firebird.FBText),
(firebird_dialect, NCHAR(), firebird.FBString),
]:
assert isinstance(start.dialect_impl(dialect), test), "wanted %r got %r" % (test, start.dialect_impl(dialect))
[1200, 1500, 900],
[1800, 2250, 1350],
l
-
+
):
for col in row[1:8]:
self.assertEquals(col, assertstr)
self.assertEquals(row[9], assertint2)
for col in (row[4], row[5], row[7]):
assert isinstance(col, unicode)
-
+
def setUpAll(self):
global users, metadata
unicode_table = Table('unicode_table', metadata,
Column('id', Integer, Sequence('uni_id_seq', optional=True), primary_key=True),
Column('unicode_varchar', Unicode(250)),
- Column('unicode_text', Unicode),
+ Column('unicode_text', UnicodeText),
Column('plain_varchar', String(250))
)
unicode_table.create()
self.assert_(isinstance(x['unicode_text'], unicode) and x['unicode_text'] == unicodedata)
if isinstance(x['plain_varchar'], unicode):
# SQLLite and MSSQL return non-unicode data as unicode
- self.assert_(testbase.db.name in ('sqlite', 'mssql'))
+ self.assert_(testing.against('sqlite', 'mssql'))
self.assert_(x['plain_varchar'] == unicodedata)
print "it's %s!" % testbase.db.name
else:
self.assert_(not isinstance(x['plain_varchar'], unicode) and x['plain_varchar'] == rawdata)
-
+
def testassert(self):
import warnings
warnings.filterwarnings("always", r".*non-unicode bind")
-
- # test that data still goes in if warning is emitted....
- unicode_table.insert().execute(unicode_varchar='im not unicode')
- assert select([unicode_table.c.unicode_varchar]).execute().fetchall() == [('im not unicode', )]
-
+
+ ## test that data still goes in if warning is emitted....
+ unicode_table.insert().execute(unicode_varchar='not unicode')
+ assert (select([unicode_table.c.unicode_varchar]).execute().fetchall()
+ == [('not unicode', )])
+
warnings.filterwarnings("error", r".*non-unicode bind")
-
try:
- unicode_table.insert().execute(unicode_varchar='im not unicode')
- assert False
- except RuntimeWarning, e:
- assert str(e) == "Unicode type received non-unicode bind param value 'im not unicode'", str(e)
+ try:
+ unicode_table.insert().execute(unicode_varchar='not unicode')
+ assert False
+ except RuntimeWarning, e:
+ assert str(e) == "Unicode type received non-unicode bind param value 'not unicode'", str(e)
+ finally:
+ warnings.filterwarnings("always", r".*non-unicode bind")
- unicode_engine = engines.utf8_engine(options={'convert_unicode':True, 'assert_unicode':True})
+ unicode_engine = engines.utf8_engine(options={'convert_unicode':True,
+ 'assert_unicode':True})
try:
try:
unicode_engine.execute(unicode_table.insert(), plain_varchar='im not unicode')
finally:
unicode_engine.dispose()
- @testing.unsupported('oracle')
+ @testing.fails_on('oracle')
def testblanks(self):
unicode_table.insert().execute(unicode_varchar=u'')
assert select([unicode_table.c.unicode_varchar]).scalar() == u''
return process
def adapt_operator(self, op):
return {operators.add:operators.sub, operators.sub:operators.add}.get(op, op)
-
+
meta = MetaData(testbase.db)
- test_table = Table('test', meta,
+ test_table = Table('test', meta,
Column('id', Integer, primary_key=True),
Column('data', String(30)),
Column('timestamp', Date),
Column('value', MyCustomType))
-
+
meta.create_all()
-
+
test_table.insert().execute({'id':1, 'data':'somedata', 'timestamp':datetime.date(2007, 10, 15), 'value':25})
-
+
def tearDownAll(self):
meta.drop_all()
-
+
def test_control(self):
assert testbase.db.execute("select value from test").scalar() == 250
-
+
assert test_table.select().execute().fetchall() == [(1, 'somedata', datetime.date(2007, 10, 15), 25)]
-
+
def test_bind_adapt(self):
expr = test_table.c.timestamp == bindparam("thedate")
assert expr.right.type.__class__ == test_table.c.timestamp.type.__class__
-
+
assert testbase.db.execute(test_table.select().where(expr), {"thedate":datetime.date(2007, 10, 15)}).fetchall() == [(1, 'somedata', datetime.date(2007, 10, 15), 25)]
expr = test_table.c.value == bindparam("somevalue")
assert expr.right.type.__class__ == test_table.c.value.type.__class__
assert testbase.db.execute(test_table.select().where(expr), {"somevalue":25}).fetchall() == [(1, 'somedata', datetime.date(2007, 10, 15), 25)]
-
+
def test_operator_adapt(self):
"""test type-based overloading of operators"""
-
+
# test string concatenation
expr = test_table.c.data + "somedata"
assert testbase.db.execute(select([expr])).scalar() == "somedatasomedata"
# test custom operator conversion
expr = test_table.c.value + 40
assert expr.type.__class__ is test_table.c.value.type.__class__
-
+
# + operator converted to -
# value is calculated as: (250 - (40 * 10)) / 10 == -15
assert testbase.db.execute(select([expr.label('foo')])).scalar() == -15
# this one relies upon anonymous labeling to assemble result
# processing rules on the column.
assert testbase.db.execute(select([expr])).scalar() == -15
-
+
class DateTest(AssertMixin):
def setUpAll(self):
global users_with_date, insert_data
self.assert_(x.adatetime.__class__ == datetime.datetime)
t.delete().execute()
-
+
# test mismatched date/datetime
t.insert().execute(adate=d2, adatetime=d2)
self.assertEquals(select([t.c.adate, t.c.adatetime], t.c.adate==d1).execute().fetchall(), [(d1, d2)])
self.assertEquals(select([t.c.adate, t.c.adatetime], t.c.adate==d1).execute().fetchall(), [(d1, d2)])
-
+
finally:
t.drop(checkfirst=True)
class StringTest(AssertMixin):
def test_nolen_string_deprecated(self):
metadata = MetaData(testbase.db)
- foo =Table('foo', metadata,
+ foo =Table('foo', metadata,
Column('one', String))
import warnings
assert False
except SADeprecationWarning, e:
assert "Using String type with no length" in str(e)
-
+
bar = Table('bar', metadata, Column('one', String(40)))
try:
finally:
bar.drop()
warnings.filterwarnings("always", r"Using String type with no length.*")
-
-
+
+
class NumericTest(AssertMixin):
def setUpAll(self):
global numeric_table, metadata
projects / thirdparty / sqlalchemy / sqlalchemy.git / commitdiff
