refactor: unify responses in backend wallets

lnbits/wallets/__init__.py

@@ -0,0 +1,4 @@
+# flake8: noqa
+
+from .lnd import LndWallet
+from .lntxbot import LntxbotWallet

lnbits/wallets/base.py

@@ -0,0 +1,32 @@
+from abc import ABC, abstractmethod
+from requests import Response
+from typing import NamedTuple, Optional
+
+
+class InvoiceResponse(NamedTuple):
+    raw_response: Response
+    payment_hash: Optional[str] = None
+    payment_request: Optional[str] = None
+
+
+class TxStatus(NamedTuple):
+    raw_response: Response
+    settled: Optional[bool] = None
+
+
+class Wallet(ABC):
+    @abstractmethod
+    def create_invoice(self, amount: int, memo: str = "") -> InvoiceResponse:
+        pass
+
+    @abstractmethod
+    def pay_invoice(self, bolt11: str) -> Response:
+        pass
+
+    @abstractmethod
+    def get_invoice_status(self, payment_hash: str, wait: bool = True) -> TxStatus:
+        pass
+
+    @abstractmethod
+    def get_payment_status(self, payment_hash: str) -> TxStatus:
+        pass

lnbits/wallets/lnd.py

@@ -0,0 +1,48 @@
+from requests import Response, get, post
+
+from .base import InvoiceResponse, TxStatus, Wallet
+
+
+class LndWallet(Wallet):
+    """https://api.lightning.community/rest/index.html#lnd-rest-api-reference"""
+
+    def __init__(self, *, endpoint: str, admin_macaroon: str):
+        self.endpoint = endpoint[:-1] if endpoint.endswith("/") else endpoint
+        self.auth_admin = {"Grpc-Metadata-macaroon": admin_macaroon}
+
+    def create_invoice(self, amount: int, memo: str = "") -> InvoiceResponse:
+        payment_hash, payment_request = None, None
+        r = post(
+            url=f"{self.endpoint}/v1/invoices",
+            headers=self.auth_admin,
+            json={"value": f"{amount}", "description_hash": memo},  # , "private": True},
+        )
+
+        if r.ok:
+            data = r.json()
+            payment_hash, payment_request = data["r_hash"], data["payment_request"]
+
+        return InvoiceResponse(r, payment_hash, payment_request)
+
+    def pay_invoice(self, bolt11: str) -> Response:
+        raise NotImplementedError
+
+    def get_invoice_status(self, payment_hash: str, wait: bool = True) -> TxStatus:
+        r = get(url=f"{self.endpoint}/v1/invoice", headers=self.auth_admin, params={"r_hash": payment_hash})
+
+        if not r.ok:
+            return TxStatus(r, None)
+
+        return TxStatus(r, r.json()["settled"])
+
+    def get_payment_status(self, payment_hash: str) -> TxStatus:
+        r = get(url=f"{self.endpoint}/v1/payments", headers=self.auth_admin, params={"include_incomplete": True})
+
+        if not r.ok:
+            return TxStatus(r, None)
+
+        payments = [p for p in r.json()["payments"] if p["payment_hash"] == payment_hash]
+        payment = payments[0] if payments else None
+
+        # check payment.status: https://api.lightning.community/rest/index.html?python#peersynctype
+        return TxStatus(r, {0: None, 1: None, 2: True, 3: False}[payment["status"]] if payment else None)

lnbits/wallets/lntxbot.py

@@ -0,0 +1,47 @@
+from requests import Response, post
+
+from .base import InvoiceResponse, TxStatus, Wallet
+
+
+class LntxbotWallet(Wallet):
+    """https://github.com/fiatjaf/lntxbot/blob/master/api.go"""
+
+    def __init__(self, *, endpoint: str, admin_key: str, invoice_key: str):
+        self.endpoint = endpoint[:-1] if endpoint.endswith("/") else endpoint
+        self.auth_admin = {"Authorization": f"Basic {admin_key}"}
+        self.auth_invoice = {"Authorization": f"Basic {invoice_key}"}
+
+    def create_invoice(self, amount: int, memo: str = "") -> InvoiceResponse:
+        payment_hash, payment_request = None, None
+        r = post(url=f"{self.endpoint}/addinvoice", headers=self.auth_invoice, json={"amt": str(amount), "memo": memo})
+
+        if r.ok:
+            data = r.json()
+            payment_hash, payment_request = data["payment_hash"], data["pay_req"]
+
+        return InvoiceResponse(r, payment_hash, payment_request)
+
+    def pay_invoice(self, bolt11: str) -> Response:
+        return post(url=f"{self.endpoint}/payinvoice", headers=self.auth_admin, json={"invoice": bolt11})
+
+    def get_invoice_status(self, payment_hash: str, wait: bool = True) -> TxStatus:
+        wait = "true" if wait else "false"
+        r = post(url=f"{self.endpoint}/invoicestatus/{payment_hash}?wait={wait}", headers=self.auth_invoice)
+        data = r.json()
+
+        if not r.ok or "error" in data:
+            return TxStatus(r, None)
+
+        if "preimage" not in data or not data["preimage"]:
+            return TxStatus(r, False)
+
+        return TxStatus(r, True)
+
+    def get_payment_status(self, payment_hash: str) -> TxStatus:
+        r = post(url=f"{self.endpoint}/paymentstatus/{payment_hash}", headers=self.auth_invoice)
+        data = r.json()
+
+        if not r.ok or "error" in data:
+            return TxStatus(r, None)
+
+        return TxStatus(r, {"complete": True, "failed": False, "unknown": None}[data.get("status", "unknown")])