mirror of
https://github.com/lnbits/lnbits-legend.git
synced 2025-01-19 05:33:47 +01:00
Merge pull request #1283 from lnbits/fix/mypy-offlineshop
fix mypy offlineshop issues
This commit is contained in:
commit
7e4a3a6831
@ -1,15 +1,9 @@
|
|||||||
import hashlib
|
from fastapi import Query
|
||||||
|
from lnurl import LnurlErrorResponse, LnurlPayActionResponse, LnurlPayResponse
|
||||||
from fastapi.params import Query
|
from lnurl.models import ClearnetUrl, LightningInvoice, MilliSatoshi
|
||||||
from lnurl import ( # type: ignore
|
|
||||||
LnurlErrorResponse,
|
|
||||||
LnurlPayActionResponse,
|
|
||||||
LnurlPayResponse,
|
|
||||||
)
|
|
||||||
from starlette.requests import Request
|
from starlette.requests import Request
|
||||||
|
|
||||||
from lnbits.core.services import create_invoice
|
from lnbits.core.services import create_invoice
|
||||||
from lnbits.extensions.offlineshop.models import Item
|
|
||||||
from lnbits.utils.exchange_rates import fiat_amount_as_satoshis
|
from lnbits.utils.exchange_rates import fiat_amount_as_satoshis
|
||||||
|
|
||||||
from . import offlineshop_ext
|
from . import offlineshop_ext
|
||||||
@ -17,8 +11,8 @@ from .crud import get_item, get_shop
|
|||||||
|
|
||||||
|
|
||||||
@offlineshop_ext.get("/lnurl/{item_id}", name="offlineshop.lnurl_response")
|
@offlineshop_ext.get("/lnurl/{item_id}", name="offlineshop.lnurl_response")
|
||||||
async def lnurl_response(req: Request, item_id: int = Query(...)):
|
async def lnurl_response(req: Request, item_id: int = Query(...)) -> dict:
|
||||||
item = await get_item(item_id) # type: Item
|
item = await get_item(item_id)
|
||||||
if not item:
|
if not item:
|
||||||
return {"status": "ERROR", "reason": "Item not found."}
|
return {"status": "ERROR", "reason": "Item not found."}
|
||||||
|
|
||||||
@ -32,9 +26,11 @@ async def lnurl_response(req: Request, item_id: int = Query(...)):
|
|||||||
) * 1000
|
) * 1000
|
||||||
|
|
||||||
resp = LnurlPayResponse(
|
resp = LnurlPayResponse(
|
||||||
callback=req.url_for("offlineshop.lnurl_callback", item_id=item.id),
|
callback=ClearnetUrl(
|
||||||
min_sendable=price_msat,
|
req.url_for("offlineshop.lnurl_callback", item_id=item.id), scheme="https"
|
||||||
max_sendable=price_msat,
|
),
|
||||||
|
minSendable=MilliSatoshi(price_msat),
|
||||||
|
maxSendable=MilliSatoshi(price_msat),
|
||||||
metadata=await item.lnurlpay_metadata(),
|
metadata=await item.lnurlpay_metadata(),
|
||||||
)
|
)
|
||||||
|
|
||||||
@ -43,7 +39,7 @@ async def lnurl_response(req: Request, item_id: int = Query(...)):
|
|||||||
|
|
||||||
@offlineshop_ext.get("/lnurl/cb/{item_id}", name="offlineshop.lnurl_callback")
|
@offlineshop_ext.get("/lnurl/cb/{item_id}", name="offlineshop.lnurl_callback")
|
||||||
async def lnurl_callback(request: Request, item_id: int):
|
async def lnurl_callback(request: Request, item_id: int):
|
||||||
item = await get_item(item_id) # type: Item
|
item = await get_item(item_id)
|
||||||
if not item:
|
if not item:
|
||||||
return {"status": "ERROR", "reason": "Couldn't find item."}
|
return {"status": "ERROR", "reason": "Couldn't find item."}
|
||||||
|
|
||||||
@ -67,6 +63,7 @@ async def lnurl_callback(request: Request, item_id: int):
|
|||||||
).dict()
|
).dict()
|
||||||
|
|
||||||
shop = await get_shop(item.shop)
|
shop = await get_shop(item.shop)
|
||||||
|
assert shop
|
||||||
|
|
||||||
try:
|
try:
|
||||||
payment_hash, payment_request = await create_invoice(
|
payment_hash, payment_request = await create_invoice(
|
||||||
@ -77,14 +74,15 @@ async def lnurl_callback(request: Request, item_id: int):
|
|||||||
extra={"tag": "offlineshop", "item": item.id},
|
extra={"tag": "offlineshop", "item": item.id},
|
||||||
)
|
)
|
||||||
except Exception as exc:
|
except Exception as exc:
|
||||||
return LnurlErrorResponse(reason=exc.message).dict()
|
return LnurlErrorResponse(reason=str(exc)).dict()
|
||||||
|
|
||||||
resp = LnurlPayActionResponse(
|
if shop.method:
|
||||||
pr=payment_request,
|
success_action = item.success_action(shop, payment_hash, request)
|
||||||
success_action=item.success_action(shop, payment_hash, request)
|
assert success_action
|
||||||
if shop.method
|
resp = LnurlPayActionResponse(
|
||||||
else None,
|
pr=LightningInvoice(payment_request),
|
||||||
routes=[],
|
successAction=success_action,
|
||||||
)
|
routes=[],
|
||||||
|
)
|
||||||
|
|
||||||
return resp.dict()
|
return resp.dict()
|
||||||
|
@ -5,9 +5,9 @@ from collections import OrderedDict
|
|||||||
from sqlite3 import Row
|
from sqlite3 import Row
|
||||||
from typing import Dict, List, Optional
|
from typing import Dict, List, Optional
|
||||||
|
|
||||||
from lnurl import encode as lnurl_encode # type: ignore
|
from lnurl import encode as lnurl_encode
|
||||||
from lnurl.models import LnurlPaySuccessAction, UrlAction # type: ignore
|
from lnurl.models import ClearnetUrl, Max144Str, UrlAction
|
||||||
from lnurl.types import LnurlPayMetadata # type: ignore
|
from lnurl.types import LnurlPayMetadata
|
||||||
from pydantic import BaseModel
|
from pydantic import BaseModel
|
||||||
from starlette.requests import Request
|
from starlette.requests import Request
|
||||||
|
|
||||||
@ -119,11 +119,16 @@ class Item(BaseModel):
|
|||||||
|
|
||||||
def success_action(
|
def success_action(
|
||||||
self, shop: Shop, payment_hash: str, req: Request
|
self, shop: Shop, payment_hash: str, req: Request
|
||||||
) -> Optional[LnurlPaySuccessAction]:
|
) -> Optional[UrlAction]:
|
||||||
if not shop.wordlist:
|
if not shop.wordlist:
|
||||||
return None
|
return None
|
||||||
|
|
||||||
return UrlAction(
|
return UrlAction(
|
||||||
url=req.url_for("offlineshop.confirmation_code", p=payment_hash),
|
url=ClearnetUrl(
|
||||||
description="Open to get the confirmation code for your purchase.",
|
req.url_for("offlineshop.confirmation_code", p=payment_hash),
|
||||||
|
scheme="https",
|
||||||
|
),
|
||||||
|
description=Max144Str(
|
||||||
|
"Open to get the confirmation code for your purchase."
|
||||||
|
),
|
||||||
)
|
)
|
||||||
|
@ -3,8 +3,7 @@ from datetime import datetime
|
|||||||
from http import HTTPStatus
|
from http import HTTPStatus
|
||||||
from typing import List
|
from typing import List
|
||||||
|
|
||||||
from fastapi import HTTPException, Request
|
from fastapi import Depends, HTTPException, Query, Request
|
||||||
from fastapi.params import Depends, Query
|
|
||||||
from starlette.responses import HTMLResponse
|
from starlette.responses import HTMLResponse
|
||||||
|
|
||||||
from lnbits.core.crud import get_standalone_payment
|
from lnbits.core.crud import get_standalone_payment
|
||||||
@ -25,10 +24,10 @@ async def index(request: Request, user: User = Depends(check_user_exists)):
|
|||||||
|
|
||||||
|
|
||||||
@offlineshop_ext.get("/print", response_class=HTMLResponse)
|
@offlineshop_ext.get("/print", response_class=HTMLResponse)
|
||||||
async def print_qr_codes(request: Request, items: List[int] = None):
|
async def print_qr_codes(request: Request):
|
||||||
items = []
|
items = []
|
||||||
for item_id in request.query_params.get("items").split(","):
|
for item_id in request.query_params.get("items").split(","):
|
||||||
item = await get_item(item_id) # type: Item
|
item = await get_item(item_id)
|
||||||
if item:
|
if item:
|
||||||
items.append(
|
items.append(
|
||||||
{
|
{
|
||||||
@ -53,7 +52,8 @@ async def confirmation_code(p: str = Query(...)):
|
|||||||
|
|
||||||
payment_hash = p
|
payment_hash = p
|
||||||
await api_payment(payment_hash)
|
await api_payment(payment_hash)
|
||||||
payment: Payment = await get_standalone_payment(payment_hash)
|
|
||||||
|
payment = await get_standalone_payment(payment_hash)
|
||||||
if not payment:
|
if not payment:
|
||||||
raise HTTPException(
|
raise HTTPException(
|
||||||
status_code=HTTPStatus.NOT_FOUND,
|
status_code=HTTPStatus.NOT_FOUND,
|
||||||
@ -72,8 +72,13 @@ async def confirmation_code(p: str = Query(...)):
|
|||||||
detail="Too much time has passed." + style,
|
detail="Too much time has passed." + style,
|
||||||
)
|
)
|
||||||
|
|
||||||
item = await get_item(payment.extra.get("item"))
|
assert payment.extra
|
||||||
|
item_id = payment.extra.get("item")
|
||||||
|
assert item_id
|
||||||
|
item = await get_item(item_id)
|
||||||
|
assert item
|
||||||
shop = await get_shop(item.shop)
|
shop = await get_shop(item.shop)
|
||||||
|
assert shop
|
||||||
|
|
||||||
return (
|
return (
|
||||||
f"""
|
f"""
|
||||||
|
@ -1,13 +1,9 @@
|
|||||||
from http import HTTPStatus
|
from http import HTTPStatus
|
||||||
from typing import Optional
|
from typing import Optional
|
||||||
|
|
||||||
from fastapi import Query
|
from fastapi import Depends, HTTPException, Query, Request, Response
|
||||||
from fastapi.params import Depends
|
|
||||||
from lnurl.exceptions import InvalidUrl as LnurlInvalidUrl
|
from lnurl.exceptions import InvalidUrl as LnurlInvalidUrl
|
||||||
from pydantic.main import BaseModel
|
from pydantic import BaseModel
|
||||||
from starlette.exceptions import HTTPException
|
|
||||||
from starlette.requests import Request
|
|
||||||
from starlette.responses import HTMLResponse # type: ignore
|
|
||||||
|
|
||||||
from lnbits.decorators import WalletTypeInfo, get_key_type
|
from lnbits.decorators import WalletTypeInfo, get_key_type
|
||||||
from lnbits.utils.exchange_rates import currencies
|
from lnbits.utils.exchange_rates import currencies
|
||||||
@ -34,6 +30,7 @@ async def api_shop_from_wallet(
|
|||||||
r: Request, wallet: WalletTypeInfo = Depends(get_key_type)
|
r: Request, wallet: WalletTypeInfo = Depends(get_key_type)
|
||||||
):
|
):
|
||||||
shop = await get_or_create_shop_by_wallet(wallet.wallet.id)
|
shop = await get_or_create_shop_by_wallet(wallet.wallet.id)
|
||||||
|
assert shop
|
||||||
items = await get_items(shop.id)
|
items = await get_items(shop.id)
|
||||||
try:
|
try:
|
||||||
return {
|
return {
|
||||||
@ -62,6 +59,7 @@ async def api_add_or_update_item(
|
|||||||
data: CreateItemsData, item_id=None, wallet: WalletTypeInfo = Depends(get_key_type)
|
data: CreateItemsData, item_id=None, wallet: WalletTypeInfo = Depends(get_key_type)
|
||||||
):
|
):
|
||||||
shop = await get_or_create_shop_by_wallet(wallet.wallet.id)
|
shop = await get_or_create_shop_by_wallet(wallet.wallet.id)
|
||||||
|
assert shop
|
||||||
if data.unit != "sat":
|
if data.unit != "sat":
|
||||||
data.price = data.price * 100
|
data.price = data.price * 100
|
||||||
if item_id == None:
|
if item_id == None:
|
||||||
@ -71,11 +69,11 @@ async def api_add_or_update_item(
|
|||||||
data.name,
|
data.name,
|
||||||
data.description,
|
data.description,
|
||||||
data.image,
|
data.image,
|
||||||
data.price,
|
int(data.price),
|
||||||
data.unit,
|
data.unit,
|
||||||
data.fiat_base_multiplier,
|
data.fiat_base_multiplier,
|
||||||
)
|
)
|
||||||
return HTMLResponse(status_code=HTTPStatus.CREATED)
|
return Response(status_code=HTTPStatus.CREATED)
|
||||||
else:
|
else:
|
||||||
await update_item(
|
await update_item(
|
||||||
shop.id,
|
shop.id,
|
||||||
@ -83,7 +81,7 @@ async def api_add_or_update_item(
|
|||||||
data.name,
|
data.name,
|
||||||
data.description,
|
data.description,
|
||||||
data.image,
|
data.image,
|
||||||
data.price,
|
int(data.price),
|
||||||
data.unit,
|
data.unit,
|
||||||
data.fiat_base_multiplier,
|
data.fiat_base_multiplier,
|
||||||
)
|
)
|
||||||
@ -92,6 +90,7 @@ async def api_add_or_update_item(
|
|||||||
@offlineshop_ext.delete("/api/v1/offlineshop/items/{item_id}")
|
@offlineshop_ext.delete("/api/v1/offlineshop/items/{item_id}")
|
||||||
async def api_delete_item(item_id, wallet: WalletTypeInfo = Depends(get_key_type)):
|
async def api_delete_item(item_id, wallet: WalletTypeInfo = Depends(get_key_type)):
|
||||||
shop = await get_or_create_shop_by_wallet(wallet.wallet.id)
|
shop = await get_or_create_shop_by_wallet(wallet.wallet.id)
|
||||||
|
assert shop
|
||||||
await delete_item_from_shop(shop.id, item_id)
|
await delete_item_from_shop(shop.id, item_id)
|
||||||
return "", HTTPStatus.NO_CONTENT
|
return "", HTTPStatus.NO_CONTENT
|
||||||
|
|
||||||
@ -107,7 +106,7 @@ async def api_set_method(
|
|||||||
):
|
):
|
||||||
method = data.method
|
method = data.method
|
||||||
|
|
||||||
wordlist = data.wordlist.split("\n") if data.wordlist else None
|
wordlist = data.wordlist.split("\n") if data.wordlist else []
|
||||||
wordlist = [word.strip() for word in wordlist if word.strip()]
|
wordlist = [word.strip() for word in wordlist if word.strip()]
|
||||||
|
|
||||||
shop = await get_or_create_shop_by_wallet(wallet.wallet.id)
|
shop = await get_or_create_shop_by_wallet(wallet.wallet.id)
|
||||||
|
@ -96,7 +96,6 @@ exclude = """(?x)(
|
|||||||
| ^lnbits/extensions/livestream.
|
| ^lnbits/extensions/livestream.
|
||||||
| ^lnbits/extensions/lnaddress.
|
| ^lnbits/extensions/lnaddress.
|
||||||
| ^lnbits/extensions/lnurldevice.
|
| ^lnbits/extensions/lnurldevice.
|
||||||
| ^lnbits/extensions/offlineshop.
|
|
||||||
| ^lnbits/extensions/satspay.
|
| ^lnbits/extensions/satspay.
|
||||||
| ^lnbits/extensions/streamalerts.
|
| ^lnbits/extensions/streamalerts.
|
||||||
| ^lnbits/extensions/watchonly.
|
| ^lnbits/extensions/watchonly.
|
||||||
|
Loading…
Reference in New Issue
Block a user