Collation Support
-----------------
-MSSQL specific string types support a collation parameter that
-creates a column-level specific collation for the column. The
-collation parameter accepts a Windows Collation Name or a SQL
-Collation Name. Supported types are MSChar, MSNChar, MSString,
-MSNVarchar, MSText, and MSNText. For example::
+Character collations are supported by the base string types,
+specified by the string argument "collation"::
- from sqlalchemy.dialects.mssql import VARCHAR
+ from sqlalchemy import VARCHAR
Column('login', VARCHAR(32, collation='Latin1_General_CI_AS'))
When such a column is associated with a :class:`.Table`, the
login VARCHAR(32) COLLATE Latin1_General_CI_AS NULL
+.. versionadded:: 0.8 Character collations are now part of the base string
+ types.
+
LIMIT/OFFSET Support
--------------------
from ... import types as sqltypes
from ...types import INTEGER, BIGINT, SMALLINT, DECIMAL, NUMERIC, \
FLOAT, TIMESTAMP, DATETIME, DATE, BINARY,\
- VARBINARY
+ VARBINARY, TEXT, VARCHAR, NVARCHAR, CHAR, NCHAR
+
from ...util import update_wrapper
from . import information_schema as ischema
"""Base for MSSQL string types."""
def __init__(self, collation=None):
- self.collation = collation
-
+ super(_StringType, self).__init__(collation=collation)
-class TEXT(_StringType, sqltypes.TEXT):
- """MSSQL TEXT type, for variable-length text up to 2^31 characters."""
- def __init__(self, length=None, collation=None, **kw):
- """Construct a TEXT.
-
- :param collation: Optional, a column-level collation for this string
- value. Accepts a Windows Collation Name or a SQL Collation Name.
-
- """
- _StringType.__init__(self, collation)
- sqltypes.Text.__init__(self, length, **kw)
-class NTEXT(_StringType, sqltypes.UnicodeText):
+class NTEXT(sqltypes.UnicodeText):
"""MSSQL NTEXT type, for variable-length unicode text up to 2^30
characters."""
__visit_name__ = 'NTEXT'
- def __init__(self, length=None, collation=None, **kw):
- """Construct a NTEXT.
-
- :param collation: Optional, a column-level collation for this string
- value. Accepts a Windows Collation Name or a SQL Collation Name.
-
- """
- _StringType.__init__(self, collation)
- sqltypes.UnicodeText.__init__(self, length, **kw)
-
-
-class VARCHAR(_StringType, sqltypes.VARCHAR):
- """MSSQL VARCHAR type, for variable-length non-Unicode data with a maximum
- of 8,000 characters."""
-
- def __init__(self, length=None, collation=None, **kw):
- """Construct a VARCHAR.
-
- :param length: Optinal, maximum data length, in characters.
-
- :param convert_unicode: defaults to False. If True, convert
- ``unicode`` data sent to the database to a ``str``
- bytestring, and convert bytestrings coming back from the
- database into ``unicode``.
-
- Bytestrings are encoded using the dialect's
- :attr:`~sqlalchemy.engine.Dialect.encoding`, which
- defaults to `utf-8`.
-
- If False, may be overridden by
- :attr:`sqlalchemy.engine.Dialect.convert_unicode`.
-
- :param collation: Optional, a column-level collation for this string
- value. Accepts a Windows Collation Name or a SQL Collation Name.
-
- """
- _StringType.__init__(self, collation)
- sqltypes.VARCHAR.__init__(self, length, **kw)
-
-
-class NVARCHAR(_StringType, sqltypes.NVARCHAR):
- """MSSQL NVARCHAR type.
-
- For variable-length unicode character data up to 4,000 characters."""
-
- def __init__(self, length=None, collation=None, **kw):
- """Construct a NVARCHAR.
-
- :param length: Optional, Maximum data length, in characters.
-
- :param collation: Optional, a column-level collation for this string
- value. Accepts a Windows Collation Name or a SQL Collation Name.
-
- """
- _StringType.__init__(self, collation)
- sqltypes.NVARCHAR.__init__(self, length, **kw)
-
-
-class CHAR(_StringType, sqltypes.CHAR):
- """MSSQL CHAR type, for fixed-length non-Unicode data with a maximum
- of 8,000 characters."""
-
- def __init__(self, length=None, collation=None, **kw):
- """Construct a CHAR.
-
- :param length: Optinal, maximum data length, in characters.
-
- :param convert_unicode: defaults to False. If True, convert
- ``unicode`` data sent to the database to a ``str``
- bytestring, and convert bytestrings coming back from the
- database into ``unicode``.
-
- Bytestrings are encoded using the dialect's
- :attr:`~sqlalchemy.engine.Dialect.encoding`, which
- defaults to `utf-8`.
-
- If False, may be overridden by
- :attr:`sqlalchemy.engine.Dialect.convert_unicode`.
-
- :param collation: Optional, a column-level collation for this string
- value. Accepts a Windows Collation Name or a SQL Collation Name.
-
- """
- _StringType.__init__(self, collation)
- sqltypes.CHAR.__init__(self, length, **kw)
-
-
-class NCHAR(_StringType, sqltypes.NCHAR):
- """MSSQL NCHAR type.
-
- For fixed-length unicode character data up to 4,000 characters."""
-
- def __init__(self, length=None, collation=None, **kw):
- """Construct an NCHAR.
-
- :param length: Optional, Maximum data length, in characters.
-
- :param collation: Optional, a column-level collation for this string
- value. Accepts a Windows Collation Name or a SQL Collation Name.
-
- """
- _StringType.__init__(self, collation)
- sqltypes.NCHAR.__init__(self, length, **kw)
class IMAGE(sqltypes.LargeBinary):
# -*- encoding: utf-8
-from sqlalchemy.testing import eq_
+from sqlalchemy.testing import eq_, engines, pickleable
import datetime
import os
-import re
-import warnings
from sqlalchemy import *
from sqlalchemy import types, exc, schema, event
from sqlalchemy.orm import *
self.assert_compile(t.select().with_hint(t, 'WITH (NOLOCK)'),
'SELECT sometable.somecolumn FROM sometable WITH (NOLOCK)')
- def test_join_with_hint (self):
+ def test_join_with_hint(self):
t1 = table('t1',
column('a', Integer),
column('b', String),
class ReflectionTest(fixtures.TestBase, ComparesTables):
__only_on__ = 'mssql'
+ @testing.provide_metadata
def test_basic_reflection(self):
- meta = MetaData(testing.db)
+ meta = self.metadata
users = Table(
'engine_users',
server_default='5'),
Column('test9', types.BINARY(100)),
Column('test_numeric', types.Numeric()),
- test_needs_fk=True,
)
addresses = Table(
Column('remote_user_id', types.Integer,
ForeignKey(users.c.user_id)),
Column('email_address', types.String(20)),
- test_needs_fk=True,
)
meta.create_all()
- try:
- meta2 = MetaData()
- reflected_users = Table('engine_users', meta2,
- autoload=True,
- autoload_with=testing.db)
- reflected_addresses = Table('engine_email_addresses',
- meta2, autoload=True, autoload_with=testing.db)
- self.assert_tables_equal(users, reflected_users)
- self.assert_tables_equal(addresses, reflected_addresses)
- finally:
- meta.drop_all()
+ meta2 = MetaData()
+ reflected_users = Table('engine_users', meta2,
+ autoload=True,
+ autoload_with=testing.db)
+ reflected_addresses = Table('engine_email_addresses',
+ meta2, autoload=True, autoload_with=testing.db)
+ self.assert_tables_equal(users, reflected_users)
+ self.assert_tables_equal(addresses, reflected_addresses)
@testing.provide_metadata
def test_identity(self):
result_processor = mssql_time_type.result_processor(None, None)
eq_(expected, result_processor(value))
-class TypesTest(fixtures.TestBase, AssertsExecutionResults, ComparesTables):
+
+class TypeDDLTest(fixtures.TestBase):
+ def test_boolean(self):
+ "Exercise type specification for boolean type."
+
+ columns = [
+ # column type, args, kwargs, expected ddl
+ (Boolean, [], {},
+ 'BIT'),
+ ]
+
+ metadata = MetaData()
+ table_args = ['test_mssql_boolean', metadata]
+ for index, spec in enumerate(columns):
+ type_, args, kw, res = spec
+ table_args.append(
+ Column('c%s' % index, type_(*args, **kw), nullable=None))
+
+ boolean_table = Table(*table_args)
+ dialect = mssql.dialect()
+ gen = dialect.ddl_compiler(dialect, schema.CreateTable(boolean_table))
+
+ for col in boolean_table.c:
+ index = int(col.name[1:])
+ testing.eq_(gen.get_column_specification(col),
+ "%s %s" % (col.name, columns[index][3]))
+ self.assert_(repr(col))
+
+
+ def test_numeric(self):
+ "Exercise type specification and options for numeric types."
+
+ columns = [
+ # column type, args, kwargs, expected ddl
+ (types.NUMERIC, [], {},
+ 'NUMERIC'),
+ (types.NUMERIC, [None], {},
+ 'NUMERIC'),
+ (types.NUMERIC, [12, 4], {},
+ 'NUMERIC(12, 4)'),
+
+ (types.Float, [], {},
+ 'FLOAT'),
+ (types.Float, [None], {},
+ 'FLOAT'),
+ (types.Float, [12], {},
+ 'FLOAT(12)'),
+ (mssql.MSReal, [], {},
+ 'REAL'),
+
+ (types.Integer, [], {},
+ 'INTEGER'),
+ (types.BigInteger, [], {},
+ 'BIGINT'),
+ (mssql.MSTinyInteger, [], {},
+ 'TINYINT'),
+ (types.SmallInteger, [], {},
+ 'SMALLINT'),
+ ]
+
+ metadata = MetaData()
+ table_args = ['test_mssql_numeric', metadata]
+ for index, spec in enumerate(columns):
+ type_, args, kw, res = spec
+ table_args.append(
+ Column('c%s' % index, type_(*args, **kw), nullable=None))
+
+ numeric_table = Table(*table_args)
+ dialect = mssql.dialect()
+ gen = dialect.ddl_compiler(dialect, schema.CreateTable(numeric_table))
+
+ for col in numeric_table.c:
+ index = int(col.name[1:])
+ testing.eq_(gen.get_column_specification(col),
+ "%s %s" % (col.name, columns[index][3]))
+ self.assert_(repr(col))
+
+
+ def test_char(self):
+ """Exercise COLLATE-ish options on string types."""
+
+ columns = [
+ (mssql.MSChar, [], {},
+ 'CHAR'),
+ (mssql.MSChar, [1], {},
+ 'CHAR(1)'),
+ (mssql.MSChar, [1], {'collation': 'Latin1_General_CI_AS'},
+ 'CHAR(1) COLLATE Latin1_General_CI_AS'),
+
+ (mssql.MSNChar, [], {},
+ 'NCHAR'),
+ (mssql.MSNChar, [1], {},
+ 'NCHAR(1)'),
+ (mssql.MSNChar, [1], {'collation': 'Latin1_General_CI_AS'},
+ 'NCHAR(1) COLLATE Latin1_General_CI_AS'),
+
+ (mssql.MSString, [], {},
+ 'VARCHAR(max)'),
+ (mssql.MSString, [1], {},
+ 'VARCHAR(1)'),
+ (mssql.MSString, [1], {'collation': 'Latin1_General_CI_AS'},
+ 'VARCHAR(1) COLLATE Latin1_General_CI_AS'),
+
+ (mssql.MSNVarchar, [], {},
+ 'NVARCHAR(max)'),
+ (mssql.MSNVarchar, [1], {},
+ 'NVARCHAR(1)'),
+ (mssql.MSNVarchar, [1], {'collation': 'Latin1_General_CI_AS'},
+ 'NVARCHAR(1) COLLATE Latin1_General_CI_AS'),
+
+ (mssql.MSText, [], {},
+ 'TEXT'),
+ (mssql.MSText, [], {'collation': 'Latin1_General_CI_AS'},
+ 'TEXT COLLATE Latin1_General_CI_AS'),
+
+ (mssql.MSNText, [], {},
+ 'NTEXT'),
+ (mssql.MSNText, [], {'collation': 'Latin1_General_CI_AS'},
+ 'NTEXT COLLATE Latin1_General_CI_AS'),
+ ]
+
+ metadata = MetaData()
+ table_args = ['test_mssql_charset', metadata]
+ for index, spec in enumerate(columns):
+ type_, args, kw, res = spec
+ table_args.append(
+ Column('c%s' % index, type_(*args, **kw), nullable=None))
+
+ charset_table = Table(*table_args)
+ dialect = mssql.dialect()
+ gen = dialect.ddl_compiler(dialect, schema.CreateTable(charset_table))
+
+ for col in charset_table.c:
+ index = int(col.name[1:])
+ testing.eq_(gen.get_column_specification(col),
+ "%s %s" % (col.name, columns[index][3]))
+ self.assert_(repr(col))
+
+
+ def test_timestamp(self):
+ """Exercise TIMESTAMP column."""
+
+ dialect = mssql.dialect()
+
+ metadata = MetaData()
+ spec, expected = (TIMESTAMP, 'TIMESTAMP')
+ t = Table('mssql_ts', metadata,
+ Column('id', Integer, primary_key=True),
+ Column('t', spec, nullable=None))
+ gen = dialect.ddl_compiler(dialect, schema.CreateTable(t))
+ testing.eq_(gen.get_column_specification(t.c.t), "t %s" % expected)
+ self.assert_(repr(t.c.t))
+
+ def test_money(self):
+ """Exercise type specification for money types."""
+
+ columns = [(mssql.MSMoney, [], {}, 'MONEY'),
+ (mssql.MSSmallMoney, [], {}, 'SMALLMONEY')]
+ metadata = MetaData()
+ table_args = ['test_mssql_money', metadata]
+ for index, spec in enumerate(columns):
+ type_, args, kw, res = spec
+ table_args.append(Column('c%s' % index, type_(*args, **kw),
+ nullable=None))
+ money_table = Table(*table_args)
+ dialect = mssql.dialect()
+ gen = dialect.ddl_compiler(dialect,
+ schema.CreateTable(money_table))
+ for col in money_table.c:
+ index = int(col.name[1:])
+ testing.eq_(gen.get_column_specification(col), '%s %s'
+ % (col.name, columns[index][3]))
+ self.assert_(repr(col))
+
+ def test_binary(self):
+ "Exercise type specification for binary types."
+
+ columns = [
+ # column type, args, kwargs, expected ddl
+ (mssql.MSBinary, [], {},
+ 'BINARY'),
+ (mssql.MSBinary, [10], {},
+ 'BINARY(10)'),
+
+ (types.BINARY, [], {},
+ 'BINARY'),
+ (types.BINARY, [10], {},
+ 'BINARY(10)'),
+
+ (mssql.MSVarBinary, [], {},
+ 'VARBINARY(max)'),
+ (mssql.MSVarBinary, [10], {},
+ 'VARBINARY(10)'),
+
+ (types.VARBINARY, [10], {},
+ 'VARBINARY(10)'),
+ (types.VARBINARY, [], {},
+ 'VARBINARY(max)'),
+
+ (mssql.MSImage, [], {},
+ 'IMAGE'),
+
+ (mssql.IMAGE, [], {},
+ 'IMAGE'),
+
+ (types.LargeBinary, [], {},
+ 'IMAGE'),
+ ]
+
+ metadata = MetaData()
+ table_args = ['test_mssql_binary', metadata]
+ for index, spec in enumerate(columns):
+ type_, args, kw, res = spec
+ table_args.append(Column('c%s' % index, type_(*args, **kw),
+ nullable=None))
+ binary_table = Table(*table_args)
+ dialect = mssql.dialect()
+ gen = dialect.ddl_compiler(dialect,
+ schema.CreateTable(binary_table))
+ for col in binary_table.c:
+ index = int(col.name[1:])
+ testing.eq_(gen.get_column_specification(col), '%s %s'
+ % (col.name, columns[index][3]))
+ self.assert_(repr(col))
+
+class TypeRoundTripTest(fixtures.TestBase, AssertsExecutionResults, ComparesTables):
__only_on__ = 'mssql'
@classmethod
except Exception, e:
raise e
- def test_money(self):
- """Exercise type specification for money types."""
-
- columns = [(mssql.MSMoney, [], {}, 'MONEY'),
- (mssql.MSSmallMoney, [], {}, 'SMALLMONEY')]
- table_args = ['test_mssql_money', metadata]
- for index, spec in enumerate(columns):
- type_, args, kw, res = spec
- table_args.append(Column('c%s' % index, type_(*args, **kw),
- nullable=None))
- money_table = Table(*table_args)
- dialect = mssql.dialect()
- gen = dialect.ddl_compiler(dialect,
- schema.CreateTable(money_table))
- for col in money_table.c:
- index = int(col.name[1:])
- testing.eq_(gen.get_column_specification(col), '%s %s'
- % (col.name, columns[index][3]))
- self.assert_(repr(col))
- try:
- money_table.create(checkfirst=True)
- assert True
- except:
- raise
- money_table.drop()
# todo this should suppress warnings, but it does not
@emits_warning_on('mssql+mxodbc', r'.*does not have any indexes.*')
== d1).execute().fetchall(), [(d1, t1, d2)])
@emits_warning_on('mssql+mxodbc', r'.*does not have any indexes.*')
- def test_binary(self):
+ @testing.provide_metadata
+ def test_binary_reflection(self):
"Exercise type specification for binary types."
columns = [
'IMAGE'),
]
+ metadata = self.metadata
table_args = ['test_mssql_binary', metadata]
for index, spec in enumerate(columns):
type_, args, kw, res = spec
table_args.append(Column('c%s' % index, type_(*args, **kw),
nullable=None))
binary_table = Table(*table_args)
- dialect = mssql.dialect()
- gen = dialect.ddl_compiler(dialect,
- schema.CreateTable(binary_table))
- for col in binary_table.c:
- index = int(col.name[1:])
- testing.eq_(gen.get_column_specification(col), '%s %s'
- % (col.name, columns[index][3]))
- self.assert_(repr(col))
metadata.create_all()
reflected_binary = Table('test_mssql_binary',
MetaData(testing.db), autoload=True)
testing.eq_(col.type.length,
binary_table.c[col.name].type.length)
- def test_boolean(self):
- "Exercise type specification for boolean type."
-
- columns = [
- # column type, args, kwargs, expected ddl
- (Boolean, [], {},
- 'BIT'),
- ]
-
- table_args = ['test_mssql_boolean', metadata]
- for index, spec in enumerate(columns):
- type_, args, kw, res = spec
- table_args.append(
- Column('c%s' % index, type_(*args, **kw), nullable=None))
-
- boolean_table = Table(*table_args)
- dialect = mssql.dialect()
- gen = dialect.ddl_compiler(dialect, schema.CreateTable(boolean_table))
-
- for col in boolean_table.c:
- index = int(col.name[1:])
- testing.eq_(gen.get_column_specification(col),
- "%s %s" % (col.name, columns[index][3]))
- self.assert_(repr(col))
-
- metadata.create_all()
-
- def test_numeric(self):
- "Exercise type specification and options for numeric types."
-
- columns = [
- # column type, args, kwargs, expected ddl
- (types.NUMERIC, [], {},
- 'NUMERIC'),
- (types.NUMERIC, [None], {},
- 'NUMERIC'),
- (types.NUMERIC, [12, 4], {},
- 'NUMERIC(12, 4)'),
-
- (types.Float, [], {},
- 'FLOAT'),
- (types.Float, [None], {},
- 'FLOAT'),
- (types.Float, [12], {},
- 'FLOAT(12)'),
- (mssql.MSReal, [], {},
- 'REAL'),
-
- (types.Integer, [], {},
- 'INTEGER'),
- (types.BigInteger, [], {},
- 'BIGINT'),
- (mssql.MSTinyInteger, [], {},
- 'TINYINT'),
- (types.SmallInteger, [], {},
- 'SMALLINT'),
- ]
-
- table_args = ['test_mssql_numeric', metadata]
- for index, spec in enumerate(columns):
- type_, args, kw, res = spec
- table_args.append(
- Column('c%s' % index, type_(*args, **kw), nullable=None))
-
- numeric_table = Table(*table_args)
- dialect = mssql.dialect()
- gen = dialect.ddl_compiler(dialect, schema.CreateTable(numeric_table))
-
- for col in numeric_table.c:
- index = int(col.name[1:])
- testing.eq_(gen.get_column_specification(col),
- "%s %s" % (col.name, columns[index][3]))
- self.assert_(repr(col))
-
- metadata.create_all()
-
- def test_char(self):
- """Exercise COLLATE-ish options on string types."""
-
- columns = [
- (mssql.MSChar, [], {},
- 'CHAR'),
- (mssql.MSChar, [1], {},
- 'CHAR(1)'),
- (mssql.MSChar, [1], {'collation': 'Latin1_General_CI_AS'},
- 'CHAR(1) COLLATE Latin1_General_CI_AS'),
-
- (mssql.MSNChar, [], {},
- 'NCHAR'),
- (mssql.MSNChar, [1], {},
- 'NCHAR(1)'),
- (mssql.MSNChar, [1], {'collation': 'Latin1_General_CI_AS'},
- 'NCHAR(1) COLLATE Latin1_General_CI_AS'),
-
- (mssql.MSString, [], {},
- 'VARCHAR(max)'),
- (mssql.MSString, [1], {},
- 'VARCHAR(1)'),
- (mssql.MSString, [1], {'collation': 'Latin1_General_CI_AS'},
- 'VARCHAR(1) COLLATE Latin1_General_CI_AS'),
-
- (mssql.MSNVarchar, [], {},
- 'NVARCHAR(max)'),
- (mssql.MSNVarchar, [1], {},
- 'NVARCHAR(1)'),
- (mssql.MSNVarchar, [1], {'collation': 'Latin1_General_CI_AS'},
- 'NVARCHAR(1) COLLATE Latin1_General_CI_AS'),
-
- (mssql.MSText, [], {},
- 'TEXT'),
- (mssql.MSText, [], {'collation': 'Latin1_General_CI_AS'},
- 'TEXT COLLATE Latin1_General_CI_AS'),
-
- (mssql.MSNText, [], {},
- 'NTEXT'),
- (mssql.MSNText, [], {'collation': 'Latin1_General_CI_AS'},
- 'NTEXT COLLATE Latin1_General_CI_AS'),
- ]
-
- table_args = ['test_mssql_charset', metadata]
- for index, spec in enumerate(columns):
- type_, args, kw, res = spec
- table_args.append(
- Column('c%s' % index, type_(*args, **kw), nullable=None))
-
- charset_table = Table(*table_args)
- dialect = mssql.dialect()
- gen = dialect.ddl_compiler(dialect, schema.CreateTable(charset_table))
-
- for col in charset_table.c:
- index = int(col.name[1:])
- testing.eq_(gen.get_column_specification(col),
- "%s %s" % (col.name, columns[index][3]))
- self.assert_(repr(col))
-
- metadata.create_all()
-
- def test_timestamp(self):
- """Exercise TIMESTAMP column."""
-
- dialect = mssql.dialect()
-
- spec, expected = (TIMESTAMP,'TIMESTAMP')
- t = Table('mssql_ts', metadata,
- Column('id', Integer, primary_key=True),
- Column('t', spec, nullable=None))
- gen = dialect.ddl_compiler(dialect, schema.CreateTable(t))
- testing.eq_(gen.get_column_specification(t.c.t), "t %s" % expected)
- self.assert_(repr(t.c.t))
- t.create(checkfirst=True)
def test_autoincrement(self):
Table('ai_1', metadata,
projects / thirdparty / sqlalchemy / sqlalchemy.git / commitdiff
