ENH/TST: Raise error if invalid value passed to if_exists argument

The if_exists argument in io.sql.write_frame needed data validation
because the logic of the function implicitly used 'append' if the argument
value was any string that was not either 'fail' or 'replace'.
I added a new unit test to support the requirement.

BUG: Fix if_exists='replace' functionality in io.sql.write_frame

This should resolve issues pandas-dev#2971 and pandas-dev#4110

CLN: Refactor in between test clean ups to be more DRY

TST: Complete test coverage for if_exists uses in io.sql.write_frame

CLN: Refactor to make interaction between exists and if_exists clearer

This refactor results in the function logic being clearer, since if_exists is
only relevant when exists is True, the program flow is better served to
have if_exists control flow only when exists is True

BUG: Fix regression introduced by c28f11a0041a9f3b25f33b0539e42fa802b1d8d4

sqlite3 convenience function executescript not available
in other database flavors.

TST: Adding if_exist test for mysql flavor

Author: davidshinn

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'],