lnbits-exts/satmachineclient
1
0
Fork 0
Code Issues Pull requests Projects Releases 1 Packages Wiki Activity Actions

format

Browse source
This commit is contained in:
benarc 2024-02-01 17:18:55 +00:00
parent 66d44f95fb
commit 2cac36be17
12 changed files with 272 additions and 311 deletions
Download patch file Download diff file Expand all files Collapse all files

63
crud.py
View file

@ -9,7 +9,10 @@ from fastapi import Request
from lnurl import encode as lnurl_encode
import shortuuid
async def create_myextension(wallet_id: str, data: CreateMyExtensionData, req: Request) -> MyExtension:
async def create_myextension(
wallet_id: str, data: CreateMyExtensionData, req: Request
) -> MyExtension:
myextension_id = urlsafe_short_hash()
await db.execute(
"""
@ -21,7 +24,7 @@ async def create_myextension(wallet_id: str, data: CreateMyExtensionData, req: R
wallet_id,
data.name,
data.lnurlpayamount,
data.lnurlwithdrawamount
data.lnurlwithdrawamount,
),
)
myextension = await get_myextension(myextension_id, req)
@ -29,25 +32,33 @@ async def create_myextension(wallet_id: str, data: CreateMyExtensionData, req: R
return myextension
async def get_myextension(myextension_id: str, req: Optional[Request] = None) -> Optional[MyExtension]:
async def get_myextension(
myextension_id: str, req: Optional[Request] = None
) -> Optional[MyExtension]:
logger.debug(myextension_id)
row = await db.fetchone("SELECT * FROM myextension.maintable WHERE id = ?", (myextension_id,))
row = await db.fetchone(
"SELECT * FROM myextension.maintable WHERE id = ?", (myextension_id,)
)
if not row:
return None
rowAmended = MyExtension(**row)
if req:
rowAmended.lnurlpay = lnurl_encode(
req.url_for("myextension.api_lnurl_pay",
myextension_id=row.id)._url
)
req.url_for("myextension.api_lnurl_pay", myextension_id=row.id)._url
)
rowAmended.lnurlwithdraw = lnurl_encode(
req.url_for("myextension.api_lnurl_withdraw",
myextension_id=row.id,
tickerhash=shortuuid.uuid(name=rowAmended.id + str(rowAmended.ticker)))._url
)
req.url_for(
"myextension.api_lnurl_withdraw",
myextension_id=row.id,
tickerhash=shortuuid.uuid(name=rowAmended.id + str(rowAmended.ticker)),
)._url
)
return rowAmended
async def get_myextensions(wallet_ids: Union[str, List[str]], req: Optional[Request] = None) -> List[MyExtension]:
async def get_myextensions(
wallet_ids: Union[str, List[str]], req: Optional[Request] = None
) -> List[MyExtension]:
if isinstance(wallet_ids, str):
wallet_ids = [wallet_ids]
@ -59,25 +70,33 @@ async def get_myextensions(wallet_ids: Union[str, List[str]], req: Optional[Requ
if req:
for row in tempRows:
row.lnurlpay = lnurl_encode(
req.url_for("myextension.api_lnurl_pay",
myextension_id=row.id)._url
)
req.url_for("myextension.api_lnurl_pay", myextension_id=row.id)._url
)
row.lnurlwithdraw = lnurl_encode(
req.url_for("myextension.api_lnurl_withdraw",
myextension_id=row.id,
tickerhash=shortuuid.uuid(name=row.id + str(row.ticker)))._url
req.url_for(
"myextension.api_lnurl_withdraw",
myextension_id=row.id,
tickerhash=shortuuid.uuid(name=row.id + str(row.ticker)),
)._url
)
return tempRows
async def update_myextension(myextension_id: str, req: Optional[Request] = None, **kwargs) -> MyExtension:
async def update_myextension(
myextension_id: str, req: Optional[Request] = None, **kwargs
) -> MyExtension:
q = ", ".join([f"{field[0]} = ?" for field in kwargs.items()])
logger.debug( kwargs.items())
logger.debug(kwargs.items())
await db.execute(
f"UPDATE myextension.maintable SET {q} WHERE id = ?", (*kwargs.values(), myextension_id)
f"UPDATE myextension.maintable SET {q} WHERE id = ?",
(*kwargs.values(), myextension_id),
)
myextension = await get_myextension(myextension_id, req)
assert myextension, "Newly updated myextension couldn't be retrieved"
return myextension
async def delete_myextension(myextension_id: str) -> None:
await db.execute("DELETE FROM myextension.maintable WHERE id = ?", (myextension_id,))
await db.execute(
"DELETE FROM myextension.maintable WHERE id = ?", (myextension_id,)
)