watchonly done

lnbits/extensions/watchonly/views_api.py

@@ -1,11 +1,15 @@
 import hashlib
-from quart import g, jsonify, url_for, request
 from http import HTTPStatus
 import httpx
 import json
 
 from lnbits.core.crud import get_user
-from lnbits.decorators import api_check_wallet_key, api_validate_post_request
+from lnbits.decorators import WalletTypeInfo, get_key_type
+
+from fastapi import Query
+from fastapi.params import Depends
+from starlette.exceptions import HTTPException
+from .models import CreateWallet
 
 from lnbits.extensions.watchonly import watchonly_ext
 from .crud import (
@@ -24,86 +28,83 @@ from .crud import (
 ###################WALLETS#############################
 
 
-@watchonly_ext.route("/api/v1/wallet", methods=["GET"])
-@api_check_wallet_key("invoice")
-async def api_wallets_retrieve():
+@watchonly_ext.get("/api/v1/wallet")
+async def api_wallets_retrieve(wallet: WalletTypeInfo = Depends(get_key_type)):
 
     try:
-        return (
-            jsonify(
-                [wallet._asdict() for wallet in await get_watch_wallets(g.wallet.user)]
-            ),
-            HTTPStatus.OK,
-        )
+        return [wallet.dict() for wallet in await get_watch_wallets(wallet.wallet.user)]
     except:
         return ""
 
 
-@watchonly_ext.route("/api/v1/wallet/<wallet_id>", methods=["GET"])
-@api_check_wallet_key("invoice")
-async def api_wallet_retrieve(wallet_id):
-    wallet = await get_watch_wallet(wallet_id)
+@watchonly_ext.get("/api/v1/wallet/{wallet_id}")
+async def api_wallet_retrieve(wallet_id, wallet: WalletTypeInfo = Depends(get_key_type)):
+    w_wallet = await get_watch_wallet(wallet_id)
 
-    if not wallet:
-        return jsonify({"message": "wallet does not exist"}), HTTPStatus.NOT_FOUND
+    if not w_wallet:
+        raise HTTPException(
+            status_code=HTTPStatus.NOT_FOUND,
+            detail="Wallet does not exist."
+        )
 
-    return jsonify(wallet._asdict()), HTTPStatus.OK
+    return w_wallet.dict()
 
 
-@watchonly_ext.route("/api/v1/wallet", methods=["POST"])
-@api_check_wallet_key("admin")
-@api_validate_post_request(
-    schema={
-        "masterpub": {"type": "string", "empty": False, "required": True},
-        "title": {"type": "string", "empty": False, "required": True},
-    }
-)
-async def api_wallet_create_or_update(wallet_id=None):
+@watchonly_ext.post("/api/v1/wallet")
+async def api_wallet_create_or_update(data: CreateWallet, wallet_id=None, w: WalletTypeInfo = Depends(get_key_type)):
     try:
         wallet = await create_watch_wallet(
-            user=g.wallet.user, masterpub=g.data["masterpub"], title=g.data["title"]
+            user=w.wallet.user, masterpub=data.masterpub, title=data.title
         )
     except Exception as e:
-        return jsonify({"message": str(e)}), HTTPStatus.BAD_REQUEST
-    mempool = await get_mempool(g.wallet.user)
+        raise HTTPException(
+            status_code=HTTPStatus.BAD_REQUEST,
+            detail=str(e)
+        )
+
+    mempool = await get_mempool(w.wallet.user)
     if not mempool:
-        create_mempool(user=g.wallet.user)
-    return jsonify(wallet._asdict()), HTTPStatus.CREATED
+        create_mempool(user=w.wallet.user)
+    return wallet.dict()
 
 
-@watchonly_ext.route("/api/v1/wallet/<wallet_id>", methods=["DELETE"])
-@api_check_wallet_key("admin")
-async def api_wallet_delete(wallet_id):
+@watchonly_ext.delete("/api/v1/wallet/{wallet_id}")
+async def api_wallet_delete(wallet_id, w: WalletTypeInfo = Depends(get_key_type)):
     wallet = await get_watch_wallet(wallet_id)
 
     if not wallet:
-        return jsonify({"message": "Wallet link does not exist."}), HTTPStatus.NOT_FOUND
+        raise HTTPException(
+            status_code=HTTPStatus.NOT_FOUND,
+            detail="Wallet does not exist."
+        )
 
     await delete_watch_wallet(wallet_id)
 
-    return jsonify({"deleted": "true"}), HTTPStatus.NO_CONTENT
+    raise HTTPException(status_code=HTTPStatus.NO_CONTENT)
 
 
 #############################ADDRESSES##########################
 
 
-@watchonly_ext.route("/api/v1/address/<wallet_id>", methods=["GET"])
-@api_check_wallet_key("invoice")
-async def api_fresh_address(wallet_id):
+@watchonly_ext.get("/api/v1/address/{wallet_id}")
+async def api_fresh_address(wallet_id, w: WalletTypeInfo = Depends(get_key_type)):
     await get_fresh_address(wallet_id)
 
     addresses = await get_addresses(wallet_id)
 
-    return jsonify([address._asdict() for address in addresses]), HTTPStatus.OK
+    return [address.dict() for address in addresses]
 
 
-@watchonly_ext.route("/api/v1/addresses/<wallet_id>", methods=["GET"])
-@api_check_wallet_key("invoice")
-async def api_get_addresses(wallet_id):
+@watchonly_ext.get("/api/v1/addresses/{wallet_id}")
+
+async def api_get_addresses(wallet_id, w: WalletTypeInfo = Depends(get_key_type)):
     wallet = await get_watch_wallet(wallet_id)
 
     if not wallet:
-        return jsonify({"message": "wallet does not exist"}), HTTPStatus.NOT_FOUND
+        raise HTTPException(
+            status_code=HTTPStatus.NOT_FOUND,
+            detail="Wallet does not exist."
+        )
 
     addresses = await get_addresses(wallet_id)
 
@@ -111,28 +112,21 @@ async def api_get_addresses(wallet_id):
         await get_fresh_address(wallet_id)
         addresses = await get_addresses(wallet_id)
 
-    return jsonify([address._asdict() for address in addresses]), HTTPStatus.OK
+    return [address.dict() for address in addresses]
 
 
 #############################MEMPOOL##########################
 
 
-@watchonly_ext.route("/api/v1/mempool", methods=["PUT"])
-@api_check_wallet_key("admin")
-@api_validate_post_request(
-    schema={
-        "endpoint": {"type": "string", "empty": False, "required": True},
-    }
-)
-async def api_update_mempool():
-    mempool = await update_mempool(user=g.wallet.user, **g.data)
-    return jsonify(mempool._asdict()), HTTPStatus.OK
+@watchonly_ext.put("/api/v1/mempool")
+async def api_update_mempool(endpoint: str = Query(...), w: WalletTypeInfo = Depends(get_key_type)):
+    mempool = await update_mempool(endpoint, user=w.wallet.user)
+    return mempool.dict()
 
 
-@watchonly_ext.route("/api/v1/mempool", methods=["GET"])
-@api_check_wallet_key("admin")
-async def api_get_mempool():
-    mempool = await get_mempool(g.wallet.user)
+@watchonly_ext.get("/api/v1/mempool")
+async def api_get_mempool(w: WalletTypeInfo = Depends(get_key_type)):
+    mempool = await get_mempool(w.wallet.user)
     if not mempool:
-        mempool = await create_mempool(user=g.wallet.user)
-    return jsonify(mempool._asdict()), HTTPStatus.OK
+        mempool = await create_mempool(user=w.wallet.user)
+    return mempool.dict()