feat: implement option 'delete_rows' of argument 'if_exists' in 'DataFrame.to_sql' API.

Author: gmcrocetti

doc/source/whatsnew/v3.0.0.rst

@@ -73,6 +73,7 @@ Other enhancements
 - :py:class:`frozenset` elements in pandas objects are now natively printed (:issue:`60690`)
 - Errors occurring during SQL I/O will now throw a generic :class:`.DatabaseError` instead of the raw Exception type from the underlying driver manager library (:issue:`60748`)
 - Implemented :meth:`Series.str.isascii` and :meth:`Series.str.isascii` (:issue:`59091`)
+- Add ``"delete_rows"`` option to ``if_exists`` argument in :meth:`DataFrame.to_sql` deleting all records of the table before inserting data (:issue:`37210`).
 - Multiplying two :class:`DateOffset` objects will now raise a ``TypeError`` instead of a ``RecursionError`` (:issue:`59442`)
 - Restore support for reading Stata 104-format and enable reading 103-format dta files (:issue:`58554`)
 - Support passing a :class:`Iterable[Hashable]` input to :meth:`DataFrame.drop_duplicates` (:issue:`59237`)

pandas/io/sql.py

@@ -738,7 +738,7 @@ def to_sql(
     name: str,
     con,
     schema: str | None = None,
-    if_exists: Literal["fail", "replace", "append"] = "fail",
+    if_exists: Literal["fail", "replace", "append", "delete_rows"] = "fail",
     index: bool = True,
     index_label: IndexLabel | None = None,
     chunksize: int | None = None,
@@ -764,10 +764,11 @@ def to_sql(
     schema : str, optional
         Name of SQL schema in database to write to (if database flavor
         supports this). If None, use default schema (default).
-    if_exists : {'fail', 'replace', 'append'}, default 'fail'
+    if_exists : {'fail', 'replace', 'append', 'delete_rows'}, default 'fail'
         - fail: If table exists, do nothing.
         - replace: If table exists, drop it, recreate it, and insert data.
         - append: If table exists, insert data. Create if does not exist.
+        - delete_rows: If a table exists, delete all records and insert data.
     index : bool, default True
         Write DataFrame index as a column.
     index_label : str or sequence, optional
@@ -818,7 +819,7 @@ def to_sql(
     `sqlite3 <https://docs.python.org/3/library/sqlite3.html#sqlite3.Cursor.rowcount>`__ or
     `SQLAlchemy <https://docs.sqlalchemy.org/en/14/core/connections.html#sqlalchemy.engine.BaseCursorResult.rowcount>`__
     """  # noqa: E501
-    if if_exists not in ("fail", "replace", "append"):
+    if if_exists not in ("fail", "replace", "append", "delete_rows"):
         raise ValueError(f"'{if_exists}' is not valid for if_exists")
 
     if isinstance(frame, Series):
@@ -926,7 +927,7 @@ def __init__(
         pandas_sql_engine,
         frame=None,
         index: bool | str | list[str] | None = True,
-        if_exists: Literal["fail", "replace", "append"] = "fail",
+        if_exists: Literal["fail", "replace", "append", "delete_rows"] = "fail",
         prefix: str = "pandas",
         index_label=None,
         schema=None,
@@ -974,11 +975,13 @@ def create(self) -> None:
         if self.exists():
             if self.if_exists == "fail":
                 raise ValueError(f"Table '{self.name}' already exists.")
-            if self.if_exists == "replace":
+            elif self.if_exists == "replace":
                 self.pd_sql.drop_table(self.name, self.schema)
                 self._execute_create()
             elif self.if_exists == "append":
                 pass
+            elif self.if_exists == "delete_rows":
+                self.pd_sql.delete_rows(self.name, self.schema)
             else:
                 raise ValueError(f"'{self.if_exists}' is not valid for if_exists")
         else:
@@ -1480,7 +1483,7 @@ def to_sql(
         self,
         frame,
         name: str,
-        if_exists: Literal["fail", "replace", "append"] = "fail",
+        if_exists: Literal["fail", "replace", "append", "delete_rows"] = "fail",
         index: bool = True,
         index_label=None,
         schema=None,
@@ -1874,7 +1877,7 @@ def prep_table(
         self,
         frame,
         name: str,
-        if_exists: Literal["fail", "replace", "append"] = "fail",
+        if_exists: Literal["fail", "replace", "append", "delete_rows"] = "fail",
         index: bool | str | list[str] | None = True,
         index_label=None,
         schema=None,
@@ -1951,7 +1954,7 @@ def to_sql(
         self,
         frame,
         name: str,
-        if_exists: Literal["fail", "replace", "append"] = "fail",
+        if_exists: Literal["fail", "replace", "append", "delete_rows"] = "fail",
         index: bool = True,
         index_label=None,
         schema: str | None = None,
@@ -1969,10 +1972,11 @@ def to_sql(
         frame : DataFrame
         name : string
             Name of SQL table.
-        if_exists : {'fail', 'replace', 'append'}, default 'fail'
+        if_exists : {'fail', 'replace', 'append', 'delete_rows'}, default 'fail'
             - fail: If table exists, do nothing.
             - replace: If table exists, drop it, recreate it, and insert data.
             - append: If table exists, insert data. Create if does not exist.
+            - delete_rows: If a table exists, delete all records and insert data.
         index : boolean, default True
             Write DataFrame index as a column.
         index_label : string or sequence, default None
@@ -2069,6 +2073,18 @@ def drop_table(self, table_name: str, schema: str | None = None) -> None:
                 self.get_table(table_name, schema).drop(bind=self.con)
             self.meta.clear()
 
+    def delete_rows(self, table_name: str, schema: str | None = None) -> None:
+        schema = schema or self.meta.schema
+        if self.has_table(table_name, schema):
+            self.meta.reflect(
+                bind=self.con, only=[table_name], schema=schema, views=True
+            )
+            with self.run_transaction() as con:
+                table = self.get_table(table_name, schema)
+                con.execute(table.delete())
+
+            self.meta.clear()
+
     def _create_sql_schema(
         self,
         frame: DataFrame,
@@ -2304,7 +2320,7 @@ def to_sql(
         self,
         frame,
         name: str,
-        if_exists: Literal["fail", "replace", "append"] = "fail",
+        if_exists: Literal["fail", "replace", "append", "delete_rows"] = "fail",
         index: bool = True,
         index_label=None,
         schema: str | None = None,
@@ -2326,6 +2342,7 @@ def to_sql(
             - fail: If table exists, do nothing.
             - replace: If table exists, drop it, recreate it, and insert data.
             - append: If table exists, insert data. Create if does not exist.
+            - delete_rows: If a table exists, delete all records and insert data.
         index : boolean, default True
             Write DataFrame index as a column.
         index_label : string or sequence, default None
@@ -2416,6 +2433,12 @@ def has_table(self, name: str, schema: str | None = None) -> bool:
 
         return False
 
+    def delete_rows(self, name: str, schema: str | None = None) -> None:
+        delete_sql = f"DELETE FROM {schema}.{name}" if schema else f"DELETE FROM {name}"
+        if self.has_table(name, schema):
+            with self.con.cursor() as cur:
+                cur.execute(delete_sql)
+
     def _create_sql_schema(
         self,
         frame: DataFrame,
@@ -2790,10 +2813,11 @@ def to_sql(
         frame: DataFrame
         name: string
             Name of SQL table.
-        if_exists: {'fail', 'replace', 'append'}, default 'fail'
+        if_exists: {'fail', 'replace', 'append', 'delete_rows'}, default 'fail'
             fail: If table exists, do nothing.
             replace: If table exists, drop it, recreate it, and insert data.
             append: If table exists, insert data. Create if it does not exist.
+            delete_rows: If a table exists, delete all records and insert data.
         index : bool, default True
             Write DataFrame index as a column
         index_label : string or sequence, default None
@@ -2869,6 +2893,12 @@ def drop_table(self, name: str, schema: str | None = None) -> None:
         drop_sql = f"DROP TABLE {_get_valid_sqlite_name(name)}"
         self.execute(drop_sql)
 
+    def delete_rows(self, name: str, schema: str | None = None) -> None:
+        delete_sql = f"DELETE FROM {_get_valid_sqlite_name(name)}"
+        if self.has_table(name, schema):
+            with self.run_transaction() as cur:
+                cur.execute(delete_sql)
+
     def _create_sql_schema(
         self,
         frame,

pandas/tests/io/test_sql.py

@@ -1068,7 +1068,9 @@ def test_to_sql(conn, method, test_frame1, request):
 
 
 @pytest.mark.parametrize("conn", all_connectable)
-@pytest.mark.parametrize("mode, num_row_coef", [("replace", 1), ("append", 2)])
+@pytest.mark.parametrize(
+    "mode, num_row_coef", [("replace", 1), ("append", 2), ("delete_rows", 1)]
+)
 def test_to_sql_exist(conn, mode, num_row_coef, test_frame1, request):
     conn = request.getfixturevalue(conn)
     with pandasSQL_builder(conn, need_transaction=True) as pandasSQL:
@@ -2698,6 +2700,64 @@ def test_drop_table(conn, request):
         assert not insp.has_table("temp_frame")
 
 
+@pytest.mark.parametrize("conn", all_connectable)
+def test_delete_rows_success(conn, test_frame1, request):
+    table_name = "temp_frame"
+    conn = request.getfixturevalue(conn)
+    pandasSQL = pandasSQL_builder(conn)
+
+    with pandasSQL.run_transaction():
+        assert pandasSQL.to_sql(test_frame1, table_name) == test_frame1.shape[0]
+
+    with pandasSQL.run_transaction():
+        assert pandasSQL.delete_rows(table_name) is None
+
+    assert count_rows(conn, table_name) == 0
+    assert pandasSQL.has_table("temp_frame")
+
+
+@pytest.mark.parametrize("conn", all_connectable)
+def test_delete_rows_is_atomic(conn, request):
+    adbc_driver_manager = pytest.importorskip("adbc_driver_manager")
+    sqlalchemy = pytest.importorskip("sqlalchemy")
+
+    if "sqlite" in conn:
+        reason = "This test relies on strict column types, SQLite has a dynamic one"
+        request.applymarker(
+            pytest.mark.xfail(
+                reason=reason,
+                strict=True,
+            )
+        )
+
+    table_name = "temp_frame"
+    original_df = DataFrame({"a": [1, 2, 3]})
+    replacing_df = DataFrame({"a": ["a", "b", "c", "d"]})
+
+    conn = request.getfixturevalue(conn)
+    pandasSQL = pandasSQL_builder(conn)
+
+    if isinstance(conn, adbc_driver_manager.dbapi.Connection):
+        expected_exception = adbc_driver_manager.ProgrammingError
+    else:
+        expected_exception = sqlalchemy.exc.DataError
+
+    with pandasSQL.run_transaction():
+        pandasSQL.to_sql(original_df, table_name, if_exists="fail", index=False)
+
+    # trying to insert strings in an integer column
+    with pytest.raises(expected_exception):
+        with pandasSQL.run_transaction():
+            pandasSQL.to_sql(
+                replacing_df, table_name, if_exists="delete_rows", index=False
+            )
+
+    # failed "delete_rows" is rolled back preserving original data
+    with pandasSQL.run_transaction():
+        result_df = pandasSQL.read_query(f"SELECT * FROM {table_name}")
+        tm.assert_frame_equal(result_df, original_df)
+
+
 @pytest.mark.parametrize("conn", all_connectable)
 def test_roundtrip(conn, request, test_frame1):
     if conn == "sqlite_str":