Add "INSERT IGNORE" mode to mysql.to_sql() (#1429)

* Add mode for INSERT INGORE

* Update comment

* Update new mode name

* Update name of new mode

* Fix syntax

* Reduce line length

* Empty commit to retry build

* Add test

* Fix test

Co-authored-by: jaidisido <[email protected]>
Co-authored-by: Lucas Hanson <[email protected]>

Author: 3 people

awswrangler/mysql.py

@@ -335,7 +335,7 @@ def to_sql(
     schema : str
         Schema name
     mode : str
-        Append, overwrite, upsert_duplicate_key, upsert_replace_into, upsert_distinct.
+        Append, overwrite, upsert_duplicate_key, upsert_replace_into, upsert_distinct, ignore.
             append: Inserts new records into table.
             overwrite: Drops table and recreates.
             upsert_duplicate_key: Performs an upsert using `ON DUPLICATE KEY` clause. Requires table schema to have
@@ -345,6 +345,7 @@ def to_sql(
             upsert_distinct: Inserts new records, including duplicates, then recreates the table and inserts `DISTINCT`
             records from old table. This is the least efficient approach but handles scenarios where there are no
             keys on table.
+            ignore: Inserts new records into table using `INSERT IGNORE` clause.
 
     index : bool
         True to store the DataFrame index as a column in the table,
@@ -395,6 +396,7 @@ def to_sql(
         "upsert_replace_into",
         "upsert_duplicate_key",
         "upsert_distinct",
+        "ignore",
     ]
     _db_utils.validate_mode(mode=mode, allowed_modes=allowed_modes)
     _validate_connection(con=con)
@@ -416,6 +418,7 @@ def to_sql(
             insertion_columns = ""
             upsert_columns = ""
             upsert_str = ""
+            ignore_str = " IGNORE" if mode == "ignore" else ""
             if use_column_names:
                 insertion_columns = f"(`{'`, `'.join(df.columns)}`)"
             if mode == "upsert_duplicate_key":
@@ -429,7 +432,8 @@ def to_sql(
                 if mode == "upsert_replace_into":
                     sql = f"REPLACE INTO `{schema}`.`{table}` {insertion_columns} VALUES {placeholders}"
                 else:
-                    sql = f"INSERT INTO `{schema}`.`{table}` {insertion_columns} VALUES {placeholders}{upsert_str}"
+                    sql = f"""INSERT{ignore_str} INTO `{schema}`.`{table}` {insertion_columns}
+VALUES {placeholders}{upsert_str}"""
                 _logger.debug("sql: %s", sql)
                 cursor.executemany(sql, (parameters,))
             con.commit()

tests/test_mysql.py

@@ -369,3 +369,36 @@ def test_dfs_are_equal_for_different_chunksizes(mysql_table, mysql_con, chunksiz
     df["c1"] = df["c1"].astype("string")
 
     assert df.equals(df2)
+
+
+def test_ignore(mysql_table, mysql_con):
+    create_table_sql = (
+        f"CREATE TABLE test.{mysql_table} "
+        "(c0 varchar(100) PRIMARY KEY, "
+        "c1 INT DEFAULT 42 NULL, "
+        "c2 INT NOT NULL);"
+    )
+    with mysql_con.cursor() as cursor:
+        cursor.execute(create_table_sql)
+        mysql_con.commit()
+
+    df = pd.DataFrame({"c0": ["foo", "bar"], "c2": [1, 2]})
+
+    wr.mysql.to_sql(df=df, con=mysql_con, schema="test", table=mysql_table, mode="ignore", use_column_names=True)
+    wr.mysql.to_sql(df=df, con=mysql_con, schema="test", table=mysql_table, mode="ignore", use_column_names=True)
+    df2 = wr.mysql.read_sql_table(con=mysql_con, schema="test", table=mysql_table)
+    assert bool(len(df2) == 2)
+
+    wr.mysql.to_sql(df=df, con=mysql_con, schema="test", table=mysql_table, mode="ignore", use_column_names=True)
+    df3 = pd.DataFrame({"c0": ["baz", "bar"], "c2": [3, 2]})
+    wr.mysql.to_sql(df=df3, con=mysql_con, schema="test", table=mysql_table, mode="ignore", use_column_names=True)
+    df4 = wr.mysql.read_sql_table(con=mysql_con, schema="test", table=mysql_table)
+    assert bool(len(df4) == 3)
+
+    df5 = pd.DataFrame({"c0": ["foo", "bar"], "c2": [4, 5]})
+    wr.mysql.to_sql(df=df5, con=mysql_con, schema="test", table=mysql_table, mode="ignore", use_column_names=True)
+
+    df6 = wr.mysql.read_sql_table(con=mysql_con, schema="test", table=mysql_table)
+    assert bool(len(df6) == 3)
+    assert bool(len(df6.loc[(df6["c0"] == "foo") & (df6["c2"] == 1)]) == 1)
+    assert bool(len(df6.loc[(df6["c0"] == "bar") & (df6["c2"] == 2)]) == 1)