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

refactor: a wallet is a wallet is a wallet

Browse Source
This commit is contained in:
Eneko Illarramendi
2020-03-31 19:05:25 +02:00
parent 75d97ddfc1
commit d03785558b
15 changed files with 207 additions and 167 deletions
Download Patch File Download Diff File Expand all files Collapse all files
lnbits/wallets/base.py
+7 -8
View File
@@ -1,23 +1,22 @@
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
ok: bool
checking_id: Optional[str] = None # payment_hash, rpc_id
payment_request: Optional[str] = None
error_message: Optional[str] = None
class PaymentResponse(NamedTuple):
raw_response: Response
failed: bool = False
ok: bool
checking_id: Optional[str] = None # payment_hash, rcp_id
fee_msat: int = 0
error_message: Optional[str] = None
class PaymentStatus(NamedTuple):
raw_response: Response
paid: Optional[bool] = None
@property
@@ -35,9 +34,9 @@ class Wallet(ABC):
pass
@abstractmethod
def get_invoice_status(self, payment_hash: str) -> PaymentStatus:
def get_invoice_status(self, checking_id: str) -> PaymentStatus:
pass
@abstractmethod
def get_payment_status(self, payment_hash: str) -> PaymentStatus:
def get_payment_status(self, checking_id: str) -> PaymentStatus:
pass
lnbits/wallets/lnd.py
+29 -19
View File
@@ -1,36 +1,38 @@
from os import getenv
from requests import get, post
from .base import InvoiceResponse, PaymentResponse, PaymentStatus, Wallet
class LndWallet(Wallet):
"""https://api.lightning.community/rest/index.html#lnd-rest-api-reference"""
def __init__(self, *, endpoint: str, admin_macaroon: str, invoice_macaroon: str, read_macaroon: str):
def __init__(self):
endpoint = getenv("LND_API_ENDPOINT")
self.endpoint = endpoint[:-1] if endpoint.endswith("/") else endpoint
self.auth_admin = {"Grpc-Metadata-macaroon": admin_macaroon}
self.auth_invoice = {"Grpc-Metadata-macaroon": invoice_macaroon}
self.auth_read = {"Grpc-Metadata-macaroon": read_macaroon}
self.auth_admin = {"Grpc-Metadata-macaroon": getenv("LND_ADMIN_MACAROON")}
self.auth_invoice = {"Grpc-Metadata-macaroon": getenv("LND_INVOICE_MACAROON")}
self.auth_read = {"Grpc-Metadata-macaroon": getenv("LND_READ_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,
verify=False,
json={"value": amount, "memo": memo, "private": True},
)
ok, checking_id, payment_request, error_message = r.ok, None, None, None
if r.ok:
data = r.json()
payment_request = data["payment_request"]
rr = get(url=f"{self.endpoint}/v1/payreq/{payment_request}", headers=self.auth_read, verify=False,)
rr = get(url=f"{self.endpoint}/v1/payreq/{payment_request}", headers=self.auth_read, verify=False)
if rr.ok:
dataa = rr.json()
payment_hash = dataa["payment_hash"]
if rr.ok:
checking_id = rr.json()["payment_hash"]
return InvoiceResponse(r, payment_hash, payment_request)
return InvoiceResponse(ok, checking_id, payment_request, error_message)
def pay_invoice(self, bolt11: str) -> PaymentResponse:
r = post(
@@ -39,17 +41,25 @@ class LndWallet(Wallet):
verify=False,
json={"payment_request": bolt11},
)
return PaymentResponse(r, not r.ok)
ok, checking_id, fee_msat, error_message = r.ok, None, 0, None
data = r.json()["data"]
def get_invoice_status(self, payment_hash: str) -> PaymentStatus:
r = get(url=f"{self.endpoint}/v1/invoice/{payment_hash}", headers=self.auth_read, verify=False)
if "payment_error" in data and data["payment_error"]:
ok, error_message = False, data["payment_error"]
else:
checking_id = data["payment_hash"]
return PaymentResponse(ok, checking_id, fee_msat, error_message)
def get_invoice_status(self, checking_id: str) -> PaymentStatus:
r = get(url=f"{self.endpoint}/v1/invoice/{checking_id}", headers=self.auth_read, verify=False)
if not r.ok or "settled" not in r.json():
return PaymentStatus(r, None)
return PaymentStatus(None)
return PaymentStatus(r, r.json()["settled"])
return PaymentStatus(r.json()["settled"])
def get_payment_status(self, payment_hash: str) -> PaymentStatus:
def get_payment_status(self, checking_id: str) -> PaymentStatus:
r = get(
url=f"{self.endpoint}/v1/payments",
headers=self.auth_admin,
@@ -58,11 +68,11 @@ class LndWallet(Wallet):
)
if not r.ok:
return PaymentStatus(r, None)
return PaymentStatus(None)
payments = [p for p in r.json()["payments"] if p["payment_hash"] == payment_hash]
payments = [p for p in r.json()["payments"] if p["payment_hash"] == checking_id]
payment = payments[0] if payments else None
# check payment.status: https://api.lightning.community/rest/index.html?python#peersynctype
statuses = {"UNKNOWN": None, "IN_FLIGHT": None, "SUCCEEDED": True, "FAILED": False}
return PaymentStatus(r, statuses[payment["status"]] if payment else None)
return PaymentStatus(statuses[payment["status"]] if payment else None)
lnbits/wallets/lnpay.py
+22 -17
View File
@@ -1,3 +1,4 @@
from os import getenv
from requests import get, post
from .base import InvoiceResponse, PaymentResponse, PaymentStatus, Wallet
@@ -6,27 +7,27 @@ from .base import InvoiceResponse, PaymentResponse, PaymentStatus, Wallet
class LNPayWallet(Wallet):
"""https://docs.lnpay.co/"""
def __init__(self, *, endpoint: str, admin_key: str, invoice_key: str, api_key: str, read_key: str):
def __init__(self):
endpoint = getenv("LNPAY_API_ENDPOINT")
self.endpoint = endpoint[:-1] if endpoint.endswith("/") else endpoint
self.auth_admin = admin_key
self.auth_invoice = invoice_key
self.auth_read = read_key
self.auth_api = {"X-Api-Key": api_key}
self.auth_admin = getenv("LNPAY_ADMIN_KEY")
self.auth_invoice = getenv("LNPAY_INVOICE_KEY")
self.auth_read = getenv("LNPAY_READ_KEY")
self.auth_api = {"X-Api-Key": getenv("LNPAY_API_KEY")}
def create_invoice(self, amount: int, memo: str = "") -> InvoiceResponse:
payment_hash, payment_request = None, None
r = post(
url=f"{self.endpoint}/user/wallet/{self.auth_invoice}/invoice",
headers=self.auth_api,
json={"num_satoshis": f"{amount}", "memo": memo},
)
ok, checking_id, payment_request, error_message = r.status_code == 201, None, None, None
if r.ok:
if ok:
data = r.json()
payment_hash, payment_request = data["id"], data["payment_request"]
checking_id, payment_request = data["id"], data["payment_request"]
return InvoiceResponse(r, payment_hash, payment_request)
return InvoiceResponse(ok, checking_id, payment_request, error_message)
def pay_invoice(self, bolt11: str) -> PaymentResponse:
r = post(
@@ -34,17 +35,21 @@ class LNPayWallet(Wallet):
headers=self.auth_api,
json={"payment_request": bolt11},
)
ok, checking_id, fee_msat, error_message = r.status_code == 201, None, 0, None
return PaymentResponse(r, not r.ok)
if ok:
checking_id = r.json()["lnTx"]["id"]
def get_invoice_status(self, payment_hash: str) -> PaymentStatus:
return self.get_payment_status(payment_hash)
return PaymentResponse(ok, checking_id, fee_msat, error_message)
def get_payment_status(self, payment_hash: str) -> PaymentStatus:
r = get(url=f"{self.endpoint}/user/lntx/{payment_hash}", headers=self.auth_api)
def get_invoice_status(self, checking_id: str) -> PaymentStatus:
return self.get_payment_status(checking_id)
def get_payment_status(self, checking_id: str) -> PaymentStatus:
r = get(url=f"{self.endpoint}/user/lntx/{checking_id}", headers=self.auth_api)
if not r.ok:
return PaymentStatus(r, None)
return PaymentStatus(None)
statuses = {0: None, 1: True, -1: False}
return PaymentStatus(r, statuses[r.json()["settled"]])
return PaymentStatus(statuses[r.json()["settled"]])
lnbits/wallets/lntxbot.py
+29 -26
View File
@@ -1,3 +1,4 @@
from os import getenv
from requests import post
from .base import InvoiceResponse, PaymentResponse, PaymentStatus, Wallet
@@ -6,56 +7,58 @@ from .base import InvoiceResponse, PaymentResponse, PaymentStatus, Wallet
class LntxbotWallet(Wallet):
"""https://github.com/fiatjaf/lntxbot/blob/master/api.go"""
def __init__(self, *, endpoint: str, admin_key: str, invoice_key: str):
def __init__(self):
endpoint = getenv("LNTXBOT_API_ENDPOINT")
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}"}
self.auth_admin = {"Authorization": f"Basic {getenv('LNTXBOT_ADMIN_KEY')}"}
self.auth_invoice = {"Authorization": f"Basic {getenv('LNTXBOT_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})
ok, checking_id, payment_request, error_message = r.ok, None, None, None
if r.ok:
data = r.json()
payment_hash, payment_request = data["payment_hash"], data["pay_req"]
checking_id, payment_request = data["payment_hash"], data["pay_req"]
return InvoiceResponse(r, payment_hash, payment_request)
if "error" in data and data["error"]:
ok = False
error_message = data["message"]
return InvoiceResponse(ok, checking_id, payment_request, error_message)
def pay_invoice(self, bolt11: str) -> PaymentResponse:
r = post(url=f"{self.endpoint}/payinvoice", headers=self.auth_admin, json={"invoice": bolt11})
failed, fee_msat, error_message = not r.ok, 0, None
ok, checking_id, fee_msat, error_message = r.ok, None, 0, None
if r.ok:
data = r.json()
if "error" in data and data["error"]:
failed = True
error_message = data["message"]
elif "fee_msat" in data:
fee_msat = data["fee_msat"]
return PaymentResponse(r, failed, fee_msat, error_message)
if "payment_hash" in data:
checking_id, fee_msat = data["decoded"]["payment_hash"], data["fee_msat"]
elif "error" in data and data["error"]:
ok, error_message = False, data["message"]
def get_invoice_status(self, payment_hash: str) -> PaymentStatus:
r = post(url=f"{self.endpoint}/invoicestatus/{payment_hash}?wait=false", headers=self.auth_invoice)
return PaymentResponse(ok, checking_id, fee_msat, error_message)
if not r.ok:
return PaymentStatus(r, None)
def get_invoice_status(self, checking_id: str) -> PaymentStatus:
r = post(url=f"{self.endpoint}/invoicestatus/{checking_id}?wait=false", headers=self.auth_invoice)
if not r.ok or "error" in r.json():
return PaymentStatus(None)
data = r.json()
if "error" in data:
return PaymentStatus(r, None)
if "preimage" not in data or not data["preimage"]:
return PaymentStatus(r, False)
return PaymentStatus(False)
return PaymentStatus(r, True)
return PaymentStatus(True)
def get_payment_status(self, payment_hash: str) -> PaymentStatus:
r = post(url=f"{self.endpoint}/paymentstatus/{payment_hash}", headers=self.auth_invoice)
def get_payment_status(self, checking_id: str) -> PaymentStatus:
r = post(url=f"{self.endpoint}/paymentstatus/{checking_id}", headers=self.auth_invoice)
if not r.ok or "error" in r.json():
return PaymentStatus(r, None)
return PaymentStatus(None)
statuses = {"complete": True, "failed": False, "pending": None, "unknown": None}
return PaymentStatus(r, statuses[r.json().get("status", "unknown")])
return PaymentStatus(statuses[r.json().get("status", "unknown")])
lnbits/wallets/opennode.py
+31 -18
View File
@@ -1,47 +1,60 @@
from os import getenv
from requests import get, post
from .base import InvoiceResponse, PaymentResponse, PaymentStatus, Wallet
class OpenNodeWallet(Wallet):
"""https://api.lightning.community/rest/index.html#lnd-rest-api-reference"""
"""https://developers.opennode.com/"""
def __init__(self, *, endpoint: str, admin_key: str, invoice_key: str):
def __init__(self):
endpoint = getenv("OPENNODE_API_ENDPOINT")
self.endpoint = endpoint[:-1] if endpoint.endswith("/") else endpoint
self.auth_admin = {"Authorization": admin_key}
self.auth_invoice = {"Authorization": invoice_key}
self.auth_admin = {"Authorization": getenv("OPENNODE_ADMIN_KEY")}
self.auth_invoice = {"Authorization": getenv("OPENNODE_INVOICE_KEY")}
def create_invoice(self, amount: int, memo: str = "") -> InvoiceResponse:
payment_hash, payment_request = None, None
r = post(
url=f"{self.endpoint}/v1/charges",
headers=self.auth_invoice,
json={"amount": f"{amount}", "description": memo}, # , "private": True},
)
if r.ok:
data = r.json()
payment_hash, payment_request = data["data"]["id"], data["data"]["lightning_invoice"]["payreq"]
ok, checking_id, payment_request, error_message = r.ok, None, None, None
return InvoiceResponse(r, payment_hash, payment_request)
if r.ok:
data = r.json()["data"]
checking_id, payment_request = data["id"], data["lightning_invoice"]["payreq"]
else:
error_message = r.json()["message"]
return InvoiceResponse(ok, checking_id, payment_request, error_message)
def pay_invoice(self, bolt11: str) -> PaymentResponse:
r = post(url=f"{self.endpoint}/v2/withdrawals", headers=self.auth_admin, json={"type": "ln", "address": bolt11})
return PaymentResponse(r, not r.ok)
ok, checking_id, fee_msat, error_message = r.ok, None, 0, None
def get_invoice_status(self, payment_hash: str) -> PaymentStatus:
r = get(url=f"{self.endpoint}/v1/charge/{payment_hash}", headers=self.auth_invoice)
if r.ok:
data = r.json()["data"]
checking_id, fee_msat = data["id"], data["fee"] * 1000
else:
error_message = r.json()["message"]
return PaymentResponse(ok, checking_id, fee_msat, error_message)
def get_invoice_status(self, checking_id: str) -> PaymentStatus:
r = get(url=f"{self.endpoint}/v1/charge/{checking_id}", headers=self.auth_invoice)
if not r.ok:
return PaymentStatus(r, None)
return PaymentStatus(None)
statuses = {"processing": None, "paid": True, "unpaid": False}
return PaymentStatus(r, statuses[r.json()["data"]["status"]])
return PaymentStatus(statuses[r.json()["data"]["status"]])
def get_payment_status(self, payment_hash: str) -> PaymentStatus:
r = get(url=f"{self.endpoint}/v1/withdrawal/{payment_hash}", headers=self.auth_admin)
def get_payment_status(self, checking_id: str) -> PaymentStatus:
r = get(url=f"{self.endpoint}/v1/withdrawal/{checking_id}", headers=self.auth_admin)
if not r.ok:
return PaymentStatus(r, None)
return PaymentStatus(None)
statuses = {"pending": None, "confirmed": True, "error": False, "failed": False}
return PaymentStatus(r, statuses[r.json()["data"]["status"]])
return PaymentStatus(statuses[r.json()["data"]["status"]])