feat: add shortcuts for insert_query and update_query into Database

example: await db.insert("table_name", base_model)

lnbits/db.py

@@ -158,6 +158,18 @@ class Connection(Compat):
         result.close()
         return row
 
+    async def update(self, table_name: str, model: BaseModel, where: str = "id = :id"):
+        result = await self.conn.execute(
+            text(update_query(table_name, model, where)), model.dict()
+        )
+        result.close()
+
+    async def insert(self, table_name: str, model: BaseModel):
+        result = await self.conn.execute(
+            text(insert_query(table_name, model)), model.dict()
+        )
+        result.close()
+
     async def fetch_page(
         self,
         query: str,
@@ -304,6 +316,16 @@ class Database(Compat):
         async with self.connect() as conn:
             return await conn.fetchone(query, values)
 
+    async def insert(self, table_name: str, model: BaseModel) -> None:
+        async with self.connect() as conn:
+            await conn.insert(table_name, model)
+
+    async def update(
+        self, table_name: str, model: BaseModel, where: str = "id = :id"
+    ) -> None:
+        async with self.connect() as conn:
+            await conn.update(table_name, model, where)
+
     async def fetch_page(
         self,
         query: str,
@@ -518,3 +540,34 @@ class Filters(BaseModel, Generic[TFilterModel]):
         if self.search and self.model:
             values["search"] = f"%{self.search}%"
         return values
+
+
+def insert_query(table_name: str, model: BaseModel) -> str:
+    """
+    Generate an insert query with placeholders for a given table and model
+    :param table_name: Name of the table
+    :param model: Pydantic model
+    """
+    placeholders = []
+    for field in model.dict().keys():
+        placeholders.append(get_placeholder(model, field))
+    fields = ", ".join(model.dict().keys())
+    values = ", ".join(placeholders)
+    return f"INSERT INTO {table_name} ({fields}) VALUES ({values})"
+
+
+def update_query(
+    table_name: str, model: BaseModel, where: str = "WHERE id = :id"
+) -> str:
+    """
+    Generate an update query with placeholders for a given table and model
+    :param table_name: Name of the table
+    :param model: Pydantic model
+    :param where: Where string, default to `WHERE id = :id`
+    """
+    fields = []
+    for field in model.dict().keys():
+        placeholder = get_placeholder(model, field)
+        fields.append(f"{field} = {placeholder}")
+    query = ", ".join(fields)
+    return f"UPDATE {table_name} SET {query} {where}"

lnbits/helpers.py

@@ -2,23 +2,24 @@ import json
 import re
 from datetime import datetime, timedelta
 from pathlib import Path
-from typing import Any, List, Optional, Type
+from typing import Any, Optional, Type
 
 import jinja2
 import jwt
 import shortuuid
-from pydantic import BaseModel
 from pydantic.schema import field_schema
 
 from lnbits.core.extensions.models import Extension
-from lnbits.db import get_placeholder
 from lnbits.jinja2_templating import Jinja2Templates
 from lnbits.nodes import get_node_class
 from lnbits.requestvars import g
 from lnbits.settings import settings
 from lnbits.utils.crypto import AESCipher
 
-from .db import FilterModel
+# import insert_query, update_query here is deprecated
+# use shortcut on `Database` class instead
+# example: await db.insert("table_name", base_model)
+from .db import FilterModel, insert_query, update_query  # noqa: F401
 
 
 def get_db_vendor_name():
@@ -51,7 +52,7 @@ def static_url_for(static: str, path: str) -> str:
     return f"/{static}/{path}?v={settings.server_startup_time}"
 
 
-def template_renderer(additional_folders: Optional[List] = None) -> Jinja2Templates:
+def template_renderer(additional_folders: Optional[list] = None) -> Jinja2Templates:
     folders = ["lnbits/templates", "lnbits/core/templates"]
     if additional_folders:
         additional_folders += [
@@ -175,37 +176,6 @@ def generate_filter_params_openapi(model: Type[FilterModel], keep_optional=False
     }
 
 
-def insert_query(table_name: str, model: BaseModel) -> str:
-    """
-    Generate an insert query with placeholders for a given table and model
-    :param table_name: Name of the table
-    :param model: Pydantic model
-    """
-    placeholders = []
-    for field in model.dict().keys():
-        placeholders.append(get_placeholder(model, field))
-    fields = ", ".join(model.dict().keys())
-    values = ", ".join(placeholders)
-    return f"INSERT INTO {table_name} ({fields}) VALUES ({values})"
-
-
-def update_query(
-    table_name: str, model: BaseModel, where: str = "WHERE id = :id"
-) -> str:
-    """
-    Generate an update query with placeholders for a given table and model
-    :param table_name: Name of the table
-    :param model: Pydantic model
-    :param where: Where string, default to `WHERE id = :id`
-    """
-    fields = []
-    for field in model.dict().keys():
-        placeholder = get_placeholder(model, field)
-        fields.append(f"{field} = {placeholder}")
-    query = ", ".join(fields)
-    return f"UPDATE {table_name} SET {query} {where}"
-
-
 def is_valid_email_address(email: str) -> bool:
     email_regex = r"[A-Za-z0-9\._%+-]+@[A-Za-z0-9\.-]+\.[A-Za-z]{2,63}"
     return re.fullmatch(email_regex, email) is not None

tests/unit/test_helpers_query.py

@@ -1,6 +1,6 @@
 import pytest
 
-from lnbits.helpers import (
+from lnbits.db import (
     insert_query,
     update_query,
 )