Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def poll_updates(invoice_id):
obj = models.Invoice.objects.get(id=invoice_id)
address = obj.bitcoin_address
if not address:
return
btc_instance = BTC(RPC_URL, xpub=obj.products.all()[0].store.wallet.xpub,
rpc_user=RPC_USER, rpc_pass=RPC_PASS)
while True:
invoice_data = btc_instance.getrequest(address)
if invoice_data["status"] != "Pending":
if invoice_data["status"] == "Unknown":
obj.status = "invalid"
if invoice_data["status"] == "Expired":
obj.status = "expired"
if invoice_data["status"] == "Paid":
obj.status = "complete"
obj.save()
async_to_sync(channel_layer.group_send)(
invoice_id, {"type": "notify", "status": obj.status})
return
time.sleep(1)
def get_queryset(self):
queryset = models.Invoice.objects.all().order_by("-date")
status = self.request.query_params.get('status', None)
if status is not None:
queryset = queryset.filter(status=status)
currency = self.request.query_params.get('currency', 'BTC')
if currency == "USD":
exchange_rate = decimal.Decimal(
BTC(RPC_URL, rpc_user=RPC_USER, rpc_pass=RPC_PASS).server.exchange_rate())
for i in queryset:
i.amount = (decimal.Decimal(i.amount) *
exchange_rate).quantize(PRECISION)
return queryset
def get_wallet_history(model, response):
txes = BTC(RPC_URL, xpub=model.xpub, rpc_user=RPC_USER,
rpc_pass=RPC_PASS).server.history()["transactions"]
for i in txes:
response.append({
"date": i["date"],
"txid": truncate(i["txid"], 20),
"amount": i["value"]
})
def sync_wallet(wallet_id, xpub):
model = models.Wallet.objects.get(id=wallet_id)
try:
balance = BTC(RPC_URL, xpub=xpub, rpc_user=RPC_USER,
rpc_pass=RPC_PASS).balance()
model.balance = balance["confirmed"] or 0
model.updated_time = timezone.now()
model.save()
time.sleep(0.5)
async_to_sync(channel_layer.group_send)(wallet_id, {
"type": "notify", "status": "success", "balance": balance["confirmed"] or 0})
except Exception:
model.delete()
async_to_sync(channel_layer.group_send)(
wallet_id, {"type": "notify", "status": "error", "balance": 0})
def create(self, validated_data):
data_got = BTC(RPC_URL, xpub=validated_data["products"][0].store.wallet.xpub, rpc_user=RPC_USER, rpc_pass=RPC_PASS).addrequest(
validated_data["amount"], description=validated_data["products"][0].description)
validated_data["bitcoin_address"] = data_got["address"]
validated_data["bitcoin_url"] = data_got["URI"]
validated_data = edit_image(validated_data)
return super().create(validated_data)
def get_balance(self, obj):
return BTC(RPC_URL, xpub=obj.xpub, rpc_user=RPC_USER,
rpc_pass=RPC_PASS).balance()['confirmed']
def get(self, request, format=None):
btc_amount = decimal.Decimal(request.query_params.get("btc", 1))
usd_price = decimal.Decimal(
BTC(RPC_URL, rpc_user=RPC_USER, rpc_pass=RPC_PASS).server.exchange_rate())
usd_price = (
btc_amount * usd_price).quantize(PRECISION)
return Response(usd_price)
return {"is_active": False}
except Exception:
return {"is_active": False}
if request.user.is_authenticated:
products = models.Product.objects.filter(
store__wallet__user=request.user)
products = products.order_by("-date")
products_count = len(products)
stores_count = models.Store.objects.filter(
wallet__user=request.user).count()
wallets = models.Wallet.objects.filter(user=request.user)
wallets_count = len(wallets)
wallets_balance = decimal.Decimal(0)
for i in wallets:
if timezone.now() - i.updated_date >= timezone.timedelta(hours=2):
wallets_balance += decimal.Decimal(BTC(
RPC_URL, xpub=i.xpub, rpc_user=RPC_USER, rpc_pass=RPC_PASS).balance()['confirmed'])
i.updated_date = timezone.now()
i.save()
else:
wallets_balance += decimal.Decimal(i.balance)
wallets_balance = format(
wallets_balance, ".08f").rstrip("0").rstrip(".")
return {"is_active": True, "products": products, "stores_count": stores_count, "wallets_count": wallets_count,
"products_count": products_count, "wallets_balance": wallets_balance}
else:
return {"is_active": False}