mssql+mxodbc should use executedirect for all selects and execute for insert/update/delete. To support this, an is_crud property has been added to the DefaultExecutionContext. The behavior is forcable either way per execution using execution_options(native_odbc_parameters=True|False). Some tests have been added to demonstrate usage. (patch by zzzeek committed by bradallen)

1 files changed, 69 insertions, 0 deletions

diff --git a/test/dialect/test_mxodbc.py b/test/dialect/test_mxodbc.py
new file mode 100644
index 000000000..938d457fb
--- /dev/null
+++ b/test/dialect/test_mxodbc.py
@@ -0,0 +1,69 @@
+from sqlalchemy import *
+from sqlalchemy.test.testing import eq_, TestBase
+from sqlalchemy.test import engines
+
+# TODO: we should probably build mock bases for
+# these to share with test_reconnect, test_parseconnect
+class MockDBAPI(object):
+    paramstyle = 'qmark'
+    def __init__(self):
+        self.log = []
+    def connect(self, *args, **kwargs):
+        return MockConnection(self)
+    
+class MockConnection(object):
+    def __init__(self, parent):
+        self.parent = parent
+    def cursor(self):
+        return MockCursor(self)
+    def close(self):
+        pass
+    def rollback(self):
+        pass
+    def commit(self):
+        pass
+
+class MockCursor(object):
+    description = None
+    rowcount = None
+    def __init__(self, parent):
+        self.parent = parent
+    def execute(self, *args, **kwargs):
+        self.parent.parent.log.append('execute')
+    def executedirect(self, *args, **kwargs):
+        self.parent.parent.log.append('executedirect')
+    def close(self):
+        pass
+
+
+class MxODBCTest(TestBase):
+    def test_native_odbc_execute(self):
+        t1 = Table('t1', MetaData(), Column('c1', Integer))
+
+        dbapi = MockDBAPI()
+        engine = engines.testing_engine(
+                                'mssql+mxodbc://localhost', 
+                                options={'module':dbapi, 
+                                        '_initialize':False}
+                            )
+        conn = engine.connect()
+        
+        # crud: uses execute
+        conn.execute(t1.insert().values(c1='foo'))
+        conn.execute(t1.delete().where(t1.c.c1=='foo'))
+        conn.execute(t1.update().where(t1.c.c1=='foo').values(c1='bar'))
+        
+        # select: uses executedirect
+        conn.execute(t1.select())
+        
+        # manual flagging
+        conn.execution_options(native_odbc_execute=True).execute(t1.select())
+        conn.execution_options(native_odbc_execute=False).execute(t1.insert().values(c1='foo'))
+        
+        eq_(
+            dbapi.log,
+            ['execute', 'execute', 'execute', 
+                'executedirect', 'execute', 'executedirect']
+        )
+        
+        
\ No newline at end of file