Merge pull request #6164 from hayd/4304r

4304r if_exists == 'fail' should actually work

Author: jreback

doc/source/release.rst

@@ -154,6 +154,7 @@ Bug Fixes
   - Possible segfault when chained indexing with an object array under numpy 1.7.1 (:issue:`6026`, :issue:`6056`)
   - Bug in setting using fancy indexing a single element with a non-scalar (e.g. a list),
     (:issue:`6043`)
+  - ``to_sql`` did not respect ``if_exists`` (:issue:`4110` :issue:`4304`)
   - Regression in ``.get(None)`` indexing from 0.12 (:issue:`5652`)
   - Subtle ``iloc`` indexing bug, surfaced in (:issue:`6059`)
   - Bug with insert of strings into DatetimeIndex (:issue:`5818`)

pandas/io/sql.py

@@ -200,15 +200,25 @@ def write_frame(frame, name, con, flavor='sqlite', if_exists='fail', **kwargs):
             if_exists = 'append'
         else:
             if_exists = 'fail'
+
+    if if_exists not in ('fail', 'replace', 'append'):
+        raise ValueError("'%s' is not valid for if_exists" % if_exists)
+
     exists = table_exists(name, con, flavor)
     if if_exists == 'fail' and exists:
         raise ValueError("Table '%s' already exists." % name)
 
-    #create or drop-recreate if necessary
+    # creation/replacement dependent on the table existing and if_exist criteria
     create = None
-    if exists and if_exists == 'replace':
-        create = "DROP TABLE %s" % name
-    elif not exists:
+    if exists:
+        if if_exists == 'fail':
+            raise ValueError("Table '%s' already exists." % name)
+        elif if_exists == 'replace':
+            cur = con.cursor()
+            cur.execute("DROP TABLE %s;" % name)
+            cur.close()
+            create = get_schema(frame, name, flavor)
+    else:
         create = get_schema(frame, name, flavor)
 
     if create is not None:

pandas/io/tests/test_sql.py

@@ -240,6 +240,65 @@ def test_onecolumn_of_integer(self):
         result = sql.read_frame("select * from mono_df",con_x)
         tm.assert_frame_equal(result,mono_df)
 
+    def test_if_exists(self):
+        df_if_exists_1 = DataFrame({'col1': [1, 2], 'col2': ['A', 'B']})
+        df_if_exists_2 = DataFrame({'col1': [3, 4, 5], 'col2': ['C', 'D', 'E']})
+        table_name = 'table_if_exists'
+        sql_select = "SELECT * FROM %s" % table_name
+
+        def clean_up(test_table_to_drop):
+            """
+            Drops tables created from individual tests
+            so no dependencies arise from sequential tests
+            """
+            if sql.table_exists(test_table_to_drop, self.db, flavor='sqlite'):
+                cur = self.db.cursor()
+                cur.execute("DROP TABLE %s" % test_table_to_drop)
+                cur.close()
+
+        # test if invalid value for if_exists raises appropriate error
+        self.assertRaises(ValueError,
+                          sql.write_frame,
+                          frame=df_if_exists_1,
+                          con=self.db,
+                          name=table_name,
+                          flavor='sqlite',
+                          if_exists='notvalidvalue')
+        clean_up(table_name)
+
+        # test if_exists='fail'
+        sql.write_frame(frame=df_if_exists_1, con=self.db, name=table_name,
+                        flavor='sqlite', if_exists='fail')
+        self.assertRaises(ValueError,
+                          sql.write_frame,
+                          frame=df_if_exists_1,
+                          con=self.db,
+                          name=table_name,
+                          flavor='sqlite',
+                          if_exists='fail')
+
+        # test if_exists='replace'
+        sql.write_frame(frame=df_if_exists_1, con=self.db, name=table_name,
+                        flavor='sqlite', if_exists='replace')
+        self.assertEqual(sql.tquery(sql_select, con=self.db),
+                         [(1, 'A'), (2, 'B')])
+        sql.write_frame(frame=df_if_exists_2, con=self.db, name=table_name,
+                        flavor='sqlite', if_exists='replace')
+        self.assertEqual(sql.tquery(sql_select, con=self.db),
+                         [(3, 'C'), (4, 'D'), (5, 'E')])
+        clean_up(table_name)
+                        
+        # test if_exists='append'
+        sql.write_frame(frame=df_if_exists_1, con=self.db, name=table_name,
+                        flavor='sqlite', if_exists='fail')
+        self.assertEqual(sql.tquery(sql_select, con=self.db),
+                         [(1, 'A'), (2, 'B')])
+        sql.write_frame(frame=df_if_exists_2, con=self.db, name=table_name,
+                        flavor='sqlite', if_exists='append')
+        self.assertEqual(sql.tquery(sql_select, con=self.db),
+                         [(1, 'A'), (2, 'B'), (3, 'C'), (4, 'D'), (5, 'E')])
+        clean_up(table_name)
+
 
 class TestMySQL(tm.TestCase):
 
@@ -483,6 +542,66 @@ def test_keyword_as_column_names(self):
         sql.write_frame(df, con = self.db, name = 'testkeywords',
                         if_exists='replace', flavor='mysql')
 
+    def test_if_exists(self):
+        _skip_if_no_MySQLdb()
+        df_if_exists_1 = DataFrame({'col1': [1, 2], 'col2': ['A', 'B']})
+        df_if_exists_2 = DataFrame({'col1': [3, 4, 5], 'col2': ['C', 'D', 'E']})
+        table_name = 'table_if_exists'
+        sql_select = "SELECT * FROM %s" % table_name
+
+        def clean_up(test_table_to_drop):
+            """
+            Drops tables created from individual tests
+            so no dependencies arise from sequential tests
+            """
+            if sql.table_exists(test_table_to_drop, self.db, flavor='mysql'):
+                cur = self.db.cursor()
+                cur.execute("DROP TABLE %s" % test_table_to_drop)
+                cur.close()
+
+        # test if invalid value for if_exists raises appropriate error
+        self.assertRaises(ValueError,
+                          sql.write_frame,
+                          frame=df_if_exists_1,
+                          con=self.db,
+                          name=table_name,
+                          flavor='mysql',
+                          if_exists='notvalidvalue')
+        clean_up(table_name)
+
+        # test if_exists='fail'
+        sql.write_frame(frame=df_if_exists_1, con=self.db, name=table_name,
+                        flavor='mysql', if_exists='fail')
+        self.assertRaises(ValueError,
+                          sql.write_frame,
+                          frame=df_if_exists_1,
+                          con=self.db,
+                          name=table_name,
+                          flavor='mysql',
+                          if_exists='fail')
+
+        # test if_exists='replace'
+        sql.write_frame(frame=df_if_exists_1, con=self.db, name=table_name,
+                        flavor='mysql', if_exists='replace')
+        self.assertEqual(sql.tquery(sql_select, con=self.db),
+                         [(1, 'A'), (2, 'B')])
+        sql.write_frame(frame=df_if_exists_2, con=self.db, name=table_name,
+                        flavor='mysql', if_exists='replace')
+        self.assertEqual(sql.tquery(sql_select, con=self.db),
+                         [(3, 'C'), (4, 'D'), (5, 'E')])
+        clean_up(table_name)
+                        
+        # test if_exists='append'
+        sql.write_frame(frame=df_if_exists_1, con=self.db, name=table_name,
+                        flavor='mysql', if_exists='fail')
+        self.assertEqual(sql.tquery(sql_select, con=self.db),
+                         [(1, 'A'), (2, 'B')])
+        sql.write_frame(frame=df_if_exists_2, con=self.db, name=table_name,
+                        flavor='mysql', if_exists='append')
+        self.assertEqual(sql.tquery(sql_select, con=self.db),
+                         [(1, 'A'), (2, 'B'), (3, 'C'), (4, 'D'), (5, 'E')])
+        clean_up(table_name)
+
 
 if __name__ == '__main__':
     nose.runmodule(argv=[__file__, '-vvs', '-x', '--pdb', '--pdb-failure'],