gn3: db: Create a raw update query

* gn3/db/__init__.py (update_raw): New function.

1 files changed, 15 insertions, 1 deletions

diff --git a/gn3/db/__init__.py b/gn3/db/__init__.py
index 5ab9f3c..24ae8f1 100644
--- a/gn3/db/__init__.py
+++ b/gn3/db/__init__.py
@@ -1,7 +1,7 @@
 # pylint: disable=[R0902, R0903]
 """Module that exposes common db operations"""
 from dataclasses import asdict, astuple
-from typing import Any, Dict, List, Optional, Generator, Union
+from typing import Any, Dict, List, Optional, Generator, Tuple, Union
 from typing_extensions import Protocol
 
 from gn3.db.metadata_audit import MetadataAudit
@@ -64,6 +64,20 @@ def update(conn: Any,
         return cursor.rowcount
 
 
+def update_raw(conn: Any, table: str,
+               set_: List[Tuple[str, Any]],
+               where: Tuple[str, Tuple]):
+    """Run a generic raw statement"""
+    sql = f"UPDATE {table} SET "
+    sql += ", ".join([f"{k} = '%s'" for k, v in set_])
+    sql += f" WHERE {where[0]}"
+    with conn.cursor() as cursor:
+        cursor.execute(sql,
+                       tuple(v for _, v in set_) + where[1])
+        conn.commit()
+        return cursor.rowcount
+
+
 def fetchone(conn: Any,
              table: str,
              where: Optional[Dataclass],