SQL: fix type conversion for Timestamp object typed columns (GH9085)

doc/source/whatsnew/v0.16.0.txt

@@ -58,6 +58,8 @@ Bug Fixes
 
 
 
+- Fixed bug in ``to_sql`` when mapping a Timestamp object column (datetime
+  column with timezone info) to the according sqlalchemy type (:issue:`9085`).
 
 
 

pandas/io/sql.py

@@ -911,7 +911,7 @@ def _sqlalchemy_type(self, col):
         from sqlalchemy.types import (BigInteger, Float, Text, Boolean,
             DateTime, Date, Time)
 
-        if col_type == 'datetime64': 
+        if col_type == 'datetime64' or col_type == 'datetime':
             try:
                 tz = col.tzinfo
                 return DateTime(timezone=True)

pandas/io/tests/test_sql.py

@@ -792,6 +792,15 @@ def _get_index_columns(self, tbl_name):
         ixs = [i['column_names'] for i in ixs]
         return ixs
 
+    def test_sqlalchemy_type_mapping(self):
+
+        # Test Timestamp objects (no datetime64 because of timezone) (GH9085)
+        df = DataFrame({'time': to_datetime(['201412120154', '201412110254'],
+                                            utc=True)})
+        db = sql.SQLDatabase(self.conn)
+        table = sql.SQLTable("test_type", db, frame=df)
+        self.assertTrue(isinstance(table.table.c['time'].type, sqltypes.DateTime))
+
 
 class TestSQLiteFallbackApi(_TestSQLApi):
     """
@@ -853,6 +862,24 @@ def test_uquery(self):
             rows = sql.uquery("SELECT * FROM iris LIMIT 1", con=self.conn)
         self.assertEqual(rows, -1)
 
+    def _get_sqlite_column_type(self, schema, column):
+
+        for col in schema.split('\n'):
+            if col.split()[0].strip('[]') == column:
+                return col.split()[1]
+        raise ValueError('Column %s not found' % (column))
+
+    def test_sqlite_type_mapping(self):
+
+        # Test Timestamp objects (no datetime64 because of timezone) (GH9085)
+        df = DataFrame({'time': to_datetime(['201412120154', '201412110254'],
+                                            utc=True)})
+        db = sql.SQLiteDatabase(self.conn, self.flavor)
+        table = sql.SQLiteTable("test_type", db, frame=df)
+        schema = table.sql_schema()
+        self.assertEqual(self._get_sqlite_column_type(schema, 'time'),
+                         "TIMESTAMP")
+
 
 #------------------------------------------------------------------------------
 #--- Database flavor specific tests
@@ -1550,15 +1577,15 @@ def test_dtype(self):
 
         # sqlite stores Boolean values as INTEGER
         self.assertEqual(self._get_sqlite_column_type('dtype_test', 'B'), 'INTEGER')
-        
+
         self.assertEqual(self._get_sqlite_column_type('dtype_test2', 'B'), 'STRING')
         self.assertRaises(ValueError, df.to_sql,
                           'error', self.conn, dtype={'B': bool})
 
     def test_notnull_dtype(self):
         if self.flavor == 'mysql':
             raise nose.SkipTest('Not applicable to MySQL legacy')
-            
+
         cols = {'Bool': Series([True,None]),
                 'Date': Series([datetime(2012, 5, 1), None]),
                 'Int' : Series([1, None], dtype='object'),