feat: add using_values to upsert_on_duplicated and build_fly_table

Author: NightMarcher

README.md

@@ -115,6 +115,7 @@ await AccountMgr.upsert_on_duplicated(
     ],
     insert_fields=["id", "gender", "name", "locale", "extend"],
     upsert_fields=["name", "locale"],
+    using_values=False,
 )
 ```
 Generate sql and execute
@@ -123,8 +124,7 @@ Generate sql and execute
         (id, gender, name, locale, extend)
     VALUES
         (7, 1, '斉藤 修平', 'ja_JP', '{}'), (8, 1, 'Ojas Salvi', 'en_IN', '{}'), (9, 1, '羊淑兰', 'zh_CN', '{}')
-    AS `new_account`
-    ON DUPLICATE KEY UPDATE name=`new_account`.name, locale=`new_account`.locale
+    AS `new_account` ON DUPLICATE KEY UPDATE name=`new_account`.name, locale=`new_account`.locale
 ```
 
 ### **insert_into_select**
@@ -158,6 +158,7 @@ await AccountMgr.bulk_update_with_fly_table(
     ],
     join_fields=["id"],
     update_fields=["active", "gender"],
+    using_values=True,
 )
 ```
 Generate sql and execute
@@ -166,7 +167,7 @@ Generate sql and execute
     JOIN (
         SELECT * FROM (
             VALUES
-                ROW(7, False, 1), ROW(15, True, 0)
+            ROW(7, False, 1), ROW(15, True, 0)
         ) AS fly_table (id, active, gender)
     ) tmp ON account.id=tmp.id
     SET account.active=tmp.active, account.gender=tmp.gender

examples/service/routers/account.py

@@ -153,6 +153,7 @@ async def bulk_upsert_view():
         dicts,
         insert_fields=["id", "gender", "name", "locale", "extend"],
         upsert_fields=["name", "locale"],
+        using_values=False,
     )
     return {"row_cnt": row_cnt}
 
@@ -187,5 +188,6 @@ async def bulk_update_view(
         [d.dict() for d in dicts],
         join_fields=["id"],
         update_fields=["active", "gender"],
+        using_values=True,
     )
     return {"row_cnt": row_cnt}

fastapi_esql/orm/base_manager.py

@@ -110,12 +110,14 @@ async def upsert_on_duplicated(
         dicts: List[Dict[str, Any]],
         insert_fields: List[str],
         upsert_fields: List[str],
+        using_values: bool = False,
     ):
         sql = SQLizer.upsert_on_duplicated(
             cls.table,
             dicts,
             insert_fields,
             upsert_fields,
+            using_values,
         )
         return await CursorHandler.sum_row_cnt(sql, cls.rw_conn, logger)
 
@@ -143,11 +145,13 @@ async def bulk_update_with_fly_table(
         dicts: List[Dict[str, Any]],
         join_fields: List[str],
         update_fields: List[str],
+        using_values: bool = True,
     ):
         sql = SQLizer.bulk_update_with_fly_table(
             cls.table,
             dicts,
             join_fields,
             update_fields,
+            using_values,
         )
         return await CursorHandler.sum_row_cnt(sql, cls.rw_conn, logger)

fastapi_esql/utils/sqlizer.py

@@ -164,6 +164,7 @@ def upsert_on_duplicated(
         dicts: List[Dict[str, Any]],
         insert_fields: List[str],
         upsert_fields: List[str],
+        using_values: bool = False,
     ) -> Optional[str]:
         if not all([table, dicts, insert_fields, upsert_fields]):
             raise WrongParamsError("Please check your params")
@@ -172,16 +173,23 @@ def upsert_on_duplicated(
             f"({', '.join(cls._sqlize_value(d.get(f)) for f in insert_fields)})"
             for d in dicts
         ]
-        new_table = f"`new_{table}`"
-        upserts = [f"{field}={new_table}.{field}" for field in upsert_fields]
+        # 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_duplicated = 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_duplicated = f"AS {new_table} ON DUPLICATE KEY UPDATE {', '.join(upserts)}"
 
         sql = f"""
     INSERT INTO {table}
         ({", ".join(insert_fields)})
     VALUES
         {", ".join(values)}
-    AS {new_table}
-    ON DUPLICATE KEY UPDATE {", ".join(upserts)}"""
+    {on_duplicated}
+    """
         logger.debug(sql)
         return sql
 
@@ -218,20 +226,30 @@ def build_fly_table(
         cls,
         dicts: List[Dict[str, Any]],
         fields: List[str],
+        using_values: bool = True,
     ) -> Optional[str]:
         if not all([dicts, fields]):
             raise WrongParamsError("Please check your params")
 
-        rows = [
-            f"ROW({', '.join(cls._sqlize_value(d.get(f)) for f in fields)})"
-            for d in dicts
-        ]
+        if using_values:
+            rows = [
+                f"ROW({', '.join(cls._sqlize_value(d.get(f)) for f in fields)})"
+                for d in dicts
+            ]
+            values = "VALUES\n            " + ", ".join(rows)
+            table = f"fly_table ({', '.join(fields)})"
+        else:
+            rows = [
+                f"SELECT {', '.join(f'{cls._sqlize_value(d.get(f))} {f}' for f in fields)}"
+                for d in dicts
+            ]
+            values = " UNION ".join(rows)
+            table = "fly_table"
 
         sql = f"""
         SELECT * FROM (
-            VALUES
-                {", ".join(rows)}
-        ) AS fly_table ({", ".join(fields)})"""
+            {values}
+        ) AS {table}"""
         logger.debug(sql)
         return sql
 
@@ -242,6 +260,7 @@ def bulk_update_with_fly_table(
         dicts: List[Dict[str, Any]],
         join_fields: List[str],
         update_fields: List[str],
+        using_values: bool = True,
     ) -> Optional[str]:
         if not all([table, dicts, join_fields, update_fields]):
             raise WrongParamsError("Please check your params")
@@ -251,7 +270,7 @@ def bulk_update_with_fly_table(
 
         sql = f"""
     UPDATE {table}
-    JOIN ({SQLizer.build_fly_table(dicts, join_fields + update_fields)}
+    JOIN ({SQLizer.build_fly_table(dicts, join_fields + update_fields, using_values)}
     ) tmp ON {", ".join(joins)}
     SET {", ".join(updates)}"""
         logger.debug(sql)