mirror of
https://github.com/lnbits/lnbits-legend.git
synced 2024-11-20 18:51:05 +01:00
120 lines
3.5 KiB
Python
120 lines
3.5 KiB
Python
import json
|
|
|
|
from typing import List, Optional
|
|
from fastapi.params import Depends
|
|
from pydantic.main import BaseModel
|
|
|
|
from http import HTTPStatus
|
|
from lnurl.exceptions import InvalidUrl as LnurlInvalidUrl
|
|
from starlette.exceptions import HTTPException
|
|
from starlette.requests import Request
|
|
from starlette.responses import HTMLResponse, JSONResponse # type: ignore
|
|
|
|
from lnbits.decorators import WalletTypeInfo, get_key_type
|
|
from lnbits.utils.exchange_rates import currencies
|
|
from lnbits.requestvars import g
|
|
|
|
from . import offlineshop_ext
|
|
from .crud import (
|
|
get_or_create_shop_by_wallet,
|
|
set_method,
|
|
add_item,
|
|
update_item,
|
|
get_items,
|
|
delete_item_from_shop,
|
|
)
|
|
from .models import ShopCounter
|
|
|
|
|
|
@offlineshop_ext.get("/api/v1/currencies")
|
|
async def api_list_currencies_available():
|
|
return json.dumps(list(currencies.keys()))
|
|
|
|
|
|
@offlineshop_ext.get("/api/v1/offlineshop")
|
|
# @api_check_wallet_key("invoice")
|
|
async def api_shop_from_wallet(r: Request, wallet: WalletTypeInfo = Depends(get_key_type)):
|
|
shop = await get_or_create_shop_by_wallet(wallet.wallet.id)
|
|
items = await get_items(shop.id)
|
|
|
|
try:
|
|
return {
|
|
**shop.dict(),
|
|
**{
|
|
"otp_key": shop.otp_key,
|
|
"items": [item.values(r) for item in items],
|
|
},
|
|
}
|
|
except LnurlInvalidUrl:
|
|
raise HTTPException(
|
|
status_code=HTTPStatus.UPGRADE_REQUIRED,
|
|
detail="LNURLs need to be delivered over a publically accessible `https` domain or Tor.",
|
|
)
|
|
|
|
|
|
class CreateItemsData(BaseModel):
|
|
name: str
|
|
description: str
|
|
image: Optional[str]
|
|
price: int
|
|
unit: str
|
|
|
|
|
|
@offlineshop_ext.post("/api/v1/offlineshop/items")
|
|
@offlineshop_ext.put("/api/v1/offlineshop/items/{item_id}")
|
|
# @api_check_wallet_key("invoice")
|
|
async def api_add_or_update_item(data: CreateItemsData, item_id=None, wallet: WalletTypeInfo = Depends(get_key_type)):
|
|
shop = await get_or_create_shop_by_wallet(wallet.wallet.id)
|
|
if item_id == None:
|
|
await add_item(
|
|
shop.id,
|
|
data.name,
|
|
data.description,
|
|
data.image,
|
|
data.price,
|
|
data.unit,
|
|
)
|
|
return HTMLResponse(status_code=HTTPStatus.CREATED)
|
|
else:
|
|
await update_item(
|
|
shop.id,
|
|
item_id,
|
|
data.name,
|
|
data.description,
|
|
data.image,
|
|
data.price,
|
|
data.unit,
|
|
)
|
|
|
|
|
|
@offlineshop_ext.delete("/api/v1/offlineshop/items/{item_id}")
|
|
# @api_check_wallet_key("invoice")
|
|
async def api_delete_item(item_id, wallet: WalletTypeInfo = Depends(get_key_type)):
|
|
shop = await get_or_create_shop_by_wallet(wallet.wallet.id)
|
|
await delete_item_from_shop(shop.id, item_id)
|
|
raise HTTPException(status_code=HTTPStatus.NO_CONTENT)
|
|
|
|
|
|
class CreateMethodData(BaseModel):
|
|
method: str
|
|
wordlist: Optional[str]
|
|
|
|
|
|
@offlineshop_ext.put("/api/v1/offlineshop/method")
|
|
# @api_check_wallet_key("invoice")
|
|
async def api_set_method(data: CreateMethodData, wallet: WalletTypeInfo = Depends(get_key_type)):
|
|
method = data.method
|
|
|
|
wordlist = data.wordlist.split("\n") if data.wordlist else None
|
|
wordlist = [word.strip() for word in wordlist if word.strip()]
|
|
|
|
shop = await get_or_create_shop_by_wallet(wallet.wallet.id)
|
|
if not shop:
|
|
raise HTTPException(status_code=HTTPStatus.NOT_FOUND)
|
|
|
|
updated_shop = await set_method(shop.id, method, "\n".join(wordlist))
|
|
if not updated_shop:
|
|
raise HTTPException(status_code=HTTPStatus.NOT_FOUND)
|
|
|
|
ShopCounter.reset(updated_shop)
|