mirrors/lnbits-legend
1
0
Fork 0
mirror of https://github.com/lnbits/lnbits-legend.git synced 2024-11-20 02:28:10 +01:00
Code Issues Packages Projects Releases Wiki Activity

convert to FastAPI

Browse Source
This commit is contained in:
Tiago vasconcelos 2022-03-12 14:18:09 +00:00 committed by dni ⚡
parent 68eee00b45
commit 65e1f19ed1
7 changed files with 151 additions and 107 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

11
lnbits/extensions/admin/__init__.py
View File

@ -1,10 +1,15 @@
from quart import Blueprint
from fastapi import APIRouter
from lnbits.db import Database
from lnbits.helpers import template_renderer
db = Database("ext_admin")
admin_ext: Blueprint = Blueprint("admin", __name__, static_folder="static", template_folder="templates")
admin_ext: APIRouter = APIRouter(prefix="/admin", tags=["admin"])
def admin_renderer():
return template_renderer(["lnbits/extensions/admin/templates"])
from .views_api import * # noqa
from .views import * # noqa
from .views_api import * # noqa

50
lnbits/extensions/admin/crud.py
View File

@ -1,11 +1,11 @@
from typing import List, Optional
from lnbits.core.crud import create_payment
from lnbits.helpers import urlsafe_short_hash
from lnbits.settings import *
from . import db
from .models import Admin, Funding
from lnbits.settings import *
from lnbits.helpers import urlsafe_short_hash
from lnbits.core.crud import create_payment
from lnbits.db import Connection
def update_wallet_balance(wallet_id: str, amount: int) -> str:
@ -22,38 +22,30 @@ def update_wallet_balance(wallet_id: str, amount: int) -> str:
)
return "success"
async def update_admin(
) -> Optional[Admin]:
if not CLightningWallet:
print("poo")
await db.execute(
"""
UPDATE admin
SET user = ?, site_title = ?, site_tagline = ?, site_description = ?, allowed_users = ?, default_wallet_name = ?, data_folder = ?, disabled_ext = ?, force_https = ?, service_fee = ?, funding_source = ?
WHERE 1
""",
(
),
)
row = await db.fetchone("SELECT * FROM admin WHERE 1")
return Admin.from_row(row) if row else None
async def update_admin(admin_id: str, **kwargs) -> Optional[Admin]:
async def update_admin(user: str, **kwargs) -> Admin:
q = ", ".join([f"{field[0]} = ?" for field in kwargs.items()])
print("UPDATE", q)
await db.execute(
f"UPDATE jukebox.jukebox SET {q} WHERE id = ?", (*kwargs.values(), juke_id)
f'UPDATE admin SET {q} WHERE "user" = ?', (*kwargs.values(), user)
)
row = await db.fetchone("SELECT * FROM jukebox.jukebox WHERE id = ?", (juke_id,))
return Jukebox(**row) if row else None
row = await db.fetchone('SELECT * FROM admin WHERE "user" = ?', (user,))
assert row, "Newly updated settings couldn't be retrieved"
return Admin(**row) if row else None
# async def update_admin(user: str, **kwargs) -> Optional[Admin]:
# q = ", ".join([f"{field[0]} = ?" for field in kwargs.items()])
# await db.execute(
# f"UPDATE admin SET {q} WHERE user = ?", (*kwargs.values(), user)
# )
# new_settings = await get_admin()
# return new_settings
async def get_admin() -> List[Admin]:
row = await db.fetchone("SELECT * FROM admin WHERE 1")
return Admin.from_row(row) if row else None
row = await db.fetchone("SELECT * FROM admin")
return Admin(**row) if row else None
async def get_funding() -> List[Funding]:
rows = await db.fetchall("SELECT * FROM funding")
return [Funding.from_row(row) for row in rows]
return [Funding(**row) for row in rows]

23
lnbits/extensions/admin/migrations.py
View File

@ -1,5 +1,7 @@
from sqlalchemy.exc import OperationalError # type: ignore
from os import getenv
from sqlalchemy.exc import OperationalError # type: ignore
from lnbits.helpers import urlsafe_short_hash
@ -9,7 +11,7 @@ async def m001_create_admin_table(db):
site_tagline = None
site_description = None
allowed_users = None
admin_user = None
admin_users = None
default_wallet_name = None
data_folder = None
disabled_ext = None
@ -29,8 +31,9 @@ async def m001_create_admin_table(db):
if getenv("LNBITS_ALLOWED_USERS"):
allowed_users = getenv("LNBITS_ALLOWED_USERS")
if getenv("LNBITS_ADMIN_USER"):
admin_user = getenv("LNBITS_ADMIN_USER")
if getenv("LNBITS_ADMIN_USERS"):
admin_users = "".join(getenv("LNBITS_ADMIN_USERS").split())
user = admin_users.split(',')[0]
if getenv("LNBITS_DEFAULT_WALLET_NAME"):
default_wallet_name = getenv("LNBITS_DEFAULT_WALLET_NAME")
@ -53,32 +56,32 @@ async def m001_create_admin_table(db):
await db.execute(
"""
CREATE TABLE IF NOT EXISTS admin (
user TEXT,
"user" TEXT,
site_title TEXT,
site_tagline TEXT,
site_description TEXT,
admin_user TEXT,
admin_users TEXT,
allowed_users TEXT,
default_wallet_name TEXT,
data_folder TEXT,
disabled_ext TEXT,
force_https BOOLEAN,
service_fee INT,
service_fee REAL,
funding_source TEXT
);
"""
)
await db.execute(
"""
INSERT INTO admin (user, site_title, site_tagline, site_description, admin_user, allowed_users, default_wallet_name, data_folder, disabled_ext, force_https, service_fee, funding_source)
INSERT INTO admin ("user", site_title, site_tagline, site_description, admin_users, allowed_users, default_wallet_name, data_folder, disabled_ext, force_https, service_fee, funding_source)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
""",
(
user,
user.strip(),
site_title,
site_tagline,
site_description,
admin_user,
admin_users[1:],
allowed_users,
default_wallet_name,
data_folder,

51
lnbits/extensions/admin/models.py
View File

@ -1,18 +1,35 @@
from typing import NamedTuple
from sqlite3 import Row
from typing import List, Optional
class Admin(NamedTuple):
from fastapi import Query
from pydantic import BaseModel
class UpdateAdminSettings(BaseModel):
site_title: Optional[str]
site_tagline: Optional[str]
site_description: Optional[str]
allowed_users: Optional[str]
admin_users: Optional[str]
default_wallet_name: Optional[str]
data_folder: Optional[str]
disabled_ext: Optional[str]
force_https: Optional[bool]
service_fee: Optional[float]
funding_source: Optional[str]
class Admin(BaseModel):
user: str
site_title: str
site_tagline: str
site_description:str
allowed_users: str
admin_user: str
site_title: Optional[str]
site_tagline: Optional[str]
site_description: Optional[str]
allowed_users: Optional[str]
admin_users: str
default_wallet_name: str
data_folder: str
disabled_ext: str
force_https: str
service_fee: str
force_https: Optional[bool] = Query(True)
service_fee: float
funding_source: str
@classmethod
@ -20,16 +37,16 @@ class Admin(NamedTuple):
data = dict(row)
return cls(**data)
class Funding(NamedTuple):
class Funding(BaseModel):
id: str
backend_wallet: str
endpoint: str
port: str
read_key: str
invoice_key: str
admin_key: str
cert: str
balance: int
endpoint: str = Query(None)
port: str = Query(None)
read_key: str = Query(None)
invoice_key: str = Query(None)
admin_key: str = Query(None)
cert: str = Query(None)
balance: int = Query(None)
selected: int
@classmethod

23
lnbits/extensions/admin/templates/admin/index.html
View File

@ -87,7 +87,7 @@
<q-input
filled
class="q-pr-md"
v-model="data.admin.admin_user"
v-model="data.admin.admin_users"
label="Admin user"
hint=""
></q-input>
@ -442,13 +442,14 @@
site_title: '{{admin.site_title}}',
tagline: '{{admin.site_tagline}}',
description: '{{admin.site_description}}',
admin_user: '{{admin.admin_user}}',
service_fee: parseInt('{{admin.service_fee}}'),
admin_users: '{{admin.admin_users}}',
service_fee: parseFloat('{{admin.service_fee}}'),
default_wallet_name: '{{admin.default_wallet_name}}',
data_folder: '{{admin.data_folder}}',
funding_source_primary: '{{admin.funding_source}}',
disabled_ext: '{{admin.disabled_ext}}'.split(','),
edited: [],
funding: {},
senddata: {}
}
},
@ -528,15 +529,27 @@
},
UpdateLNbits: function () {
var self = this
console.log(self.data.admin)
let {site_title, admin_users, default_wallet_name, data_folder, disabled_ext, service_fee, funding_source_primary} = this.data.admin
let data = {
site_title,
site_tagline: this.data.admin.tagline,
site_description: this.data.admin.description,
admin_users: admin_users.toString(),
default_wallet_name,
data_folder,
disabled_ext: disabled_ext.toString(),
service_fee,
funding_source: funding_source_primary}
console.log(data)
LNbits.api
.request(
'POST',
'/admin/api/v1/admin/',
self.g.user.wallets[0].adminkey,
self.data.admin
data
)
.then(function (response) {
console.log(response.data)
self.$q.notify({
type: 'positive',
message:

39
lnbits/extensions/admin/views.py
View File

@ -1,20 +1,33 @@
from quart import g, render_template, request, jsonify
import json
from email.policy import default
from os import getenv
from lnbits.decorators import check_user_exists, validate_uuids
from fastapi import Request
from fastapi.params import Depends
from fastapi.templating import Jinja2Templates
from starlette.responses import HTMLResponse
from lnbits.core.models import User
from lnbits.decorators import check_user_exists
from lnbits.extensions.admin import admin_ext
from lnbits.core.crud import get_user, create_account
from lnbits.requestvars import g
from . import admin_ext, admin_renderer
from .crud import get_admin, get_funding
from lnbits.settings import WALLET
templates = Jinja2Templates(directory="templates")
@admin_ext.route("/")
@validate_uuids(["usr"], required=True)
@check_user_exists()
async def index():
user_id = g.user
@admin_ext.get("/", response_class=HTMLResponse)
async def index(request: Request, user: User = Depends(check_user_exists)):
admin = await get_admin()
print(g())
funding = [f.dict() for f in await get_funding()]
funding = [{**funding._asdict()} for funding in await get_funding()]
return await render_template("admin/index.html", user=g.user, admin=admin, funding=funding)
print("ADMIN", admin.dict())
return admin_renderer().TemplateResponse(
"admin/index.html", {
"request": request,
"user": user.dict(),
"admin": admin.dict(),
"funding": funding
}
)

61
lnbits/extensions/admin/views_api.py
View File

@ -1,41 +1,42 @@
from quart import jsonify, g, request
from http import HTTPStatus
from .crud import update_wallet_balance
from lnbits.extensions.admin import admin_ext
from lnbits.decorators import api_check_wallet_key, api_validate_post_request
from lnbits.core.crud import get_wallet
from .crud import get_admin,update_admin
import json
@admin_ext.route("/api/v1/admin/<wallet_id>/<topup_amount>", methods=["GET"])
@api_check_wallet_key("admin")
async def api_update_balance(wallet_id, topup_amount):
print(g.data.wallet)
from fastapi import Body, Depends, Request
from starlette.exceptions import HTTPException
from lnbits.core.crud import get_wallet
from lnbits.decorators import WalletTypeInfo, require_admin_key
from lnbits.extensions.admin import admin_ext
from lnbits.extensions.admin.models import Admin, UpdateAdminSettings
from .crud import get_admin, update_admin, update_wallet_balance
@admin_ext.get("/api/v1/admin/{wallet_id}/{topup_amount}", status_code=HTTPStatus.OK)
async def api_update_balance(wallet_id, topup_amount, g: WalletTypeInfo = Depends(require_admin_key)):
print(g.wallet)
try:
wallet = await get_wallet(wallet_id)
except:
return (
jsonify({"error": "Not allowed: not an admin"}),
HTTPStatus.FORBIDDEN,
)
raise HTTPException(
status_code=HTTPStatus.FORBIDDEN, detail="Not allowed: not an admin"
)
print(wallet)
print(topup_amount)
return jsonify({"status": "Success"}), HTTPStatus.OK
return {"status": "Success"}
@admin_ext.route("/api/v1/admin/", methods=["POST"])
@api_check_wallet_key("admin")
@api_validate_post_request(schema={})
async def api_update_admin():
body = await request.get_json()
@admin_ext.post("/api/v1/admin/", status_code=HTTPStatus.OK)
async def api_update_admin(
request: Request,
data: UpdateAdminSettings = Body(...),
g: WalletTypeInfo = Depends(require_admin_key)
):
admin = await get_admin()
print(g.wallet[2])
print(body["admin_user"])
if not admin.admin_user == g.wallet[2] and admin.admin_user != None:
return (
jsonify({"error": "Not allowed: not an admin"}),
HTTPStatus.FORBIDDEN,
)
updated = await update_admin(body)
print(data)
if not admin.user == g.wallet.user:
raise HTTPException(
status_code=HTTPStatus.FORBIDDEN, detail="Not allowed: not an admin"
)
updated = await update_admin(user=g.wallet.user, **data.dict())
print(updated)
return jsonify({"status": "Success"}), HTTPStatus.OK
return {"status": "Success"}