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

fix: mypy errors

Browse Source
This commit is contained in:
Eneko Illarramendi
2020-04-26 16:58:29 +02:00
committed by Sebastian Geisler
parent 976a3d4e5c
commit c3e337a319
18 changed files with 91 additions and 94 deletions
Download Patch File Download Diff File Expand all files Collapse all files
lnbits/wallets/lndgrpc.py
+33 -53
View File
@@ -1,14 +1,13 @@
from os import getenv
import os
import base64
import lnd_grpc # https://github.com/willcl-ark/lnd_grpc
import lnd_grpc # type: ignore
from os import getenv
from .base import InvoiceResponse, PaymentResponse, PaymentStatus, Wallet
class LndWallet(Wallet):
def __init__(self):
endpoint = getenv("LND_GRPC_ENDPOINT")
self.endpoint = endpoint[:-1] if endpoint.endswith("/") else endpoint
self.port = getenv("LND_GRPC_PORT")
@@ -18,85 +17,66 @@ class LndWallet(Wallet):
self.auth_cert = getenv("LND_CERT")
lnd_rpc = lnd_grpc.Client(
lnd_dir = None,
tls_cert_path = self.auth_cert,
network = 'mainnet',
grpc_host = self.endpoint,
grpc_port = self.port
lnd_dir=None, tls_cert_path=self.auth_cert, network="mainnet", grpc_host=self.endpoint, grpc_port=self.port
)
def create_invoice(self, amount: int, memo: str = "") -> InvoiceResponse:
lnd_rpc = lnd_grpc.Client(
lnd_dir = None,
macaroon_path = self.auth_invoice,
tls_cert_path = self.auth_cert,
network = 'mainnet',
grpc_host = self.endpoint,
grpc_port = self.port
lnd_dir=None,
macaroon_path=self.auth_invoice,
tls_cert_path=self.auth_cert,
network="mainnet",
grpc_host=self.endpoint,
grpc_port=self.port,
)
lndResponse = lnd_rpc.add_invoice(
memo = memo,
value = amount,
expiry = 600,
private = True
)
decoded_hash = base64.b64encode(lndResponse.r_hash).decode('utf-8').replace("/","_")
lndResponse = lnd_rpc.add_invoice(memo=memo, value=amount, expiry=600, private=True)
decoded_hash = base64.b64encode(lndResponse.r_hash).decode("utf-8").replace("/", "_")
print(lndResponse.r_hash)
ok, checking_id, payment_request, error_message = True, decoded_hash, str(lndResponse.payment_request), None
ok, checking_id, payment_request, error_message = True, decoded_hash, str(lndResponse.payment_request), None
return InvoiceResponse(ok, checking_id, payment_request, error_message)
def pay_invoice(self, bolt11: str) -> PaymentResponse:
lnd_rpc = lnd_grpc.Client(
lnd_dir = None,
macaroon_path = self.auth_admin,
tls_cert_path = self.auth_cert,
network = 'mainnet',
grpc_host = self.endpoint,
grpc_port = self.port
lnd_dir=None,
macaroon_path=self.auth_admin,
tls_cert_path=self.auth_cert,
network="mainnet",
grpc_host=self.endpoint,
grpc_port=self.port,
)
payinvoice = lnd_rpc.pay_invoice(
payment_request = bolt11,
)
payinvoice = lnd_rpc.pay_invoice(payment_request=bolt11,)
ok, checking_id, fee_msat, error_message = True, None, 0, None
if payinvoice.payment_error:
ok, error_message = False, payinvoice.payment_error
else:
checking_id = base64.b64encode(payinvoice.payment_hash).decode('utf-8').replace("/","_")
checking_id = base64.b64encode(payinvoice.payment_hash).decode("utf-8").replace("/", "_")
return PaymentResponse(ok, checking_id, fee_msat, error_message)
def get_invoice_status(self, checking_id: str) -> PaymentStatus:
check_id = base64.b64decode(checking_id.replace("_","/"))
check_id = base64.b64decode(checking_id.replace("_", "/"))
print(check_id)
lnd_rpc = lnd_grpc.Client(
lnd_dir = None,
macaroon_path = self.auth_invoice,
tls_cert_path = self.auth_cert,
network = 'mainnet',
grpc_host = self.endpoint,
grpc_port = self.port
lnd_dir=None,
macaroon_path=self.auth_invoice,
tls_cert_path=self.auth_cert,
network="mainnet",
grpc_host=self.endpoint,
grpc_port=self.port,
)
for _response in lnd_rpc.subscribe_single_invoice(check_id):
if _response.state == 1:
return PaymentStatus(True)
invoiceThread = threading.Thread(
target=detectPayment,
args=[lndResponse.check_id, ],
daemon=True
)
invoiceThread.start()
return PaymentStatus(None)
def get_payment_status(self, checking_id: str) -> PaymentStatus:
return PaymentStatus(True)