-
Notifications
You must be signed in to change notification settings - Fork 3
/
crud.py
119 lines (95 loc) · 3.07 KB
/
crud.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
from typing import List, Optional
from lnbits.db import SQLITE, Database
from .models import Item, Shop
from .wordlists import animals
db = Database("ext_offlineshop")
async def create_shop(*, wallet_id: str) -> int:
returning = "" if db.type == SQLITE else "RETURNING ID"
method = db.execute if db.type == SQLITE else db.fetchone
result = await method(
f"""
INSERT INTO offlineshop.shops (wallet, wordlist, method)
VALUES (?, ?, 'wordlist')
{returning}
""",
(wallet_id, "\n".join(animals)),
)
if db.type == SQLITE:
return result._result_proxy.lastrowid
else:
return result[0] # type: ignore
async def get_shop(shop_id: int) -> Optional[Shop]:
row = await db.fetchone("SELECT * FROM offlineshop.shops WHERE id = ?", (shop_id,))
return Shop(**row) if row else None
async def get_or_create_shop_by_wallet(wallet: str) -> Optional[Shop]:
row = await db.fetchone(
"SELECT * FROM offlineshop.shops WHERE wallet = ?", (wallet,)
)
if not row:
# create on the fly
ls_id = await create_shop(wallet_id=wallet)
return await get_shop(ls_id)
return Shop(**row) if row else None
async def set_method(shop: int, method: str, wordlist: str = "") -> Optional[Shop]:
await db.execute(
"UPDATE offlineshop.shops SET method = ?, wordlist = ? WHERE id = ?",
(method, wordlist, shop),
)
return await get_shop(shop)
async def add_item(
shop: int,
name: str,
description: str,
image: Optional[str],
price: int,
unit: str,
fiat_base_multiplier: int,
) -> int:
result = await db.execute(
"""
INSERT INTO offlineshop.items
(shop, name, description, image, price, unit, fiat_base_multiplier)
VALUES (?, ?, ?, ?, ?, ?, ?)
""",
(shop, name, description, image, price, unit, fiat_base_multiplier),
)
return result._result_proxy.lastrowid
async def update_item(
shop: int,
item_id: int,
name: str,
description: str,
image: Optional[str],
price: int,
unit: str,
fiat_base_multiplier: int,
) -> int:
await db.execute(
"""
UPDATE offlineshop.items SET
name = ?,
description = ?,
image = ?,
price = ?,
unit = ?,
fiat_base_multiplier = ?
WHERE shop = ? AND id = ?
""",
(name, description, image, price, unit, fiat_base_multiplier, shop, item_id),
)
return item_id
async def get_item(item_id: int) -> Optional[Item]:
row = await db.fetchone(
"SELECT * FROM offlineshop.items WHERE id = ? LIMIT 1", (item_id,)
)
return Item.from_row(row) if row else None
async def get_items(shop: int) -> List[Item]:
rows = await db.fetchall("SELECT * FROM offlineshop.items WHERE shop = ?", (shop,))
return [Item.from_row(row) for row in rows]
async def delete_item_from_shop(shop: int, item_id: int):
await db.execute(
"""
DELETE FROM offlineshop.items WHERE shop = ? AND id = ?
""",
(shop, item_id),
)