support sql transactions

pandas/io/sql.py

@@ -62,7 +62,11 @@ def compile_big_int_sqlite(type_, compiler, **kw):
 
     if _SQLALCHEMY_INSTALLED:
         import sqlalchemy
-        return isinstance(con, sqlalchemy.engine.Connectable)
+        is_connectable = isinstance(con, sqlalchemy.engine.Connectable)
+        if sqlalchemy.__version__ >= '1.0.0':
+            # support sessions, if sqlalchemy version has them
+            is_connectable |= isinstance(con, sqlalchemy.orm.session.Session)
+        return is_connectable
     else:
         return False
 
@@ -362,7 +366,7 @@ def read_sql_query(sql, con, index_col=None, coerce_float=True, params=None,
     ----------
     sql : string
         SQL query to be executed
-    con : SQLAlchemy connectable(engine/connection) or sqlite3 DBAPI2 connection
+    con : SQLAlchemy connectable(engine/connection/session) or sqlite3 DBAPI2 connection
         Using SQLAlchemy makes it possible to use any DB supported by that
         library.
         If a DBAPI2 object, only sqlite3 is supported.
@@ -420,7 +424,7 @@ def read_sql(sql, con, index_col=None, coerce_float=True, params=None,
     ----------
     sql : string
         SQL query to be executed or database table name.
-    con : SQLAlchemy connectable(engine/connection) or DBAPI2 connection (fallback mode)
+    con : SQLAlchemy connectable(engine/connection/session) or DBAPI2 connection (fallback mode)
         Using SQLAlchemy makes it possible to use any DB supported by that
         library.
         If a DBAPI2 object, only sqlite3 is supported.

pandas/io/tests/test_sql.py

@@ -918,6 +918,34 @@ def test_sqlalchemy_type_mapping(self):
         table = sql.SQLTable("test_type", db, frame=df)
         self.assertTrue(isinstance(table.table.c['time'].type, sqltypes.DateTime))
 
+    def test_session(self):
+        """
+        read_sql_query should work within a session.
+        a temporary table created within the session should be able to be queried.
+        """
+        if sqlalchemy.__version__ < '1.0.0':
+            nose.SkipTest('session requires sqlalchemy>=1.0.0')
+        from sqlalchemy.orm import sessionmaker
+        session = sessionmaker(bind=self.conn)()
+        # create a temporary table within a session
+        # this is contrived example or a temporary tables but they can be really useful
+        session.execute("""CREATE TEMPORARY TABLE temp_iris AS SELECT * FROM iris LIMIT 5""")
+        # read_sql_query can read from the temporary table
+        iris_frame = sql.read_sql_query("SELECT * FROM temp_iris", session)
+        assert(len(iris_frame) == 5)
+
+    def test_session_close(self):
+        """read_sql_query shouldn't close the session"""
+        if sqlalchemy.__version__ < '1.0.0':
+            nose.SkipTest('session requires sqlalchemy>=1.0.0')
+        from sqlalchemy.orm import sessionmaker
+        session = sessionmaker(bind=self.conn)()
+        session.execute("""CREATE TEMPORARY TABLE temp_iris AS SELECT * FROM iris LIMIT 5""")
+        sql.read_sql_query("SELECT count(1) FROM temp_iris", session)
+        # run again to test that the session hasn't been closed by the last call
+        iris_frame = sql.read_sql_query("SELECT * FROM temp_iris", session)
+        assert(len(iris_frame) == 5)
+
 
 class _EngineToConnMixin(object):
     """