feat: add useful query helpers for insert and update (#2020)

* feat: add useful query helpers for insert and update

i saw this bits of code all over the codebase, this `helpers.py`
tries to abstract that away and make it more readable.
This commit is contained in:
dni ⚡ 2023-10-16 13:27:59 +02:00 committed by GitHub
parent c8661ffcf3
commit dbf687c47a
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
2 changed files with 50 additions and 0 deletions

View file

@ -4,6 +4,7 @@ from typing import Any, List, Optional, Type
import jinja2 import jinja2
import shortuuid import shortuuid
from pydantic import BaseModel
from pydantic.schema import field_schema from pydantic.schema import field_schema
from lnbits.jinja2_templating import Jinja2Templates from lnbits.jinja2_templating import Jinja2Templates
@ -135,3 +136,25 @@ def generate_filter_params_openapi(model: Type[FilterModel], keep_optional=False
return { return {
"parameters": params, "parameters": params,
} }
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 = ", ".join(["?"] * len(model.dict().keys()))
fields = ", ".join(model.dict().keys())
return f"INSERT INTO {table_name} ({fields}) VALUES ({placeholders})"
def update_query(table_name: str, model: BaseModel, where: str = "WHERE 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 = ?`
"""
query = ", ".join([f"{field} = ?" for field in model.dict().keys()])
return f"UPDATE {table_name} SET {query} {where}"

View file

@ -0,0 +1,27 @@
import pytest
from pydantic import BaseModel
from lnbits.helpers import (
insert_query,
update_query,
)
class TestModel(BaseModel):
id: int
name: str
test = TestModel(id=1, name="test")
@pytest.mark.asyncio
async def test_helpers_insert_query():
q = insert_query("test_helpers_query", test)
assert q == "INSERT INTO test_helpers_query (id, name) VALUES (?, ?)"
@pytest.mark.asyncio
async def test_helpers_update_query():
q = update_query("test_helpers_query", test)
assert q == "UPDATE test_helpers_query SET id = ?, name = ? WHERE id = ?"