- mssql
- Re-established initial support for pymssql.
+ - Various fixes for implicit returning, reflection,
+ etc. - the MS-SQL dialects aren't quite complete
+ in 0.6 yet (but are close)
+
- Added basic support for mxODBC [ticket:1710].
- Removed the text_as_varchar option.
import sys
+import re
+
from sqlalchemy.connectors import Connector
class MxODBCConnector(Connector):
supports_sane_multi_rowcount = False
supports_unicode_statements = False
supports_unicode_binds = False
-
+ supports_native_decimal = False
+
@classmethod
def dbapi(cls):
platform = sys.platform
else:
return False
-
+ def _get_server_version_info(self, connection):
+ dbapi_con = connection.connection
+ version = []
+ r = re.compile('[.\-]')
+ # 18 == pyodbc.SQL_DBMS_VER
+ for n in r.split(dbapi_con.getinfo(18)[1]):
+ try:
+ version.append(int(n))
+ except ValueError:
+ version.append(n)
+ return tuple(version)
# PyODBC unicode is broken on UCS-4 builds
supports_unicode = sys.maxunicode == 65535
supports_unicode_statements = supports_unicode
+ supports_native_decimal = True
default_paramstyle = 'named'
# for non-DSN connections, this should
class _MSNumeric(sqltypes.Numeric):
def result_processor(self, dialect, coltype):
if self.asdecimal:
- return processors.to_decimal_processor_factory(decimal.Decimal)
+ if getattr(self, 'scale', None) is not None and dialect.supports_native_decimal:
+ return None
+ else:
+ return processors.to_decimal_processor_factory(decimal.Decimal)
else:
#XXX: if the DBAPI returns a float (this is likely, given the
# processor when asdecimal is True), this should be a None
# processor instead.
return processors.to_float
-
+ return None
+
def bind_processor(self, dialect):
def process(value):
if isinstance(value, decimal.Decimal):
else:
self.cursor.execute("SELECT @@identity AS lastrowid")
# fetchall() ensures the cursor is consumed without closing it
- row = self.cursor.fetchall()[0]
+ row = self.cursor.fetchall()[0]
self._lastrowid = int(row[0])
if (self.isinsert or self.isupdate or self.isdelete) and self.compiled.returning:
self._result_proxy = base.FullyBufferedResultProxy(self)
if self._enable_identity_insert:
- self.cursor.execute("SET IDENTITY_INSERT %s OFF" % self.dialect.identifier_preparer.format_table(self.compiled.statement.table))
+ self.cursor.execute(
+ "SET IDENTITY_INSERT %s OFF" %
+ self.dialect.identifier_preparer.
+ format_table(self.compiled.statement.table)
+ )
def get_lastrowid(self):
return self._lastrowid
query_timeout=None,
use_scope_identity=True,
max_identifier_length=None,
- schema_name="dbo", **opts):
+ schema_name=u"dbo", **opts):
self.query_timeout = int(query_timeout or 0)
self.schema_name = schema_name
def initialize(self, connection):
super(MSDialect, self).initialize(connection)
- if self.server_version_info >= MS_2005_VERSION and 'implicit_returning' not in self.__dict__:
+ if self.server_version_info >= MS_2005_VERSION and \
+ 'implicit_returning' not in self.__dict__:
self.implicit_returning = True
def _get_default_schema_name(self, connection):
try:
default_schema_name = connection.scalar(query, [user_name])
if default_schema_name is not None:
- return default_schema_name
+ return unicode(default_schema_name)
except:
pass
return self.schema_name
name='%s_identity' % col_name)
break
cursor.close()
- if ic is not None:
- try:
- # is this table_fullname reliable?
- table_fullname = "%s.%s" % (current_schema, tablename)
- cursor = connection.execute(
- sql.text("select ident_seed(:seed), ident_incr(:incr)"),
- {'seed':table_fullname, 'incr':table_fullname}
+
+ if ic is not None and self.server_version_info >= MS_2005_VERSION:
+ table_fullname = "%s.%s" % (current_schema, tablename)
+ cursor = connection.execute(
+ sql.text("select ident_seed(:tname), ident_incr(:tname)",
+ bindparams=[
+ sql.bindparam('tname', table_fullname)
+ ]
)
- row = cursor.first()
- if not row is None:
- colmap[ic]['sequence'].update({
- 'start' : int(row[0]),
- 'increment' : int(row[1])
- })
- except:
- # ignoring it, works just like before
- pass
+ )
+ row = cursor.first()
+ if not row is None:
+ colmap[ic]['sequence'].update({
+ 'start' : int(row[0]),
+ 'increment' : int(row[1])
+ })
return cols
@reflection.cache
def get_primary_keys(self, connection, tablename, schema=None, **kw):
current_schema = schema or self.default_schema_name
pkeys = []
- # Add constraints
- RR = ischema.ref_constraints #information_schema.referential_constraints
- TC = ischema.constraints #information_schema.table_constraints
- C = ischema.key_constraints.alias('C') #information_schema.constraint_column_usage: the constrained column
- R = ischema.key_constraints.alias('R') #information_schema.constraint_column_usage: the referenced column
+ RR = ischema.ref_constraints # information_schema.referential_constraints
+ TC = ischema.constraints # information_schema.table_constraints
+ C = ischema.key_constraints.alias('C') # information_schema.constraint_column_usage:
+ # the constrained column
+ R = ischema.key_constraints.alias('R') # information_schema.constraint_column_usage:
+ # the referenced column
# Primary key constraints
s = sql.select([C.c.column_name, TC.c.constraint_type],
# Add constraints
RR = ischema.ref_constraints #information_schema.referential_constraints
TC = ischema.constraints #information_schema.table_constraints
- C = ischema.key_constraints.alias('C') #information_schema.constraint_column_usage: the constrained column
- R = ischema.key_constraints.alias('R') #information_schema.constraint_column_usage: the referenced column
+ C = ischema.key_constraints.alias('C') # information_schema.constraint_column_usage:
+ # the constrained column
+ R = ischema.key_constraints.alias('R') # information_schema.constraint_column_usage:
+ # the referenced column
# Foreign key constraints
s = sql.select([C.c.column_name,
R.c.table_schema, R.c.table_name, R.c.column_name,
- RR.c.constraint_name, RR.c.match_option, RR.c.update_rule, RR.c.delete_rule],
+ RR.c.constraint_name, RR.c.match_option, RR.c.update_rule,
+ RR.c.delete_rule],
sql.and_(C.c.table_name == tablename,
C.c.table_schema == current_schema,
C.c.constraint_name == RR.c.constraint_name,
return fkeys.values()
-
-# fixme. I added this for the tests to run. -Randall
-MSSQLDialect = MSDialect
MSExecutionContext_mxodbc = MSExecutionContext_pyodbc
class MSDialect_mxodbc(MxODBCConnector, MSDialect):
- supports_sane_rowcount = True
+ # FIXME: yikes, plain rowcount doesn't work ?
+ supports_sane_rowcount = False #True
supports_sane_multi_rowcount = False
execution_ctx_cls = MSExecutionContext_mxodbc
# We may have to skip over a number of result sets with no data (due to triggers, etc.)
while True:
try:
- # fetchall() ensures the cursor is consumed without closing it (FreeTDS particularly)
+ # fetchall() ensures the cursor is consumed
+ # without closing it (FreeTDS particularly)
row = self.cursor.fetchall()[0]
break
except self.dialect.dbapi.Error, e:
supports_native_enum = False
supports_native_boolean = False
+ # if the NUMERIC type
+ # returns decimal.Decimal.
+ # *not* the FLOAT type however.
+ supports_native_decimal = False
+
# Py3K
#supports_unicode_statements = True
#supports_unicode_binds = True
if c.primary_key and \
need_pks and \
(
+ implicit_returning or
not postfetch_lastrowid or
c is not stmt.table._autoincrement_column
):
exclude('mysql', '<', (4, 1, 1), 'no unicode connection support'),
)
+def sane_rowcount(fn):
+ return _chain_decorators_on(
+ fn,
+ skip_if(lambda: not testing.db.dialect.supports_sane_rowcount)
+ )
+
def python2(fn):
return _chain_decorators_on(
fn,
class IdentityInsertTest(TestBase, AssertsCompiledSQL):
__only_on__ = 'mssql'
- __dialect__ = mssql.MSSQLDialect()
+ __dialect__ = mssql.MSDialect()
@classmethod
def setup_class(cls):
meta2 = MetaData(testing.db)
try:
table2 = Table('identity_test', meta2, autoload=True)
- sequence = isinstance(table2.c['col1'].default, schema.Sequence) and table2.c['col1'].default
+ sequence = isinstance(table2.c['col1'].default, schema.Sequence) \
+ and table2.c['col1'].default
assert sequence.start == 2
assert sequence.increment == 3
finally:
'0.0000000000000000002', '0.2', '-0.0000000000000000002', '-2E-2',
'156666.458923543', '-156666.458923543', '1', '-1', '-1234', '1234',
'2E-12', '4E8', '3E-6', '3E-7', '4.1', '1E-1', '1E-2', '1E-3',
- '1E-4', '1E-5', '1E-6', '1E-7', '1E-1', '1E-8', '0.2732E2', '-0.2432E2', '4.35656E2',
+ '1E-4', '1E-5', '1E-6', '1E-7', '1E-1', '1E-8', '0.2732E2',
+ '-0.2432E2', '4.35656E2',
'-02452E-2', '45125E-2',
'1234.58965E-2', '1.521E+15', '-1E-25', '1E-25', '1254E-25', '-1203E-25',
'0', '-0.00', '-0', '4585E12', '000000000000000000012', '000000000000.32E12',
numeric_table.insert().execute(numericcol=value)
for value in select([numeric_table.c.numericcol]).execute():
- assert value[0] in test_items, "%s not in test_items" % value[0]
+ assert value[0] in test_items, "%r not in test_items" % value[0]
def test_float(self):
float_table = Table('float_table', metadata,
testing.eq_(gen.get_column_specification(t.c.t), "t %s" % expected)
self.assert_(repr(t.c.t))
t.create(checkfirst=True)
-
+
+ @testing.crashes("+mxodbc", "mxODBC doesn't do scope_identity() with DEFAULT VALUES")
def test_autoincrement(self):
Table('ai_1', metadata,
Column('int_y', Integer, primary_key=True),
Column('int_n', Integer, DefaultClause('0'),
- primary_key=True))
+ primary_key=True, autoincrement=False))
Table('ai_2', metadata,
Column('int_y', Integer, primary_key=True),
Column('int_n', Integer, DefaultClause('0'),
- primary_key=True))
+ primary_key=True, autoincrement=False))
Table('ai_3', metadata,
Column('int_n', Integer, DefaultClause('0'),
primary_key=True, autoincrement=False),
for name in table_names:
tbl = Table(name, mr, autoload=True)
+ tbl = metadata.tables[name]
for c in tbl.c:
if c.name.startswith('int_y'):
- assert c.autoincrement
+ assert c.autoincrement, name
+ assert tbl._autoincrement_column is c, name
elif c.name.startswith('int_n'):
- assert not c.autoincrement
+ assert not c.autoincrement, name
+ assert tbl._autoincrement_column is not c, name
for counter, engine in enumerate([
engines.testing_engine(options={'implicit_returning':False}),
metadata = MetaData(bind)
if testing.against('sybase', 'maxdb', 'oracle', 'mssql'):
- names = set(['plain'])
+ names = set([u'plain'])
else:
names = set([u'plain', u'Unit\u00e9ble', u'\u6e2c\u8a66'])
for name in names:
- Table(name, metadata, Column('id', sa.Integer, sa.Sequence(name + "_id_seq"), primary_key=True))
+ Table(name, metadata, Column('id', sa.Integer, sa.Sequence(name + "_id_seq"),
+ primary_key=True))
metadata.create_all()
reflected = set(bind.table_names())
detects rows that had defaults and post-fetches.
"""
+ # verify implicit_returning is working
+ if engine.dialect.implicit_returning:
+ ins = table.insert()
+ comp = ins.compile(engine, column_keys=list(values))
+ if not set(values).issuperset(c.key for c in table.primary_key):
+ assert comp.returning
+
result = engine.execute(table.insert(), **values)
ret = values.copy()
ret[col.key] = id
if result.lastrow_has_defaults():
- criterion = and_(*[col==id for col, id in zip(table.primary_key, result.inserted_primary_key)])
+ criterion = and_(*[col==id for col, id in
+ zip(table.primary_key, result.inserted_primary_key)])
row = engine.execute(table.select(criterion)).first()
for c in table.c:
ret[c.key] = row[c]
return ret
if testing.against('firebird', 'postgresql', 'oracle', 'mssql'):
+ assert testing.db.dialect.implicit_returning
+
+ if testing.db.dialect.implicit_returning:
test_engines = [
engines.testing_engine(options={'implicit_returning':False}),
engines.testing_engine(options={'implicit_returning':True}),
@testing.fails_on('postgresql', '')
@testing.fails_on('oracle', '')
+ @testing.crashes('mssql+mxodbc', 'Raises an error')
def test_executemany():
# return value is documented as failing with psycopg2/executemany
result2 = table.insert().returning(table).execute(
test_executemany()
- result3 = table.insert().returning(table.c.id).execute({'persons': 4, 'full': False})
- eq_([dict(row) for row in result3], [{'id': 4}])
@testing.exclude('firebird', '<', (2, 1), '2.1+ feature')
class FoundRowsTest(TestBase, AssertsExecutionResults):
"""tests rowcount functionality"""
+
+ __requires__ = ('sane_rowcount', )
+
@classmethod
def setup_class(cls):
metadata = MetaData(testing.db)
global employees_table
employees_table = Table('employees', metadata,
- Column('employee_id', Integer, Sequence('employee_id_seq', optional=True), primary_key=True),
+ Column('employee_id', Integer,
+ Sequence('employee_id_seq', optional=True),
+ primary_key=True),
Column('name', String(50)),
Column('department', String(1)),
)
class UserDefinedTest(TestBase):
"""tests user-defined types."""
- def testprocessing(self):
+ def test_processing(self):
global users
users.insert().execute(
[1800, 2250, 1350],
l
):
- for col in row[1:5]:
+ for col in list(row)[1:5]:
eq_(col, assertstr)
eq_(row[5], assertint)
eq_(row[6], assertint2)
global bool_table
metadata = MetaData(testing.db)
bool_table = Table('booltest', metadata,
- Column('id', Integer, primary_key=True),
+ Column('id', Integer, primary_key=True, autoincrement=False),
Column('value', Boolean),
Column('unconstrained_value', Boolean(create_constraint=False)),
)
@testing.fails_on('mysql',
"The CHECK clause is parsed but ignored by all storage engines.")
+ @testing.fails_on('mssql',
+ "FIXME: MS-SQL 2005 doesn't honor CHECK ?!?")
@testing.skip_if(lambda: testing.db.dialect.supports_native_boolean)
def test_constraint(self):
assert_raises((exc.IntegrityError, exc.ProgrammingError),
projects / thirdparty / sqlalchemy / sqlalchemy.git / commitdiff
