-
Notifications
You must be signed in to change notification settings - Fork 2
/
crud.py
82 lines (64 loc) · 2.27 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
from typing import List, Optional, Union
from lnbits.db import Database
from lnbits.helpers import urlsafe_short_hash
from .models import CreateScrubLink, ScrubLink
db = Database("ext_scrub")
async def create_scrub_link(data: CreateScrubLink) -> ScrubLink:
scrub_id = urlsafe_short_hash()
await db.execute(
"""
INSERT INTO scrub.scrub_links (
id,
wallet,
description,
payoraddress
)
VALUES (?, ?, ?, ?)
""",
(
scrub_id,
data.wallet,
data.description,
data.payoraddress,
),
)
link = await get_scrub_link(scrub_id)
assert link, "Newly created link couldn't be retrieved"
return link
async def get_scrub_link(link_id: str) -> Optional[ScrubLink]:
row = await db.fetchone("SELECT * FROM scrub.scrub_links WHERE id = ?", (link_id,))
return ScrubLink(**row) if row else None
async def get_scrub_links(wallet_ids: Union[str, List[str]]) -> List[ScrubLink]:
if isinstance(wallet_ids, str):
wallet_ids = [wallet_ids]
q = ",".join(["?"] * len(wallet_ids))
rows = await db.fetchall(
f"""
SELECT * FROM scrub.scrub_links WHERE wallet IN ({q})
ORDER BY id
""",
(*wallet_ids,),
)
return [ScrubLink(**row) for row in rows]
async def update_scrub_link(link_id: int, **kwargs) -> Optional[ScrubLink]:
q = ", ".join([f"{field[0]} = ?" for field in kwargs.items()])
await db.execute(
f"UPDATE scrub.scrub_links SET {q} WHERE id = ?",
(*kwargs.values(), link_id),
)
row = await db.fetchone("SELECT * FROM scrub.scrub_links WHERE id = ?", (link_id,))
return ScrubLink(**row) if row else None
async def delete_scrub_link(link_id: int) -> None:
await db.execute("DELETE FROM scrub.scrub_links WHERE id = ?", (link_id,))
async def get_scrub_by_wallet(wallet_id) -> Optional[ScrubLink]:
row = await db.fetchone(
"SELECT * from scrub.scrub_links WHERE wallet = ?",
(wallet_id,),
)
return ScrubLink(**row) if row else None
async def unique_scrubed_wallet(wallet_id):
(row,) = await db.fetchone(
"SELECT COUNT(wallet) FROM scrub.scrub_links WHERE wallet = ?",
(wallet_id,),
)
return row