feat: update to lnbits 1.0.0 (#41)

---------

Co-authored-by: Pavol Rusnak <[email protected]>

config.json

@@ -2,7 +2,7 @@
   "name": "Bolt Cards",
   "short_description": "Self custody Bolt Cards with one time LNURLw",
   "tile": "/boltcards/static/image/boltcard.png",
-  "min_lnbits_version": "0.12.5",
+  "min_lnbits_version": "1.0.0",
   "contributors": [
     {
       "name": "dni",

crud.py

@@ -1,6 +1,6 @@
 import secrets
 from datetime import datetime
-from typing import List, Optional
+from typing import Optional
 
 from lnbits.db import Database
 from lnbits.helpers import urlsafe_short_hash
@@ -31,175 +31,157 @@ async def create_card(data: CreateCardData, wallet_id: str) -> Card:
             k2,
             otp
         )
-        VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
+        VALUES (
+            :id, :uid, :external_id, :wallet, :card_name, :counter,
+            :tx_limit, :daily_limit, :enable, :k0, :k1, :k2, :otp
+        )
         """,
-        (
-            card_id,
-            data.uid.upper(),
-            extenal_id,
-            wallet_id,
-            data.card_name,
-            data.counter,
-            data.tx_limit,
-            data.daily_limit,
-            True,
-            data.k0,
-            data.k1,
-            data.k2,
-            secrets.token_hex(16),
-        ),
+        {
+            "id": card_id,
+            "uid": data.uid.upper(),
+            "external_id": extenal_id,
+            "wallet": wallet_id,
+            "card_name": data.card_name,
+            "counter": data.counter,
+            "tx_limit": data.tx_limit,
+            "daily_limit": data.daily_limit,
+            "enable": True,
+            "k0": data.k0,
+            "k1": data.k1,
+            "k2": data.k2,
+            "otp": secrets.token_hex(16),
+        },
     )
     card = await get_card(card_id)
     assert card, "Newly created card couldn't be retrieved"
     return card
 
 
-async def update_card(card_id: str, **kwargs) -> Optional[Card]:
-    if "is_unique" in kwargs:
-        kwargs["is_unique"] = int(kwargs["is_unique"])
-    if "uid" in kwargs:
-        kwargs["uid"] = kwargs["uid"].upper()
-    q = ", ".join([f"{field[0]} = ?" for field in kwargs.items()])
-    await db.execute(
-        f"UPDATE boltcards.cards SET {q} WHERE id = ?",
-        (*kwargs.values(), card_id),
+async def update_card(card_id: str, data: CreateCardData) -> Card:
+    card = Card(
+        id=card_id,
+        **data.dict(),
     )
-    row = await db.fetchone("SELECT * FROM boltcards.cards WHERE id = ?", (card_id,))
-    return Card(**row) if row else None
+    await db.update("boltcards.cards", card)
+    return card
 
 
-async def get_cards(wallet_ids: List[str]) -> List[Card]:
+async def get_cards(wallet_ids: list[str]) -> list[Card]:
     if len(wallet_ids) == 0:
         return []
-
-    q = ",".join(["?"] * len(wallet_ids))
-    rows = await db.fetchall(
-        f"SELECT * FROM boltcards.cards WHERE wallet IN ({q})", (*wallet_ids,)
+    q = ",".join([f"'{wallet_id}'" for wallet_id in wallet_ids])
+    return await db.fetchall(
+        f"SELECT * FROM boltcards.cards WHERE wallet IN ({q})",
+        model=Card,
     )
 
-    return [Card(**row) for row in rows]
-
 
 async def get_card(card_id: str) -> Optional[Card]:
-    row = await db.fetchone("SELECT * FROM boltcards.cards WHERE id = ?", (card_id,))
-    if not row:
-        return None
-
-    card = dict(**row)
-
-    return Card.parse_obj(card)
+    return await db.fetchone(
+        "SELECT * FROM boltcards.cards WHERE id = :id",
+        {"id": card_id},
+        Card,
+    )
 
 
 async def get_card_by_uid(card_uid: str) -> Optional[Card]:
-    row = await db.fetchone(
-        "SELECT * FROM boltcards.cards WHERE uid = ?", (card_uid.upper(),)
+    return await db.fetchone(
+        "SELECT * FROM boltcards.cards WHERE uid = :uid",
+        {"uid": card_uid.upper()},
+        Card,
     )
-    if not row:
-        return None
-
-    card = dict(**row)
-
-    return Card.parse_obj(card)
 
 
 async def get_card_by_external_id(external_id: str) -> Optional[Card]:
-    row = await db.fetchone(
-        "SELECT * FROM boltcards.cards WHERE external_id = ?", (external_id.lower(),)
+    return await db.fetchone(
+        "SELECT * FROM boltcards.cards WHERE external_id = :ext_id",
+        {"ext_id": external_id.lower()},
+        Card,
     )
-    if not row:
-        return None
-
-    card = dict(**row)
-
-    return Card.parse_obj(card)
 
 
 async def get_card_by_otp(otp: str) -> Optional[Card]:
-    row = await db.fetchone("SELECT * FROM boltcards.cards WHERE otp = ?", (otp,))
-    if not row:
-        return None
-
-    card = dict(**row)
-
-    return Card.parse_obj(card)
+    return await db.fetchone(
+        "SELECT * FROM boltcards.cards WHERE otp = :otp",
+        {"otp": otp},
+        Card,
+    )
 
 
 async def delete_card(card_id: str) -> None:
     # Delete cards
-    await db.execute("DELETE FROM boltcards.cards WHERE id = ?", (card_id,))
+    await db.execute("DELETE FROM boltcards.cards WHERE id = :id", {"id": card_id})
     # Delete hits
     hits = await get_hits([card_id])
     for hit in hits:
-        await db.execute("DELETE FROM boltcards.hits WHERE id = ?", (hit.id,))
+        await db.execute("DELETE FROM boltcards.hits WHERE id = :id", {"id": hit.id})
         # Delete refunds
         refunds = await get_refunds([hit.id])
         for refund in refunds:
             await db.execute(
-                "DELETE FROM boltcards.refunds WHERE id = ?", (refund.hit_id,)
+                "DELETE FROM boltcards.refunds WHERE id = :id", {"id": refund.id}
             )
 
 
 async def update_card_counter(counter: int, card_id: str):
     await db.execute(
-        "UPDATE boltcards.cards SET counter = ? WHERE id = ?",
-        (counter, card_id),
+        "UPDATE boltcards.cards SET counter = :counter WHERE id = :id",
+        {"counter": counter, "id": card_id},
     )
 
 
 async def enable_disable_card(enable: bool, card_id: str) -> Optional[Card]:
     await db.execute(
-        "UPDATE boltcards.cards SET enable = ? WHERE id = ?",
-        (enable, card_id),
+        "UPDATE boltcards.cards SET enable = :enable WHERE id = :id",
+        {"enable": enable, "id": card_id},
     )
     return await get_card(card_id)
 
 
 async def update_card_otp(otp: str, card_id: str):
     await db.execute(
-        "UPDATE boltcards.cards SET otp = ? WHERE id = ?",
-        (otp, card_id),
+        "UPDATE boltcards.cards SET otp = :otp WHERE id = :id",
+        {"otp": otp, "id": card_id},
     )
 
 
 async def get_hit(hit_id: str) -> Optional[Hit]:
-    row = await db.fetchone("SELECT * FROM boltcards.hits WHERE id = ?", (hit_id,))
-    if not row:
-        return None
-
-    hit = dict(**row)
-
-    return Hit.parse_obj(hit)
+    return await db.fetchone(
+        "SELECT * FROM boltcards.hits WHERE id = :id",
+        {"id": hit_id},
+        Hit,
+    )
 
 
-async def get_hits(cards_ids: List[str]) -> List[Hit]:
+async def get_hits(cards_ids: list[str]) -> list[Hit]:
     if len(cards_ids) == 0:
         return []
 
-    q = ",".join(["?"] * len(cards_ids))
-    rows = await db.fetchall(
-        f"SELECT * FROM boltcards.hits WHERE card_id IN ({q})", (*cards_ids,)
+    q = ",".join([f"'{card_id}'" for card_id in cards_ids])
+    return await db.fetchall(
+        f"SELECT * FROM boltcards.hits WHERE card_id IN ({q})",
+        model=Hit,
     )
 
-    return [Hit(**row) for row in rows]
-
 
-async def get_hits_today(card_id: str) -> List[Hit]:
+async def get_hits_today(card_id: str) -> list[Hit]:
     rows = await db.fetchall(
-        "SELECT * FROM boltcards.hits WHERE card_id = ?",
-        (card_id,),
+        "SELECT * FROM boltcards.hits WHERE card_id = :id",
+        {"id": card_id},
+        Hit,
     )
     updatedrow = []
-    for row in rows:
-        if datetime.now().date() == datetime.fromtimestamp(row.time).date():
-            updatedrow.append(row)
+    for hit in rows:
+        if datetime.now().date() == hit.time.date():
+            updatedrow.append(hit)
 
-    return [Hit(**row) for row in updatedrow]
+    return updatedrow
 
 
 async def spend_hit(card_id: str, amount: int):
     await db.execute(
-        "UPDATE boltcards.hits SET spent = ?, amount = ? WHERE id = ?",
-        (True, amount, card_id),
+        "UPDATE boltcards.hits SET spent = :spent, amount = :amount WHERE id = :id",
+        {"spent": True, "amount": amount, "id": card_id},
     )
     return await get_hit(card_id)
 
@@ -218,18 +200,18 @@ async def create_hit(card_id, ip, useragent, old_ctr, new_ctr) -> Hit:
             new_ctr,
             amount
         )
-        VALUES (?, ?, ?, ?, ?, ?, ?, ?)
+        VALUES (:id, :card_id, :ip, :spent, :useragent, :old_ctr, :new_ctr, :amount)
         """,
-        (
-            hit_id,
-            card_id,
-            ip,
-            False,
-            useragent,
-            old_ctr,
-            new_ctr,
-            0,
-        ),
+        {
+            "id": hit_id,
+            "card_id": card_id,
+            "ip": ip,
+            "spent": False,
+            "useragent": useragent,
+            "old_ctr": old_ctr,
+            "new_ctr": new_ctr,
+            "amount": 0,
+        },
     )
     hit = await get_hit(hit_id)
     assert hit, "Newly recorded hit couldn't be retrieved"
@@ -245,36 +227,32 @@ async def create_refund(hit_id, refund_amount) -> Refund:
             hit_id,
             refund_amount
         )
-        VALUES (?, ?, ?)
+        VALUES (:id, :hit_id, :refund_amount)
         """,
-        (
-            refund_id,
-            hit_id,
-            refund_amount,
-        ),
+        {
+            "id": refund_id,
+            "hit_id": hit_id,
+            "refund_amount": refund_amount,
+        },
     )
     refund = await get_refund(refund_id)
     assert refund, "Newly recorded hit couldn't be retrieved"
     return refund
 
 
 async def get_refund(refund_id: str) -> Optional[Refund]:
-    row = await db.fetchone(
-        "SELECT * FROM boltcards.refunds WHERE id = ?", (refund_id,)
+    return await db.fetchone(
+        "SELECT * FROM boltcards.refunds WHERE id = :id",
+        {"id": refund_id},
+        Refund,
     )
-    if not row:
-        return None
-    refund = dict(**row)
-    return Refund.parse_obj(refund)
 
 
-async def get_refunds(hits_ids: List[str]) -> List[Refund]:
+async def get_refunds(hits_ids: list[str]) -> list[Refund]:
     if len(hits_ids) == 0:
         return []
-
-    q = ",".join(["?"] * len(hits_ids))
-    rows = await db.fetchall(
-        f"SELECT * FROM boltcards.refunds WHERE hit_id IN ({q})", (*hits_ids,)
+    q = ",".join([f"'{hit_id}'" for hit_id in hits_ids])
+    return await db.fetchall(
+        f"SELECT * FROM boltcards.refunds WHERE hit_id IN ({q})",
+        model=Refund,
     )
-
-    return [Refund(**row) for row in rows]