feat: allow passing upsert_fields with None to SQLizer.upsert_on_duplicate

Author: NightMarcher

examples/service/routers/account.py

@@ -153,7 +153,7 @@ async def update_last_login_view(
 @router.post("/bulk_upsert")
 async def bulk_upsert_view():
     dicts = []
-    for (idx, locale) in enumerate([LocaleEnum.ja_JP, LocaleEnum.en_IN, LocaleEnum.zh_CN], 7):
+    for (idx, locale) in enumerate([LocaleEnum.ja_JP, LocaleEnum.en_IN, LocaleEnum.zh_CN], 10):
         faker = Faker(locale)
         dicts.append({
             "id": idx,
@@ -165,8 +165,7 @@ async def bulk_upsert_view():
     row_cnt = await AccountMgr.upsert_on_duplicate(
         dicts,
         insert_fields=["id", "gender", "name", "locale", "extend"],
-        upsert_fields=["name", "locale"],
-        using_values=False,
+        # upsert_fields=["name", "locale"],
     )
     return {"row_cnt": row_cnt}
 

fastapi_esql/orm/base_manager.py

@@ -149,7 +149,7 @@ async def upsert_on_duplicate(
         cls,
         dicts: List[Dict[str, Any]],
         insert_fields: List[str],
-        upsert_fields: List[str],
+        upsert_fields: Optional[List[str]] = None,
         using_values: bool = False,
     ):
         sql = SQLizer.upsert_on_duplicate(

fastapi_esql/utils/sqlizer.py

@@ -195,37 +195,38 @@ def upsert_on_duplicate(
         table: str,
         dicts: List[Dict[str, Any]],
         insert_fields: List[str],
-        upsert_fields: List[str],
+        upsert_fields: Optional[List[str]] = None,
         using_values: bool = False,
     ) -> Optional[str]:
-        if not all([table, dicts, insert_fields, upsert_fields]):
-            raise WrongParamsError("Parameters `table`, `dicts`, `insert_fields`, `upsert_fields` are required")
+        if not all([table, dicts, insert_fields]):
+            raise WrongParamsError("Parameters `table`, `dicts`, `insert_fields` are required")
 
         values = [
             f"      ({', '.join(cls.sqlize_value(d.get(f)) for f in insert_fields)})"
             for d in dicts
         ]
         # NOTE Beginning with MySQL 8.0.19, it is possible to use an alias for the row
         # https://dev.mysql.com/doc/refman/8.0/en/insert-on-duplicate.html
-        if using_values:
-            upserts = [f"{field}=VALUES({field})" for field in upsert_fields]
-            on_duplicate = f"ON DUPLICATE KEY UPDATE {', '.join(upserts)}"
-        else:
-            new_table = f"`new_{table}`"
-            upserts = [f"{field}={new_table}.{field}" for field in upsert_fields]
-            on_duplicate = f"AS {new_table} ON DUPLICATE KEY UPDATE {', '.join(upserts)}"
+        on_duplicate = ""
+        if upsert_fields:
+            if using_values:
+                upserts = [f"{field}=VALUES({field})" for field in upsert_fields]
+                on_duplicate = f"ON DUPLICATE KEY UPDATE {', '.join(upserts)}"
+            else:
+                new_table = f"`new_{table}`"
+                upserts = [f"{field}={new_table}.{field}" for field in upsert_fields]
+                on_duplicate = f"AS {new_table} ON DUPLICATE KEY UPDATE {', '.join(upserts)}"
 
         sql = """
     INSERT INTO {}
       ({})
     VALUES
-{}
-    {}
+{}{}
 """.format(
         table,
         ", ".join(insert_fields),
         ",\n".join(values),
-        on_duplicate,
+        f"\n    {on_duplicate}" if on_duplicate else "",
     )
         logger.debug(sql)
         return sql

tests/test_sqlizer.py

@@ -251,6 +251,24 @@ def test_upsert_on_duplicate(self):
     AS `new_account` ON DUPLICATE KEY UPDATE name=`new_account`.name, locale=`new_account`.locale
 """
 
+        only_insert_sql = SQLizer.upsert_on_duplicate(
+            self.table,
+            [
+                {'id': 7, 'gender': 1, 'name': '斉藤 修平', 'locale': 'ja_JP', 'extend': {}},
+                {'id': 8, 'gender': 1, 'name': 'Ojas Salvi', 'locale': 'en_IN', 'extend': {}},
+                {'id': 9, 'gender': 1, 'name': '羊淑兰', 'locale': 'zh_CN', 'extend': {}}
+            ],
+            insert_fields=["id", "gender", "name", "locale", "extend"],
+        )
+        assert only_insert_sql == """
+    INSERT INTO account
+      (id, gender, name, locale, extend)
+    VALUES
+      (7, 1, '斉藤 修平', 'ja_JP', '{}'),
+      (8, 1, 'Ojas Salvi', 'en_IN', '{}'),
+      (9, 1, '羊淑兰', 'zh_CN', '{}')
+"""
+
     def test_insert_into_select(self):
         with self.assertRaises(WrongParamsError):
             SQLizer.insert_into_select(