def connect(conn):
conn.stringformat = self.dbapi.MIXED_STRINGFORMAT
conn.datetimeformat = self.dbapi.PYDATETIME_DATETIMEFORMAT
- conn.errorhandler = error_handler
+ conn.decimalformat = self.dbapi.DECIMAL_DECIMALFORMAT
+ conn.errorhandler = self._error_handler()
# Alternatives to experiment with:
#conn.bindmethod = self.dbapi.BIND_USING_PYTHONTYPE
- #conn.bindmethod = self.dbapi.BIND_USING_SQLTYPE
+ conn.bindmethod = self.dbapi.BIND_USING_SQLTYPE
return connect
- """Return a handler that adjusts mxODBC's raised Warnings to emit Python standard warnings.
+
+ def _error_handler(self):
- #import pdb; pdb.set_trace()
++ """Return a handler that adjusts mxODBC's raised Warnings to
++ emit Python standard warnings.
+ """
+
+ from mx.ODBC.Error import Warning as MxOdbcWarning
+ def error_handler(connection, cursor, errorclass, errorvalue):
+
+ if issubclass(errorclass, MxOdbcWarning):
+ errorclass.__bases__ = (Warning,)
+ warnings.warn(message=str(errorvalue),
+ category=errorclass,
+ stacklevel=2)
+ else:
+ raise errorclass, errorvalue
+ return error_handler
def create_connect_args(self, url):
""" Return a tuple of *args,**kwargs for creating a connection.
except ValueError:
version.append(n)
return tuple(version)
-
-
-
- def error_handler(connection, cursor, errorclass, errorvalue):
- """
- Adjust mxODBC's raised Warnings to emit Python standard warnings.
- """
- if issubclass(errorclass, MxOdbcWarning):
- errorclass.__bases__ = (Warning,)
- warnings.warn(message=str(errorvalue),
- category=errorclass,
- stacklevel=2)
- else:
- raise errorclass, errorvalue
-
-
+
+class MxNumeric(sqltypes.Numeric):
+ """
+ Handle Numeric types between SQLAlchemy and mxODBC.
+ """
+ def bind_processor(self, dialect):
+ """
+ SQLAlchemy can accept a Python Decimal for bind
+ variables, so no special bind_processor is needed.
+ """
+ return None
+
+ def result_processor(self, dialect, coltype):
+ """
+ For cases when a
+ """
+ if self.asdecimal:
+ return None
+ else:
+ return processors.to_float
+
+
+class MxFloat(sqltypes.Float):
+ """
+ Handle Numeric types between SQLAlchemy and mxODBC.
+ """
+ def bind_processor(self, dialect):
+ """
+ SQLAlchemy can accept a Python Decimal for bind
+ variables, so no special bind_processor is needed.
+ """
+ return None
+
+ def result_processor(self, dialect, coltype):
+ """
+ mxODBC returns Python float values for REAL, FLOAT, and
+ DOUBLE column types.
+ """
+ if self.asdecimal:
+ return processors.to_decimal_processor_factory(Decimal)
+ else:
+ return None
+
+
import sys
from sqlalchemy import types as sqltypes
-from sqlalchemy.connectors.mxodbc import MxODBCConnector
+from sqlalchemy import util
+from sqlalchemy.connectors.mxodbc import MxODBCConnector, MxNumeric, MxFloat
from sqlalchemy.dialects.mssql.pyodbc import MSExecutionContext_pyodbc
- from sqlalchemy.dialects.mssql.base import MSExecutionContext, MSDialect
-
+ from sqlalchemy.dialects.mssql.base import MSExecutionContext, MSDialect, \
+ MSSQLCompiler, MSSQLStrictCompiler
+
class MSExecutionContext_mxodbc(MSExecutionContext_pyodbc):
"""
The pyodbc execution context is useful for enabling
# won't work.
class MSDialect_mxodbc(MxODBCConnector, MSDialect):
-
- execution_ctx_cls = MSExecutionContext_mxodbc
+
+ # TODO: may want to use this only if FreeTDS is not in use,
+ # since FreeTDS doesn't seem to use native binds.
+ statement_compiler = MSSQLStrictCompiler
+
+ execution_ctx_cls = MSExecutionContext_mxodbc
+ colspecs = util.update_copy(
+ MSDialect.colspecs,
+ {
+ sqltypes.Numeric : MxNumeric,
+ sqltypes.Float : MxFloat
+ },
+ )
+
+
+
def __init__(self, description_encoding='latin-1', **params):
super(MSDialect_mxodbc, self).__init__(**params)
self.description_encoding = description_encoding
# Null values are not outside any set
assert len(r) == 0
+ @testing.crashes('mssql+mxodbc', """Invalid bind parameter placement:
+ 'SELECT query_users.user_id, query_users.user_name \nFROM query_users
+ WHERE ? = ?' ('john', 'john')
+ """)
@testing.emits_warning('.*empty sequence.*')
- @testing.fails_on('firebird', "kinterbasdb doesn't send full type information")
+ @testing.fails_on('firebird', "uses sql-92 rules")
+ @testing.fails_on('sybase', "uses sql-92 rules")
@testing.fails_if(lambda:
testing.against('mssql+pyodbc') and not testing.db.dialect.freetds,
- "not supported by Windows ODBC driver")
+ "uses sql-92 rules")
def test_bind_in(self):
+ """test calling IN against a bind parameter.
+
+ this isn't allowed on several platforms since we
+ generate ? = ?.
+
+ """
users.insert().execute(user_id = 7, user_name = 'jack')
users.insert().execute(user_id = 8, user_name = 'fred')
users.insert().execute(user_id = 9, user_name = None)