new extension

just proof of concept

lnbits/extensions/boltcards/crud.py

@@ -0,0 +1,91 @@
+from optparse import Option
+from typing import List, Optional, Union
+from lnbits.helpers import urlsafe_short_hash
+
+from . import db
+from .models import Card, CreateCardData
+
+async def create_card(
+    data: CreateCardData, wallet_id: str
+) -> Card:
+    card_id = urlsafe_short_hash()
+    await db.execute(
+        """
+        INSERT INTO boltcards.cards (
+            id,
+            wallet,
+            card_name,
+            uid,
+            counter,
+            withdraw,
+            file_key,
+            meta_key
+        )
+        VALUES (?, ?, ?, ?, ?, ?, ?, ?)
+        """,
+        (
+            card_id,
+            wallet_id,
+            data.name,
+            data.uid,
+            data.counter,
+            data.withdraw,
+            data.file_key,
+            data.meta_key,
+        ),
+    )
+    link = await get_card(card_id, 0)
+    assert link, "Newly created card couldn't be retrieved"
+    return link
+
+async def update_card(card_id: str, **kwargs) -> Optional[Card]:
+    if "is_unique" in kwargs:
+        kwargs["is_unique"] = int(kwargs["is_unique"])
+    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),
+    )
+    row = await db.fetchone(
+        "SELECT * FROM boltcards.cards WHERE id = ?", (card_id,)
+    )
+    return Card(**row) if row else None
+
+async def get_cards(wallet_ids: Union[str, List[str]]) -> List[Card]:
+    if isinstance(wallet_ids, str):
+        wallet_ids = [wallet_ids]
+
+    q = ",".join(["?"] * len(wallet_ids))
+    rows = await db.fetchall(
+        f"SELECT * FROM boltcards.cards WHERE wallet IN ({q})", (*wallet_ids,)
+    )
+
+    return [Card(**row) for row in rows]
+
+async def get_all_cards() -> List[Card]:
+    rows = await db.fetchall(
+        f"SELECT * FROM boltcards.cards"
+    )
+
+    return [Card(**row) for row in rows]
+
+async def get_card(card_id: str, id_is_uid: bool=False) -> Optional[Card]:
+    sql = "SELECT * FROM boltcards.cards WHERE {} = ?".format("uid" if id_is_uid else "id")
+    row = await db.fetchone(
+        sql, card_id,
+    )
+    if not row:
+        return None
+
+    card = dict(**row)
+
+    return Card.parse_obj(card)
+
+async def delete_card(card_id: str) -> None:
+    await db.execute("DELETE FROM boltcards.cards WHERE id = ?", (card_id,))
+
+async def update_card_counter(counter: int, id: str):
+    await db.execute(
+        "UPDATE boltcards.cards SET counter = ? WHERE id = ?",
+        (counter, id),
+    )