upon close() - fixed that.
- Transaction object doesn't rollback or commit if it isn't
"active", allows more accurate nesting of begin/rollback/commit.
- Added basic support for mxODBC [ticket:1710].
- Python unicode objects as binds result in the Unicode type,
not string, thus eliminating a certain class of unicode errors
on drivers that don't support unicode binds.
- Bind parameters are sent as a tuple instead of a list. Some
backend drivers will not accept bind parameters as a list.
+
+ - threadlocal engine wasn't properly closing the connection
+ upon close() - fixed that.
+
+ - Transaction object doesn't rollback or commit if it isn't
+ "active", allows more accurate nesting of begin/rollback/commit.
+
+ - Python unicode objects as binds result in the Unicode type,
+ not string, thus eliminating a certain class of unicode errors
+ on drivers that don't support unicode binds.
- metadata
- Added the ability to strip schema information when using
- mssql
- Re-established initial support for pymssql.
+ - Added basic support for mxODBC [ticket:1710].
+
- Removed the text_as_varchar option.
- oracle
+
+import sys
from sqlalchemy.connectors import Connector
class MxODBCConnector(Connector):
supports_unicode_binds = False
@classmethod
- def import_dbapi(cls):
- import mxODBC as module
+ def dbapi(cls):
+ if 'win32' in sys.platform:
+ from mx.ODBC import Windows as module
+ elif 'linux' in sys.platform:
+ from mx.ODBC import unixODBC as module
+ elif 'darwin' in sys.platform:
+ from mx.ODBC import iODBC as module
+ else:
+ raise ImportError, "Unrecognized platform for mxODBC import"
return module
+ def visit_pool(self, pool):
+ def connect(conn, rec):
+ conn.stringformat = self.dbapi.MIXED_STRINGFORMAT
+ conn.datetimeformat = self.dbapi.PYDATETIME_DATETIMEFORMAT
+ #conn.bindmethod = self.dbapi.BIND_USING_PYTHONTYPE
+ #conn.bindmethod = self.dbapi.BIND_USING_SQLTYPE
+
+ pool.add_listener({'connect':connect})
+
def create_connect_args(self, url):
- '''Return a tuple of *args,**kwargs'''
- # FIXME: handle mx.odbc.Windows proprietary args
+ """ Return a tuple of *args,**kwargs for creating a connection.
+
+ The mxODBC 3.x connection constructor looks like this:
+
+ connect(dsn, user='', password='',
+ clear_auto_commit=1, errorhandler=None)
+
+ This method translates the values in the provided uri
+ into args and kwargs needed to instantiate an mxODBC Connection.
+
+ The arg 'errorhandler' is not used by SQLAlchemy and will
+ not be populated.
+ """
opts = url.translate_connect_args(username='user')
opts.update(url.query)
- argsDict = {}
- argsDict['user'] = opts['user']
- argsDict['password'] = opts['password']
- connArgs = [[opts['dsn']], argsDict]
- return connArgs
+ args = opts['host'],
+ kwargs = {'user':opts['user'],
+ 'password': opts['password']}
+ return args, kwargs
+
+ def is_disconnect(self, e):
+ if isinstance(e, self.dbapi.ProgrammingError):
+ return "connection already closed" in str(e)
+ elif isinstance(e, self.dbapi.Error):
+ return '[08S01]' in str(e)
+ else:
+ return False
+
+
def is_disconnect(self, e):
if isinstance(e, self.dbapi.ProgrammingError):
- return "The cursor's connection has been closed." in str(e) or 'Attempt to use a closed connection.' in str(e)
+ return "The cursor's connection has been closed." in str(e) or \
+ 'Attempt to use a closed connection.' in str(e)
elif isinstance(e, self.dbapi.Error):
return '[08S01]' in str(e)
else:
-from sqlalchemy.dialects.mssql import base, pyodbc, adodbapi, pymssql, zxjdbc
+from sqlalchemy.dialects.mssql import base, pyodbc, adodbapi, pymssql, zxjdbc, mxodbc
base.dialect = pyodbc.dialect
--- /dev/null
+from sqlalchemy.dialects.mssql.base import MSExecutionContext, MSDialect
+from sqlalchemy.connectors.mxodbc import MxODBCConnector
+from sqlalchemy import types as sqltypes
+import re
+import sys
+
+from sqlalchemy.dialects.mssql.pyodbc import MSExecutionContext_pyodbc
+
+# The pyodbc execution context seems to work for mxODBC; reuse it here
+MSExecutionContext_mxodbc = MSExecutionContext_pyodbc
+
+
+class MSDialect_mxodbc(MxODBCConnector, MSDialect):
+ supports_sane_rowcount = True
+ supports_sane_multi_rowcount = False
+
+ execution_ctx_cls = MSExecutionContext_mxodbc
+
+
+ def __init__(self, description_encoding='latin-1', **params):
+ super(MSDialect_mxodbc, self).__init__(**params)
+ self.description_encoding = description_encoding
+
+ def initialize(self, connection):
+ super(MSDialect_mxodbc, self).initialize(connection)
+ dbapi_con = connection.connection
+
+dialect = MSDialect_mxodbc
+from sqlalchemy.dialects.mssql.base import MSExecutionContext, MSDialect
+from sqlalchemy.connectors.mxodbc import MxODBCConnector
+from sqlalchemy import types as sqltypes
+import re
+import sys
+
+from sqlalchemy.dialects.mssql.pyodbc import MSExecutionContext_pyodbc
+
+# The pyodbc execution context seems to work for mxODBC; reuse it here
+MSExecutionContext_mxodbc = MSExecutionContext_pyodbc
+
+
+class MSDialect_mxodbc(MxODBCConnector, MSDialect):
+ supports_sane_rowcount = True
+ supports_sane_multi_rowcount = False
+
+ execution_ctx_cls = MSExecutionContext_mxodbc
+
+
+ def __init__(self, description_encoding='latin-1', **params):
+ super(MSDialect_mxodbc, self).__init__(**params)
+ self.description_encoding = description_encoding
+
+ def initialize(self, connection):
+ super(MSDialect_mxodbc, self).initialize(connection)
+ dbapi_con = connection.connection
+
+dialect = MSDialect_mxodbc
def rollback(self):
if not self._parent.is_active:
return
- self.is_active = False
self._do_rollback()
+ self.is_active = False
def _do_rollback(self):
self._parent.rollback()
self.connection._begin_impl()
def _do_rollback(self):
- self.connection._rollback_impl()
+ if self.is_active:
+ self.connection._rollback_impl()
def _do_commit(self):
- self.connection._commit_impl()
+ if self.is_active:
+ self.connection._commit_impl()
class NestedTransaction(Transaction):
self._savepoint = self.connection._savepoint_impl()
def _do_rollback(self):
- self.connection._rollback_to_savepoint_impl(self._savepoint, self._parent)
+ if self.is_active:
+ self.connection._rollback_to_savepoint_impl(self._savepoint, self._parent)
def _do_commit(self):
- self.connection._release_savepoint_impl(self._savepoint, self._parent)
+ if self.is_active:
+ self.connection._release_savepoint_impl(self._savepoint, self._parent)
class TwoPhaseTransaction(Transaction):
def close(self):
if not self.closed:
self.contextual_connect().close()
+ connection = self._connections.conn()
+ connection._force_close()
del self._connections.conn
self._connections.trans = []
# type which usually resolves to TEXT/CLOB
type_map = {
str: String(),
+ # Py3K
+ #bytes : LargeBinary(),
# Py2K
- unicode : String(),
+ unicode : Unicode(),
# end Py2K
int : Integer(),
float : Numeric(),
tlengine.execute(users.insert(), user_id=3, user_name='user3')
tlengine.rollback()
- # TODO: removing this line, the test still tends to pass in most
- # cases, except sporadically on PG. this should be nailed down
- # in TLEngine - removing this line should be guaranteed fail.
tlengine.rollback()
tlengine.execute(users.insert(), user_id=4, user_name='user4')
use_labels=labels,
order_by=[users.c.user_id.desc()]),
[(3,), (2,), (1,)])
+
+ @testing.fails_on("+pyodbc", "pyodbc row doesn't seem to accept slices")
+ def test_column_slices(self):
+ users.insert().execute(user_id=1, user_name='john')
+ users.insert().execute(user_id=2, user_name='jack')
+ addresses.insert().execute(address_id=1, user_id=2, address='foo@bar.com')
+ r = text("select * from query_addresses", bind=testing.db).execute().first()
+ self.assert_(r[0:1] == (1,))
+ self.assert_(r[1:] == (2, 'foo@bar.com'))
+ self.assert_(r[:-1] == (1, 2))
+
def test_column_accessor(self):
users.insert().execute(user_id=1, user_name='john')
users.insert().execute(user_id=2, user_name='jack')
self.assert_(r.user_id == r['user_id'] == r[users.c.user_id] == 2)
self.assert_(r.user_name == r['user_name'] == r[users.c.user_name] == 'jack')
- # test slices
- r = text("select * from query_addresses", bind=testing.db).execute().first()
- self.assert_(r[0:1] == (1,))
- self.assert_(r[1:] == (2, 'foo@bar.com'))
- self.assert_(r[:-1] == (1, 2))
-
- # test a little sqlite weirdness - with the UNION, cols come back as "query_users.user_id" in cursor.description
+ # test a little sqlite weirdness - with the UNION,
+ # cols come back as "query_users.user_id" in cursor.description
r = text("select query_users.user_id, query_users.user_name from query_users "
- "UNION select query_users.user_id, query_users.user_name from query_users", bind=testing.db).execute().first()
+ "UNION select query_users.user_id, query_users.user_name from query_users",
+ bind=testing.db).execute().first()
self.assert_(r['user_id']) == 1
self.assert_(r['user_name']) == "john"
projects / thirdparty / sqlalchemy / sqlalchemy.git / commitdiff
