heierbtc/lnbits
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

bleskomat with import error in views_api

Browse Source
This commit is contained in:
Tiago vasconcelos 2021-10-26 22:50:29 +01:00
parent 4653e81695
commit 92ba2db276
8 changed files with 136 additions and 136 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

14
lnbits/extensions/bleskomat/__init__.py
View File

@ -1,12 +1,18 @@
from quart import Blueprint
from fastapi import APIRouter
from lnbits.db import Database
from lnbits.helpers import template_renderer
db = Database("ext_bleskomat")
bleskomat_ext: Blueprint = Blueprint(
"bleskomat", __name__, static_folder="static", template_folder="templates"
bleskomat_ext: APIRouter = APIRouter(
prefix="/bleskomat",
tags=["Bleskomat"]
)
def bleskomat_renderer():
return template_renderer(["lnbits/extensions/events/templates"])
from .lnurl_api import * # noqa
from .views_api import * # noqa
from .views import * # noqa
from .views_api import * # noqa

19
lnbits/extensions/bleskomat/crud.py
View File

@ -1,19 +1,16 @@
import secrets
import time
from uuid import uuid4
from typing import List, Optional, Union
from uuid import uuid4
from . import db
from .models import Bleskomat, BleskomatLnurl
from .helpers import generate_bleskomat_lnurl_hash
from .models import Bleskomat, BleskomatLnurl, CreateBleskomat
async def create_bleskomat(
*,
data: CreateBleskomat,
wallet_id: str,
name: str,
fiat_currency: str,
exchange_rate_provider: str,
fee: str,
) -> Bleskomat:
bleskomat_id = uuid4().hex
api_key_id = secrets.token_hex(8)
@ -30,10 +27,10 @@ async def create_bleskomat(
api_key_id,
api_key_secret,
api_key_encoding,
name,
fiat_currency,
exchange_rate_provider,
fee,
data.name,
data.fiat_currency,
data.exchange_rate_provider,
data.fee,
),
)
bleskomat = await get_bleskomat(bleskomat_id)

13
lnbits/extensions/bleskomat/helpers.py
View File

@ -1,11 +1,12 @@
import base64
import hashlib
import hmac
from http import HTTPStatus
from binascii import unhexlify
from typing import Dict
from quart import url_for
import urllib
from binascii import unhexlify
from http import HTTPStatus
from typing import Dict
from starlette.requests import Request
def generate_bleskomat_lnurl_hash(secret: str):
@ -34,8 +35,8 @@ def generate_bleskomat_lnurl_secret(api_key_id: str, signature: str):
return m.hexdigest()
def get_callback_url():
return url_for("bleskomat.api_bleskomat_lnurl", _external=True)
def get_callback_url(request: Request):
return request.url_for("bleskomat.api_bleskomat_lnurl", _external=True)
def is_supported_lnurl_subprotocol(tag: str) -> bool:

32
lnbits/extensions/bleskomat/lnurl_api.py
View File

@ -1,8 +1,9 @@
import json
import math
from quart import jsonify, request
from http import HTTPStatus
import traceback
from http import HTTPStatus
from starlette.requests import Request
from . import bleskomat_ext
from .crud import (
@ -10,16 +11,12 @@ from .crud import (
get_bleskomat_by_api_key_id,
get_bleskomat_lnurl,
)
from .exchange_rates import (
fetch_fiat_exchange_rate,
)
from .exchange_rates import fetch_fiat_exchange_rate
from .helpers import (
generate_bleskomat_lnurl_signature,
generate_bleskomat_lnurl_secret,
LnurlHttpError,
LnurlValidationError,
generate_bleskomat_lnurl_secret,
generate_bleskomat_lnurl_signature,
prepare_lnurl_params,
query_to_signing_payload,
unshorten_lnurl_query,
@ -27,10 +24,10 @@ from .helpers import (
# Handles signed URL from Bleskomat ATMs and "action" callback of auto-generated LNURLs.
@bleskomat_ext.route("/u", methods=["GET"])
async def api_bleskomat_lnurl():
@bleskomat_ext.get("/u", name="bleskomat.api_bleskomat_lnurl")
async def api_bleskomat_lnurl(request: Request):
try:
query = request.args.to_dict()
query = request.query_params
# Unshorten query if "s" is used instead of "signature".
if "s" in query:
@ -99,7 +96,7 @@ async def api_bleskomat_lnurl():
)
# Reply with LNURL response object.
return jsonify(lnurl.get_info_response_object(secret)), HTTPStatus.OK
return lnurl.get_info_response_object(secret)
# No signature provided.
# Treat as "action" callback.
@ -123,12 +120,9 @@ async def api_bleskomat_lnurl():
raise LnurlHttpError(str(e), HTTPStatus.BAD_REQUEST)
except LnurlHttpError as e:
return jsonify({"status": "ERROR", "reason": str(e)}), e.http_status
return {"status": "ERROR", "reason": str(e)}
except Exception:
traceback.print_exc()
return (
jsonify({"status": "ERROR", "reason": "Unexpected error"}),
HTTPStatus.INTERNAL_SERVER_ERROR,
)
return {"status": "ERROR", "reason": "Unexpected error"}
return jsonify({"status": "OK"}), HTTPStatus.OK
return {"status": "OK"}

43
lnbits/extensions/bleskomat/models.py
View File

@ -1,13 +1,44 @@
import json
import time
from typing import NamedTuple, Dict
from typing import Dict
from fastapi.params import Query
from pydantic import BaseModel, validator
from starlette.requests import Request
from lnbits import bolt11
from lnbits.core.services import pay_invoice
from . import db
from .helpers import get_callback_url, LnurlValidationError
from .exchange_rates import exchange_rate_providers, fiat_currencies
from .helpers import LnurlValidationError, get_callback_url
class Bleskomat(NamedTuple):
class CreateBleskomat(BaseModel):
name: str = Query(...)
fiat_currency: str = Query(...)
exchange_rate_provider: str = Query(...)
fee: str = Query(...)
@validator('fiat_currency')
def allowed_fiat_currencies(cls, v):
if(v not in fiat_currencies.keys()):
raise ValueError('Not allowed currency')
return v
@validator('exchange_rate_provider')
def allowed_providers(cls, v):
if(v not in exchange_rate_providers.keys()):
raise ValueError('Not allowed provider')
return v
@validator('fee')
def fee_type(cls, v):
if(not isinstance(v, (str, float, int))):
raise ValueError('Fee type not allowed')
return v
class Bleskomat(BaseModel):
id: str
wallet: str
api_key_id: str
@ -19,7 +50,7 @@ class Bleskomat(NamedTuple):
fee: str
class BleskomatLnurl(NamedTuple):
class BleskomatLnurl(BaseModel):
id: str
bleskomat: str
wallet: str
@ -36,14 +67,14 @@ class BleskomatLnurl(NamedTuple):
# When initial uses is 0 then the LNURL has unlimited uses.
return self.initial_uses == 0 or self.remaining_uses > 0
def get_info_response_object(self, secret: str) -> Dict[str, str]:
def get_info_response_object(self, secret: str, req: Request) -> Dict[str, str]:
tag = self.tag
params = json.loads(self.params)
response = {"tag": tag}
if tag == "withdrawRequest":
for key in ["minWithdrawable", "maxWithdrawable", "defaultDescription"]:
response[key] = params[key]
response["callback"] = get_callback_url()
response["callback"] = get_callback_url(req)
response["k1"] = secret
return response

2
lnbits/extensions/bleskomat/static/js/index.js
View File

@ -84,7 +84,7 @@ new Vue({
LNbits.api
.request(
'GET',
'/bleskomat/api/v1/bleskomats?all_wallets',
'/bleskomat/api/v1/bleskomats?all_wallets=True',
this.g.user.wallets[0].adminkey
)
.then(function (response) {

25
lnbits/extensions/bleskomat/views.py
View File

@ -1,22 +1,27 @@
from quart import g, render_template
from lnbits.decorators import check_user_exists, validate_uuids
from http import HTTPStatus
from . import bleskomat_ext
from fastapi import Request
from fastapi.params import Depends
from fastapi.templating import Jinja2Templates
from starlette.exceptions import HTTPException
from starlette.responses import HTMLResponse
from lnbits.core.models import User
from lnbits.decorators import check_user_exists
from . import bleskomat_ext, bleskomat_renderer
from .exchange_rates import exchange_rate_providers_serializable, fiat_currencies
from .helpers import get_callback_url
@bleskomat_ext.route("/")
@validate_uuids(["usr"], required=True)
@check_user_exists()
async def index():
@bleskomat_ext.get("/", response_class=HTMLResponse)
async def index(request: Request, user: User = Depends(check_user_exists)):
bleskomat_vars = {
"callback_url": get_callback_url(),
"callback_url": get_callback_url(request),
"exchange_rate_providers": exchange_rate_providers_serializable,
"fiat_currencies": fiat_currencies,
}
return await render_template(
"bleskomat/index.html", user=g.user, bleskomat_vars=bleskomat_vars
return bleskomat_renderer().TemplateResponse(
"bleskomat/index.html", {"request": request, "user": user.dict(), "bleskomat_vars": bleskomat_vars}
)

124
lnbits/extensions/bleskomat/views_api.py
View File

@ -1,120 +1,86 @@
from quart import g, jsonify, request
from http import HTTPStatus
from fastapi.params import Query
from starlette.exceptions import HTTPException
from lnbits.core.crud import get_user
from lnbits.decorators import api_check_wallet_key, api_validate_post_request
from lnbits.decorators import WalletTypeInfo, require_admin_key
from lnbits.extensions.bleskomat.models import CreateBleskomat
from . import bleskomat_ext
from .crud import (
create_bleskomat,
delete_bleskomat,
get_bleskomat,
get_bleskomats,
update_bleskomat,
delete_bleskomat,
)
from .exchange_rates import (
exchange_rate_providers,
fetch_fiat_exchange_rate,
fiat_currencies,
)
from .exchange_rates import fetch_fiat_exchange_rate
@bleskomat_ext.route("/api/v1/bleskomats", methods=["GET"])
@api_check_wallet_key("admin")
async def api_bleskomats():
wallet_ids = [g.wallet.id]
@bleskomat_ext.get("/api/v1/bleskomats")
async def api_bleskomats(wallet: WalletTypeInfo(require_admin_key), all_wallets: bool = Query(False)):
wallet_ids = [wallet.wallet.id]
if "all_wallets" in request.args:
wallet_ids = (await get_user(g.wallet.user)).wallet_ids
if all_wallets:
wallet_ids = (await get_user(wallet.wallet.user)).wallet_ids
return (
jsonify(
[bleskomat._asdict() for bleskomat in await get_bleskomats(wallet_ids)]
),
HTTPStatus.OK,
)
return [bleskomat.dict() for bleskomat in await get_bleskomats(wallet_ids)]
@bleskomat_ext.route("/api/v1/bleskomat/<bleskomat_id>", methods=["GET"])
@api_check_wallet_key("admin")
async def api_bleskomat_retrieve(bleskomat_id):
@bleskomat_ext.get("/api/v1/bleskomat/{bleskomat_id}")
async def api_bleskomat_retrieve(wallet: WalletTypeInfo(require_admin_key), bleskomat_id):
bleskomat = await get_bleskomat(bleskomat_id)
if not bleskomat or bleskomat.wallet != g.wallet.id:
return (
jsonify({"message": "Bleskomat configuration not found."}),
HTTPStatus.NOT_FOUND,
)
if not bleskomat or bleskomat.wallet != wallet.wallet.id:
raise HTTPException(
status_code=HTTPStatus.NOT_FOUND,
detail="Bleskomat configuration not found."
)
return jsonify(bleskomat._asdict()), HTTPStatus.OK
return bleskomat.dict()
@bleskomat_ext.route("/api/v1/bleskomat", methods=["POST"])
@bleskomat_ext.route("/api/v1/bleskomat/<bleskomat_id>", methods=["PUT"])
@api_check_wallet_key("admin")
@api_validate_post_request(
schema={
"name": {"type": "string", "empty": False, "required": True},
"fiat_currency": {
"type": "string",
"allowed": fiat_currencies.keys(),
"required": True,
},
"exchange_rate_provider": {
"type": "string",
"allowed": exchange_rate_providers.keys(),
"required": True,
},
"fee": {"type": ["string", "float", "number", "integer"], "required": True},
}
)
async def api_bleskomat_create_or_update(bleskomat_id=None):
@bleskomat_ext.post("/api/v1/bleskomat")
@bleskomat_ext.put("/api/v1/bleskomat/{bleskomat_id}",)
async def api_bleskomat_create_or_update(data: CreateBleskomat, wallet: WalletTypeInfo(require_admin_key), bleskomat_id=None):
try:
fiat_currency = g.data["fiat_currency"]
exchange_rate_provider = g.data["exchange_rate_provider"]
fiat_currency = data.fiat_currency
exchange_rate_provider = data.exchange_rate_provider
await fetch_fiat_exchange_rate(
currency=fiat_currency, provider=exchange_rate_provider
)
except Exception as e:
print(e)
return (
jsonify(
{
"message": f'Failed to fetch BTC/{fiat_currency} currency pair from "{exchange_rate_provider}"'
}
),
HTTPStatus.INTERNAL_SERVER_ERROR,
raise HTTPException(
status_code=HTTPStatus.INTERNAL_SERVER_ERROR,
detail=f'Failed to fetch BTC/{fiat_currency} currency pair from "{exchange_rate_provider}"'
)
if bleskomat_id:
bleskomat = await get_bleskomat(bleskomat_id)
if not bleskomat or bleskomat.wallet != g.wallet.id:
return (
jsonify({"message": "Bleskomat configuration not found."}),
HTTPStatus.NOT_FOUND,
if not bleskomat or bleskomat.wallet != wallet.wallet.id:
raise HTTPException(
status_code=HTTPStatus.NOT_FOUND,
detail="Bleskomat configuration not found."
)
bleskomat = await update_bleskomat(bleskomat_id, **g.data)
bleskomat = await update_bleskomat(bleskomat_id, **data.dict())
else:
bleskomat = await create_bleskomat(wallet_id=g.wallet.id, **g.data)
bleskomat = await create_bleskomat(wallet_id=wallet.wallet.id, data=data)
return (
jsonify(bleskomat._asdict()),
HTTPStatus.OK if bleskomat_id else HTTPStatus.CREATED,
)
return bleskomat.dict()
@bleskomat_ext.route("/api/v1/bleskomat/<bleskomat_id>", methods=["DELETE"])
@api_check_wallet_key("admin")
async def api_bleskomat_delete(bleskomat_id):
@bleskomat_ext.delete("/api/v1/bleskomat/{bleskomat_id}")
async def api_bleskomat_delete(wallet: WalletTypeInfo(require_admin_key), bleskomat_id):
bleskomat = await get_bleskomat(bleskomat_id)
if not bleskomat or bleskomat.wallet != g.wallet.id:
return (
jsonify({"message": "Bleskomat configuration not found."}),
HTTPStatus.NOT_FOUND,
)
if not bleskomat or bleskomat.wallet != wallet.wallet.id:
raise HTTPException(
status_code=HTTPStatus.NOT_FOUND,
detail="Bleskomat configuration not found."
)
await delete_bleskomat(bleskomat_id)
return "", HTTPStatus.NO_CONTENT
raise HTTPException(status_code=HTTPStatus.NO_CONTENT)